Python使用watchfiles实现监控目录变更

from watchfiles import watch
# 当前目录为 /Users/satori/Desktop/project
for change in watch("."):
    print(change)

import sys
import asyncio
from asyncio import StreamReader
from watchfiles import awatch, Change
# 监视指定目录
async def watch_files(path):
    # awatch(...) 返回的是异步生成器，需要通过 async for 遍历
    async for change in awatch(path):
        print("-" * 20)
        # change 是一个集合，里面可能会涉及到多个文件的变更
        for item in change:
            if item[0] == Change.added:
                operation = "你增加了"
            elif item[0] == Change.modified:
                operation = "你修改了"
            else:
                operation = "你删除了"
            print(f"{operation} `{item[1]}`")
        print("\n")
# 读取命令行输入，但是注意：不可以使用 input 函数，因为它是同步阻塞调用
# 这种调用在协程当中是大忌，会阻塞整个线程，我们需要改造成异步模式
async def read_from_stdin():
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    loop = asyncio.get_running_loop()
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    return reader
# read_from_stdin 函数的具体细节暂时不用太关注
# 只需要知道它能异步读取命令行即可，关于这方面的内容后续会介绍
# 然后定义主协程
async def main():
    # 监视当前目录
    asyncio.create_task(watch_files("."))
    # 创建读取器
    stdin_reader = await read_from_stdin()
    while True:
        # 从命令行读取输入
        command = await stdin_reader.readline()
        # 执行命令
        procs = await asyncio.create_subprocess_shell(command)
        await procs.wait()
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

import asyncio
from watchfiles import awatch
async def watch_files(*paths, stop_event):
    async for _ in awatch(*paths, stop_event=stop_event):
        pass
    print("停止监视")
async def main():
    event = asyncio.Event()
    # 传递一个事件，准确的说，只要有 is_set 方法，任何对象都行
    asyncio.create_task(watch_files(".", stop_event=event))
    # 当 event.is_set() 为 True 的时候，停止监视
    print("is_set: ", event.is_set())
    await asyncio.sleep(3)  # sleep 3
    event.set()
    print("三秒后, is_set: ", event.is_set())
    # 等待子协程打印完毕
    await asyncio.sleep(0.1)
asyncio.run(main())
"""
is_set:  False
三秒后, is_set:  True
停止监视
"""