python watchdog 詳細講解

在實際的開發過程中，有時候需要通過 Python 去監聽某文件夾的變動，從而實現針對文件變化的操作。

Python 中有2個不錯的庫實現了該功能，分別是 pyinotify 和 watchdog 本篇博客為你介紹第三方模塊 watchdog 實現對文件夾的監控。

watchdog 安裝與準備

安裝就比較簡單了

pip install watchdog

項目地址是：https://pypi.org/project/watchdog/#description

文檔參考地址：https://python-watchdog.readthedocs.io/en/stable/

watchdog 是一個實時監控庫，其原理是通過操作系統的時間觸發，需要循環等待。

官方提供最簡單的入門案例

import sys import time import logging from watchdog.observers import Observer from watchdog.events import LoggingEventHandler if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') path = sys.argv[1] if len(sys.argv) > 1 else '.' event_handler = LoggingEventHandler() observer = Observer() observer.schedule(event_handler, path, recursive=True) observer.start() try: while True: time.sleep(1) finally: observer.stop() observer.join()

基于上述源碼，說明一下 watchdog 的相關用法。

observer = Observer()

創建一個觀察者對象。

observer.schedule()

聲明一個定時任務。

observer.start()

啟動定時任務。

observer.schedule() 的函數原型如下

schedule(event_handler, path, recursive=False)

該方法用于監視 path 路徑，并調用給定的事情 event_handler 。

最后一個參數 recursive 表示是否遞歸子目錄，即監聽子目錄，默認為 False。

start()

啟動線程，這里開啟了新的守護線程，主程序如果結束， 該線程也會停止。

每個線程對象只能調用1次，它安排對象的 run() 方法在單獨的控制線程中調用，如果在同一線程對象上多次調用此方法將引發RuntimeError。

重寫事件

接下來我們使用自定義的時間，實現對文件操作的監控

import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class MyEventHandler(FileSystemEventHandler): # 文件移動 def on_moved(self, event): print("文件移動觸發") print(event) def on_created(self, event): print("文件創建觸發") print(event) def on_deleted(self, event): print("文件刪除觸發") print(event) def on_modified(self, event): print("文件編輯觸發") print(event) if __name__ == '__main__': observer = Observer() # 創建觀察者對象 file_handler = MyEventHandler() # 創建事件處理對象 observer.schedule(file_handler, "./images", False) # 向觀察者對象綁定事件和目錄 observer.start() # 啟動 try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()

這里會出現一個問題，就是當復制進去一個文件時，會同時觸發 創建/編輯 事件，如下所示。

先說原因：

文件創建這個操作會觸發多種事件，包括 FileCreatedEvent ， FileModifiedEvent 事件，對應到代碼中就是 on_created 和 on_modified 方法被調用，其原因在于 f = open("file.txt", "w") 這樣文件創建動作會觸發 FileCreatedEvent 事件，執行 on_created 函數，往文件寫數據的 f.flush() 和 f.close() 操作，會觸發 FileModifiedEvent 事件，執行 on_modified 函數，所以觸發了 3次。

FileSystemEvent 文件類派生出來的子類包括如下內容

watchdog.events.FileCreatedEvent() ：文件被創建時觸發該事件；

watchdog.events.DirCreatedEvent() ：目錄被創建時觸發該事件；

watchdog.events.FileDeletedEvent() ：文件被刪除時觸發該事件；

watchdog.events.DirDeletedEvent() ：目錄被刪除時觸發該事件；

watchdog.events.FileModifiedEvent() ：文件被修改時觸發該事件；

watchdog.events.DirModifiedEvent() ：目錄被修改時觸發該事件；

watchdog.events.FileMovedEvent() ：文件被移動或重命名時觸發該事件（event.src_path 表示原路徑，還有 event.dest_path ）；

watchdog.events.DirMovedEvent() ：目錄被移動或重命名時觸發該事件（event.src_path 表示原路徑，還有 event.dest_path ）；

