最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Python使用Watchdog庫監(jiān)聽文件變化

 更新時間:2026年01月16日 09:36:17   作者:禿了也弱了。  
這篇文章主要介紹了Python的watchdog庫,用于高效監(jiān)聽文件系統(tǒng)事件,它支持跨平臺,性能優(yōu)越,適用于工業(yè)級開發(fā),核心組件包括EventHandler、Observer和FileSystemEvent,需要的朋友可以參考下

一、簡介

官方文檔:https://python-watchdog.readthedocs.io/en/stable/index.html

watchdog 是 Python 生態(tài)中處理文件系統(tǒng)事件的主流庫,跨平臺(Windows/Linux/macOS)、效率高,支持監(jiān)聽文件 / 目錄的創(chuàng)建、修改、刪除、移動等事件,是工業(yè)級的解決方案。

# 安裝
pip install watchdog

二、使用

核心組件:
EventHandler 事件處理器:定義監(jiān)控到文件事件后要執(zhí)行的邏輯(核心,需自定義 / 繼承)
Observer 觀察者:負責(zé)啟動監(jiān)控線程,監(jiān)聽文件系統(tǒng)事件,并將事件分發(fā)給處理器
FileSystemEvent 事件對象:包含事件類型(創(chuàng)建 / 修改等)、文件路徑、是否是目錄等信息

watchdog基于系統(tǒng)原生的文件監(jiān)控接口(如 Linux 的 inotify、Windows 的 ReadDirectoryChangesW),性能遠高于手動輪詢;
對于頻繁修改的文件(如每秒多次修改),可能會觸發(fā)多次on_modified事件,可通過加時間戳防抖處理;
跨平臺使用時,路徑建議用os.path.abspath()轉(zhuǎn)為絕對路徑,避免兼容問題。

1、監(jiān)聽單個目錄的所有文件事件

運行代碼前,先創(chuàng)建./test_dir目錄;

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 自定義事件處理器:繼承FileSystemEventHandler,重寫對應(yīng)事件方法
class MyFileHandler(FileSystemEventHandler):
    # 監(jiān)聽文件/目錄創(chuàng)建事件
    def on_created(self, event):
        # event.is_directory 判斷是否是目錄,event.src_path 是觸發(fā)事件的文件路徑
        if event.is_directory:
            print(f"[創(chuàng)建目錄] {event.src_path}")
        else:
            print(f"[創(chuàng)建文件] {event.src_path}")

    # 監(jiān)聽文件/目錄修改事件
    def on_modified(self, event):
        if not event.is_directory:  # 只關(guān)注文件修改(排除目錄修改)
            print(f"[修改文件] {event.src_path}")

    # 監(jiān)聽文件/目錄刪除事件
    def on_deleted(self, event):
        if event.is_directory:
            print(f"[刪除目錄] {event.src_path}")
        else:
            print(f"[刪除文件] {event.src_path}")

    # 監(jiān)聽文件/目錄移動事件
    def on_moved(self, event):
        # event.dest_path 是移動后的路徑
        if event.is_directory:
            print(f"[移動目錄] {event.src_path} -> {event.dest_path}")
        else:
            print(f"[移動文件] {event.src_path} -> {event.dest_path}")

if __name__ == "__main__":
    # 要監(jiān)控的目錄路徑(絕對路徑/相對路徑都可)
    monitor_dir = "./test_dir"
    # 創(chuàng)建事件處理器實例
    event_handler = MyFileHandler()
    # 創(chuàng)建觀察者實例
    observer = Observer()
    # 為觀察者綁定處理器和監(jiān)控目錄
    # recursive=False 表示只監(jiān)控當(dāng)前目錄,不包含子目錄
    observer.schedule(event_handler, path=monitor_dir, recursive=False)
    
    # 啟動觀察者(后臺線程,非阻塞)
    observer.start()
    print(f"開始監(jiān)控目錄:{monitor_dir}(按 Ctrl+C 停止)")
    
    try:
        # 保持主線程運行,避免程序退出
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 捕獲 Ctrl+C,停止觀察者
        observer.stop()
        print("\n停止監(jiān)控")
    # 等待觀察者線程結(jié)束
    observer.join()

2、監(jiān)聽特定類型文件(如.txt)

僅監(jiān)控 ./test_dir下.txt文件的修改事件,忽略其他類型文件。

import time
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class TxtFileHandler(FileSystemEventHandler):
    # 重寫修改事件,增加文件類型過濾
    def on_modified(self, event):
        # 排除目錄,只處理文件
        if not event.is_directory:
            # 獲取文件后綴名
            file_ext = os.path.splitext(event.src_path)[1]
            # 只處理 .txt 文件
            if file_ext == ".txt":
                print(f"[修改TXT文件] {event.src_path}")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = TxtFileHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"開始監(jiān)控 {monitor_dir} 下的 .txt 文件修改(按 Ctrl+C 停止)")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

3、遞歸監(jiān)控 - 監(jiān)聽目錄及所有子目錄

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class RecursiveHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"[創(chuàng)建] {event.src_path} (目錄: {event.is_directory})")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = RecursiveHandler()
    observer = Observer()
    # 關(guān)鍵:recursive=True 開啟遞歸監(jiān)控
    observer.schedule(handler, monitor_dir, recursive=True)
    observer.start()
    
    print(f"遞歸監(jiān)控 {monitor_dir} 及其所有子目錄(按 Ctrl+C 停止)")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

