在工作中难免会碰到这样的需求，监控指定目录，如果该目录下发生文件变更，那么进行一系列的处理。而如何监视一个目录，就是我们本次探讨的主题。
监视目录我们可以使用 watchfiles 模块，该模块不仅简单，而且性能也不错。主要原因是，和底层文件系统交互的代码是基于 Rust 编写的，所以性能是有保证的。
通过 pip install watchfiles 安装之后，我们来看看它的用法。
from watchfiles import watch # 当前目录为 /Users/satori/Desktop/project for change in watch("."): print(change)
我们执行此程序，会处于阻塞状态，并持续监听指定目录的变化。然后我们在当前目录创建几个文件，看看效果。
创建一个 data.txt 文本文件，程序输出如下：
{(<Change.added: 1>, '/Users/satori/Desktop/project/data.txt')}
返回的是一个集合，目前只涉及一个文件的变更，所以集合里面只有一个元素。而集合里面存储的都是元组，元组的第一个元素表示操作类型，总共有三种：分别是增加、修改和删除。
元组的第二个参数就是具体的文件路径，因此程序的输出就告诉我们，当前目录新增了一个 data.txt。
再创建一个 txt_files 目录，程序输出如下：
{(<Change.added: 1>, '/Users/satori/Desktop/project/txt_files')}
不管是目录文件还是文本文件，都属于文件，所以输出是一样的。如果想知道新增的到底是目录还是普通文件，那么还需要通过 os 模块检测一下。
我们在 txt_files 目录中创建一个 data.txt，程序输出如下：
{(<Change.added: 1>, '/Users/satori/Desktop/project/txt_files/data.txt')}
所以 watch 函数监听的不仅是指定目录，其内部的递归子目录也会一并监听。
问题来了，当前目录下存在一个 data.txt 文件和一个 txt_files 目录，而 txt_files 目录也存在一个 data.txt。那么如果将当前目录的 data.txt 移动到 txt_files 中，并同意覆盖，那么程序会输出什么呢？
{(<Change.deleted: 3>, '/Users/satori/Desktop/project/txt_files/data.txt'), (<Change.added: 1>, '/Users/satori/Desktop/project/txt_files/data.txt'), (<Change.deleted: 3>, '/Users/satori/Desktop/project/data.txt')}
此时输出的集合包含三个元组，因此该过程涉及到三次文件的变更。因为 txt_files 里面的文件被替换掉了，所以相当于先被删除、然后重新创建。而当前目录中的 data.txt 被移走了，因此相当于被删除了。
然后我们再通过 mkdir -p a/b/c 同时创建多级目录，程序输出如下：
{(<Change.added: 1>, '/Users/satori/Desktop/project/a/b/c'), (<Change.added: 1>, '/Users/satori/Desktop/project/a/b'), (<Change.added: 1>, '/Users/satori/Desktop/project/a')}
整个过程还是比较简单的，然后除了 watch 函数之外，还有一个 awatch。这两者的作用是一样的，参数也全部一样，只不过 awatch 需要和协程搭配，我们举个例子。
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()
来看一下效果：
结果没有问题，文件的变化都检测出来了。然后补充一点：watch 和 awatch 可以同时监听多个目录，因为第一个参数是 *paths。
我们同时监听多个目录来测试一下，先在当前目录创建两个子目录：boy 和 girl，然后分别监视它们。
输出正常，因此这两个函数可以监听任意多个目录。另外，由于目前监听的是当前目录的两个子目录，所以当前目录的文件变更就看不到了，因为它没有被监视。
以上就是这两个函数的基本用法，当然这两个函数还有其它参数：
这里简单介绍几个。
过滤器（watch_filter）
watchfiles 会监视目录的文件变化，但不是所有的文件都会记录。
watchfiles 有一个内置的过滤器，会将和业务无关的文件过滤掉，如果你还希望将其它格式的文件过滤掉，那么修改过滤器即可。
停止事件（stop_event）
监视文件的时候，迭代器是不会停止的，如果想自由控制它的结束，可以传递一个事件。
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 停止监视 """
是否递归监视（recursive）
如果该参数为 True，那么会递归监视子目录，否则只监视顶层目录。
其它参数基本很少用，就不再赘述了，有兴趣可以自己了解一下。
