Python3文件对象操作

1.实现os.walk方法(广度优先) import osdef walk(root): stack = [root] files = [] while stack:

1.实现os.walk方法(广度优先)

import osdef walk(root):    stack = [root]    files = []    while stack:        cur = stack.pop()        for x in os.scandir(cur):            if x.is_dir():                stack.append(x.path)            else:                files.append(x.path)    return files

2.实现一个优先队列，由用户指定比较函数

from collections import namedtupleProporityQueue = namedtuple('ProporityQueue', ['add', 'pop'])def priority_queue_factory(cmp=None):    def default_cmp(a, b):        if a > b:            return 1        if a == b:            return 0        return -1    if cmp is None:        cmp = default_cmp        data = []    def add(e):        idx = len(data)        data.append(e)        parent_idx = (idx - 1) // 2        while parent_idx >= 0:            #if data[idx] > data[parent_idx]:            if cmp(data[idx], data[parent_idx]) > 0:                data[parent_idx], data[idx] = data[idx], data[parent_idx]                idx = parent_idx                parent_idx = (idx - 1) // 2            else:                break        def pop():        if not data:            return None        if len(data) == 1:            return data.pop()        idx = 0        ret = data[idx]        data[idx] = data.pop()        left_idx = 2 * idx + 1        rigth_idx = left_idx + 1        while left_idx < len(data):            child_idx = left_idx            #if rigth_idx < len(data) and data[rigth_idx] > data[left_idx]: # 存在右子节点 并且 右子节点大于左子节点            if rigth_idx < len(data) and cmp(data[rigth_idx], data[left_idx]) > 0:                child_idx = rigth_idx            #if data[idx] < data[child_idx]:            if cmp(data[idx], data[child_idx]) < 0:                data[idx], data[child_idx] = data[child_idx], data[idx]                idx = child_idx                left_idx = 2 * idx + 1                rigth_idx = left_idx + 1            else:                break        return ret        return ProporityQueue(add, pop)

3.实现tail -f的功能

import timeimport threadingdef tailf(path):    offset = 0    event = threading.Event()    try:        while not event.is_set():            with open(path) as f:                if offset > os.stat(path).st_size:                #if offset > len(f)                    offset = 0                f.seek(offset)    #             for line in f:    #                 yield line                yield from f                offset = f.tell()            event.wait(0.1)    except KeyboardInterrupt:        event.set()        for x in tailf('utf8.txt'):    print(x,)