watchdog 默認提供的一些事件處理類

FileSystemEventHandler：文件，事件處理器的基類，用于處理事件；

PatternMatchingEventHandler：模式匹配文件；

RegexMatchingEventHandler：正則匹配文件；

LoggingEventHandler：記錄日志。

PatternMatchingEventHandler 函數原型如下

watchdog.events.PatternMatchingEventHandler(patterns=None,ignore_patterns=None,ignore_directories=False, case_sensitive=False)

該類會檢查觸發事件的 src_path 和 dest_path ，是否與 patterns 指定的模式匹配；

ignore_patterns ：需要排除不處理的模式，如果路徑匹配該模式則不處理

ignore_directories：為 True 表示不處理由目錄引起的事件；

case_sensitive：為 True 則表示路徑不區分大小寫。

RegexMatchingEventHandler 函數原型如下

watchdog.events.RegexMatchingEventHandler(regexes=[r".*"], ignore_regexes=[], ignore_directories=False, case_sensitive=False)

監聽指定內容

繼承監聽事件函數，然后監聽特定文件。

import time import os from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class MyHandler(FileSystemEventHandler): def on_modified(self, event): print(dir(event)) print(event.src_path) path = os.path.abspath(event.src_path) print(path) if path == r"F:\xxx\監控文件變動\imgs\1.log": # 監控指定文件 print("文件 %s 有編輯" % event.src_path) if __name__ == "__main__": event_handler = MyHandler() observer = Observer() observer.schedule(event_handler, path='./imgs', recursive=False) observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()

其中 event 對象的屬性如下所示：

event.is_directory：該事件是否由一個目錄觸發；

event.src_path：觸發該事件的文件或目錄路徑；

event.event_type：事件類型，例如 moved，deleted，created，modified；

event.key：元組格式返回 (event_type, src_path, is_directory)。

observer.schedule(event_handler, path, recursive=False) 的詳細說明

每一次調用 schedule() 對一個路徑（ path ）進行監控處理叫做 watch ， schedule() 方法會返回這個 watch ，我們可以對 watch 增加多個 event 事件處理器。

observer.add_handler_for_watch(event_handler, watch)：添加1個新的事件處理器到 ；

observer.remove_handler_for_watch(event_handler, watch)：從 watch 移除1個事件處理器；

observer.unschedule(watch)：移除1個 watch 及其所有事件處理器；

observer.unschedule_all()：移除所有 watch 及關聯的事件處理器；

observer.on_thread_stop()：等同于 observer.stop()

添加多個事件

import time import logging from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, LoggingEventHandler class MyHandler(FileSystemEventHandler): def on_created(self, event): print(event) if __name__ == "__main__": event_handler1 = MyHandler() observer = Observer() watch = observer.schedule(event_handler1, path='.', recursive=True) logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') event_handler2 = LoggingEventHandler() observer.add_handler_for_watch(event_handler2, watch) # 添加event handler observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()

監聽特定文件夾，特定后綴的文件

import time import logging import os from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, LoggingEventHandler logging.basicConfig(level=logging.INFO, filename='./video_log.log', filemode='a+', format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S', encoding="utf-8") class LogHandler(LoggingEventHandler): def on_created(self, event): path = event.src_path if event.is_directory: pass else: logging.info(path + "文件新增") class MyHandler(FileSystemEventHandler): def on_created(self, event): path = event.src_path file_name = os.path.basename(path) if file_name.endswith("mp4") or file_name.endswith("avi") or file_name.endswith("flv"): print("文件格式正確") else: pass print(event) if __name__ == "__main__": event_handler = MyHandler() observer = Observer() watch = observer.schedule(event_handler, path='./videos', recursive=False) log_handler = LogHandler() observer.add_handler_for_watch(log_handler, watch) # 寫入日志 observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()

Python

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。