使用 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，那麼會遞歸監視子目錄，否則只監視頂層目錄。

其它參數基本很少用，就不再贅述了，有興趣可以自己瞭解一下。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/QVz5RIx11Vy6W3X4zJQaiQ