Python使用watchfiles實現(xiàn)監(jiān)控目錄變更

更新時間：2023年09月15日 09:30:26 作者：古明地覺的編程教室

在工作中難免會碰到這樣的需求,監(jiān)控指定目錄,下面小編就來和大家介紹一下如何利用watchfiles 模塊實現(xiàn)監(jiān)控目錄的變更,感興趣的可以了解下

import?sys
import?asyncio
from?asyncio?import?StreamReader
from?watchfiles?import?awatch,?Change
#?監(jiān)視指定目錄
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 函數(shù)，因為它是同步阻塞調(diào)用
#?這種調(diào)用在協(xié)程當中是大忌，會阻塞整個線程，我們需要改造成異步模式
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?函數(shù)的具體細節(jié)暫時不用太關注
#?只需要知道它能異步讀取命令行即可，關于這方面的內(nèi)容后續(xù)會介紹
#?然后定義主協(xié)程
async?def?main():
????#?監(jiān)視當前目錄
????asyncio.create_task(watch_files("."))
????#?創(chuàng)建讀取器
????stdin_reader?=?await?read_from_stdin()
????while?True:
????????#?從命令行讀取輸入
????????command?=?await?stdin_reader.readline()
????????#?執(zhí)行命令
????????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 可以同時監(jiān)聽多個目錄，因為第一個參數(shù)是 *paths。

我們同時監(jiān)聽多個目錄來測試一下，先在當前目錄創(chuàng)建兩個子目錄：boy 和 girl，然后分別監(jiān)視它們。

輸出正常，因此這兩個函數(shù)可以監(jiān)聽任意多個目錄。另外，由于目前監(jiān)聽的是當前目錄的兩個子目錄，所以當前目錄的文件變更就看不到了，因為它沒有被監(jiān)視。

以上就是這兩個函數(shù)的基本用法，當然這兩個函數(shù)還有其它參數(shù)：

這里簡單介紹幾個。

過濾器（watch_filter）

watchfiles 會監(jiān)視目錄的文件變化，但不是所有的文件都會記錄。

watchfiles 有一個內(nèi)置的過濾器，會將和業(yè)務無關的文件過濾掉，如果你還希望將其它格式的文件過濾掉，那么修改過濾器即可。

停止事件（stop_event）

監(jiān)視文件的時候，迭代器是不會停止的，如果想自由控制它的結束，可以傳遞一個事件。

import?asyncio
from?watchfiles?import?awatch
async?def?watch_files(*paths,?stop_event):
????async?for?_?in?awatch(*paths,?stop_event=stop_event):
????????pass
????print("停止監(jiān)視")
async?def?main():
????event?=?asyncio.Event()
????#?傳遞一個事件，準確的說，只要有 is_set 方法，任何對象都行
????asyncio.create_task(watch_files(".",?stop_event=event))
????#?當?event.is_set()?為?True?的時候，停止監(jiān)視
????print("is_set:?",?event.is_set())
????await?asyncio.sleep(3)??#?sleep?3
????event.set()
????print("三秒后,?is_set:?",?event.is_set())
????#?等待子協(xié)程打印完畢
????await?asyncio.sleep(0.1)
asyncio.run(main())
"""
is_set:??False
三秒后,?is_set:??True
停止監(jiān)視
"""