4、應(yīng)用:監(jiān)控到文件修改后自動執(zhí)行腳本

監(jiān)控配置文件 ./config.ini,當(dāng)文件修改后,自動重新加載配置并執(zhí)行指定腳本(如重啟服務(wù)、刷新緩存)。

import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ConfigChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 只監(jiān)控 config.ini 文件
        if event.src_path.endswith("config.ini") and not event.is_directory:
            print(f"\n[配置文件修改] {event.src_path}")
            print("開始重新加載配置并執(zhí)行腳本...")
            
            # 自定義動作1:打印配置內(nèi)容(模擬加載配置)
            with open(event.src_path, "r", encoding="utf-8") as f:
                config_content = f.read()
                print(f"最新配置內(nèi)容:\n{config_content}")
            
            # 自定義動作2:執(zhí)行外部腳本(如 restart_service.py)
            # 注意:需提前創(chuàng)建 restart_service.py,或替換為你的腳本路徑
            try:
                result = subprocess.run(
                    ["python", "restart_service.py"],
                    capture_output=True,
                    text=True,
                    timeout=10
                )
                print(f"腳本執(zhí)行結(jié)果:\nstdout: {result.stdout}\nstderr: {result.stderr}")
            except Exception as e:
                print(f"腳本執(zhí)行失?。簕e}")

if __name__ == "__main__":
    monitor_dir = "./"  # 監(jiān)控當(dāng)前目錄
    handler = ConfigChangeHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"監(jiān)控配置文件 config.ini 修改(按 Ctrl+C 停止)")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

5、監(jiān)控多個目錄 - 同時監(jiān)聽不同路徑

同時監(jiān)控./log_dir./config_dir兩個目錄的文件變化

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class MultiDirHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 區(qū)分不同目錄的事件
        if "log_dir" in event.src_path:
            print(f"[日志目錄] 新增文件/目錄:{event.src_path}")
        elif "config_dir" in event.src_path:
            print(f"[配置目錄] 新增文件/目錄:{event.src_path}")

if __name__ == "__main__":
    # 定義多個監(jiān)控目錄
    dir1 = "./log_dir"
    dir2 = "./config_dir"
    
    handler = MultiDirHandler()
    observer = Observer()
    
    # 為觀察者添加多個監(jiān)控任務(wù)
    observer.schedule(handler, dir1, recursive=False)
    observer.schedule(handler, dir2, recursive=False)
    
    observer.start()
    print(f"同時監(jiān)控 {dir1} 和 {dir2}(按 Ctrl+C 停止)")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

6、動態(tài)啟停監(jiān)控 - 暫停 / 恢復(fù)文件監(jiān)控

支持手動觸發(fā)暫停監(jiān)控,一段時間后恢復(fù)(比如臨時維護目錄時暫停監(jiān)控)。

import time
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class SimpleHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"[監(jiān)控中] 修改文件:{event.src_path}")

def control_observer(observer):
    """單獨線程處理監(jiān)控的啟??刂?""
    while True:
        cmd = input("\n輸入命令(pause/continue/exit):").strip().lower()
        if cmd == "pause":
            observer.stop()  # 暫停監(jiān)控
            print("監(jiān)控已暫停")
        elif cmd == "continue":
            observer.start()  # 恢復(fù)監(jiān)控
            print("監(jiān)控已恢復(fù)")
        elif cmd == "exit":
            observer.stop()
            print("退出程序")
            break
        else:
            print("無效命令,請輸入 pause/continue/exit")

if __name__ == "__main__":
    monitor_dir = "./test_dir"
    handler = SimpleHandler()
    observer = Observer()
    observer.schedule(handler, monitor_dir, recursive=False)
    observer.start()
    
    print(f"開始監(jiān)控 {monitor_dir}(可輸入命令控制)")
    # 啟動控制線程,避免阻塞主線程
    control_thread = threading.Thread(target=control_observer, args=(observer,))
    control_thread.daemon = True  # 主線程退出時,控制線程也退出
    control_thread.start()
    
    # 等待控制線程結(jié)束
    control_thread.join()
    observer.join()

到此這篇關(guān)于Python使用Watchdog庫監(jiān)聽文件變化的文章就介紹到這了,更多相關(guān)Python Watchdog監(jiān)聽文件變化內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

最新評論

樟树市| 玉门市| 河北区| 云浮市| 昌邑市| 重庆市| 镇宁| 永济市| 石城县| 沭阳县| 安顺市| 临猗县| 乌恰县| 河南省| 肇庆市| 庐江县| 潮安县| 靖安县| 扶余县| 孟村| 梁河县| 菏泽市| 留坝县| 会宁县| 曲松县| 忻州市| 合水县| 海盐县| 中宁县| 富顺县| 卓资县| 泰兴市| 韶山市| 广河县| 蚌埠市| 稻城县| 会宁县| 金坛市| 哈尔滨市| 尼玛县| 清涧县|