快捷導(dǎo)航

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

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

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

問(wèn)題來(lái)了，當(dāng)前目錄下存在一個(gè) data.txt 文件和一個(gè) txt_files 目錄，而 txt_files 目錄也存在一個(gè) data.txt。那么如果將當(dāng)前目錄的 data.txt 移動(dòng)到 txt_files 中，并同意覆蓋，那么程序會(huì)輸出什么呢？

{(<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')}

此時(shí)輸出的集合包含三個(gè)元組，因此該過(guò)程涉及到三次文件的變更。因?yàn)?txt_files 里面的文件被替換掉了，所以相當(dāng)于先被刪除、然后重新創(chuàng)建。而當(dāng)前目錄中的 data.txt 被移走了，因此相當(dāng)于被刪除了。

然后我們?cè)偻ㄟ^(guò) mkdir -p a/b/c 同時(shí)創(chuàng)建多級(jí)目錄，程序輸出如下：

{(<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')}

整個(gè)過(guò)程還是比較簡(jiǎn)單的，然后除了 watch 函數(shù)之外，還有一個(gè) awatch。這兩者的作用是一樣的，參數(shù)也全部一樣，只不過(guò) awatch 需要和協(xié)程搭配，我們舉個(gè)例子。

import?sys
import?asyncio
from?asyncio?import?StreamReader
from?watchfiles?import?awatch,?Change
#?監(jiān)視指定目錄
async?def?watch_files(path):
????#?awatch(...)?返回的是異步生成器，需要通過(guò)?async?for?遍歷
????async?for?change?in?awatch(path):
????????print("-"?*?20)
????????#?change?是一個(gè)集合，里面可能會(huì)涉及到多個(gè)文件的變更
????????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ù)，因?yàn)樗峭阶枞{(diào)用
#?這種調(diào)用在協(xié)程當(dāng)中是大忌，會(huì)阻塞整個(gè)線程，我們需要改造成異步模式
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ù)的具體細(xì)節(jié)暫時(shí)不用太關(guān)注
#?只需要知道它能異步讀取命令行即可，關(guān)于這方面的內(nèi)容后續(xù)會(huì)介紹
#?然后定義主協(xié)程
async?def?main():
????#?監(jiān)視當(dāng)前目錄
????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()

來(lái)看一下效果：

結(jié)果沒(méi)有問(wèn)題，文件的變化都檢測(cè)出來(lái)了。然后補(bǔ)充一點(diǎn)：watch 和 awatch 可以同時(shí)監(jiān)聽(tīng)多個(gè)目錄，因?yàn)榈谝粋€(gè)參數(shù)是 *paths。

我們同時(shí)監(jiān)聽(tīng)多個(gè)目錄來(lái)測(cè)試一下，先在當(dāng)前目錄創(chuàng)建兩個(gè)子目錄：boy 和 girl，然后分別監(jiān)視它們。

輸出正常，因此這兩個(gè)函數(shù)可以監(jiān)聽(tīng)任意多個(gè)目錄。另外，由于目前監(jiān)聽(tīng)的是當(dāng)前目錄的兩個(gè)子目錄，所以當(dāng)前目錄的文件變更就看不到了，因?yàn)樗鼪](méi)有被監(jiān)視。

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

這里簡(jiǎn)單介紹幾個(gè)。

過(guò)濾器（watch_filter）

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

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

停止事件（stop_event）

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

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()
????#?傳遞一個(gè)事件，準(zhǔn)確的說(shuō)，只要有 is_set 方法，任何對(duì)象都行
????asyncio.create_task(watch_files(".",?stop_event=event))
????#?當(dāng)?event.is_set()?為?True?的時(shí)候，停止監(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)視
"""

是否遞歸監(jiān)視（recursive）

如果該參數(shù)為 True，那么會(huì)遞歸監(jiān)視子目錄，否則只監(jiān)視頂層目錄。

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

到此這篇關(guān)于Python使用watchfiles實(shí)現(xiàn)監(jiān)控目錄變更的文章就介紹到這了,更多相關(guān)Python監(jiān)控目錄變更內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: