Python使用watchfiles实现监控目录变更
作者：古明地觉的编程教室
在工作中难免会碰到这样的需求，监控指定目录，如果该目录下发生文件变更，那么进行一系列的处理。而如何监视一个目录，就是我们本次探讨的主题。
监视目录我们可以使用 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，那么会递归监视子目录，否则只监视顶层目录。
其它参数基本很少用，就不再赘述了，有兴趣可以自己了解一下。
到此这篇关于Python使用watchfiles实现监控目录变更的文章就介绍到这了,更多相关Python监控目录变更内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！