Watchdog库是Python中一个用于监控文件系统变化的第三方库。它能够实时监测文件或目录的创建、修改、删除等操作，并在这些事件发生时触发相应的处理逻辑，因此也被称为文件看门狗。
![[python] 基于WatchDog库实现文件系统监控](http://www.itfaba.com/wp-content/themes/kemi/images/loading.gif)
![[python] 基于WatchDog库实现文件系统监控](http://www.itfaba.com/wp-content/uploads/2025/04/20250427_680e7c574a27c.jpg)
Watchdog库的官方仓库见：watchdog，Watchdog库的官方文档见：watchdog-doc。Watchdog库的安装命令如下：
python -m pip install -U watchdog
注意：Watchdog库最新版本（2.1.5以上版本）需在Python3.9以上版本运行。若使用Python3.4或3.5，应选用Watchdog版本低于1.0.0；若使用Python3.6至3.8，应选用Watchdog版本低于2.0.0。
1 使用示例
基础示例
以下示例程序将以递归方式监控指定目录及其子文件夹中文件系统的变化情况，包括文件或文件夹的创建、修改、移动、删除，并将这些变化记录到控制台中：
import sys # 导入logging模块，用于记录程序运行时的日志信息 import logging # 从watchdog.observers模块中导入Observer类，用于监控文件系统的变化 from watchdog.observers import Observer # 从watchdog.events模块中导入LoggingEventHandler类，用于处理文件系统事件并记录日志 from watchdog.events import LoggingEventHandler if __name__ == "__main__": # 配置日志记录的基本设置 # level=logging.INFO表示只记录INFO级别及以上的日志信息 # format='%(asctime)s - %(message)s'指定日志的输出格式，包含时间戳和日志消息 # datefmt='%Y-%m-%d %H:%M:%S'指定时间戳的格式 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') # 指定要监控的文件夹路径，该文件夹必须存在 path = "demo_folder" # 创建一个LoggingEventHandler对象，用于处理文件系统事件并记录日志 event_handler = LoggingEventHandler() # 创建一个Observer对象，用于监控文件系统的变化 observer = Observer() # 安排观察者监控指定路径下的文件系统事件 # event_handler是事件处理程序 # path是要监控的路径 # recursive=True表示递归监控该路径下的所有子文件夹 observer.schedule(event_handler, path, recursive=True) # 启动观察者，开始监控文件系统的变化 observer.start() try: # 当观察者处于活动状态时，持续循环 while observer.is_alive(): # 等待1秒，让观察者有时间处理文件系统事件 observer.join(1) finally: # 停止观察者，结束文件系统的监控 observer.stop() # 等待观察者线程结束 observer.join() Watchdog事件类介绍
Watchdog库提供了一系列事件类，用于监控文件系统的各种变化，这些变化包括：
- 创建事件
on_create：涵盖文件和文件夹的创建情况，当有新的文件或者文件夹在监控路径下生成时触发相应事件。 - 删除事件
on_delete：当文件或者文件夹被从监控路径中移除时触发，分别有针对文件和文件夹的不同事件类。 - 修改事件
on_modified：文件夹内新增、删除文件或子文件夹等内容变化时触发文件夹修改事件。 - 移动 / 重命名事件
on_moved：文件或文件夹的位置发生移动，或者名称被更改时触发相应事件，该事件会记录原路径和新路径。
基于这些变化Watchdog库中存在8个事件类：
| 事件类名 | 触发场景 |
|---|---|
FileCreatedEvent |
新文件被创建时触发 |
DirCreatedEvent |
新文件夹被创建时触发 |
FileDeletedEvent |
文件被删除时触发 |
DirDeletedEvent |
文件夹被删除时触发 |
FileModifiedEvent |
文件内容被修改时触发 |
DirModifiedEvent |
文件夹内容（如添加、删除子文件或子文件夹）发生变化时触发 |
FileMovedEvent |
文件被移动或重命名时触发 |
DirMovedEvent |
文件夹被移动或重命名时触发 |
注意：在某些情形下，多个Watchdog事件可能会一同触发，这往往和文件系统操作的特性以及Watchdog对这些操作的响应机制有关。例如，当移动一个文件时，会同时触发FileMovedEvent（文件移动事件）和FileDeletedEvent（文件删除事件），因为移动文件的操作在源路径上相当于删除文件，同时在目标路径上会触发FileCreatedEvent（文件创建事件）。
以下代码的主要功能是监控指定目录下文件系统的各种事件，如文件或目录的创建、删除、移动和修改，并对这些事件进行重写和相应的处理：
# 该类用于处理文件系统事件，如文件的创建、删除、修改等 from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer # 定义一个自定义的文件事件处理类，继承自FileSystemEventHandler class FileEventHandler(FileSystemEventHandler): def __init__(self): FileSystemEventHandler.__init__(self) # 初始化一个空列表，用于存储文件名 self.file_names = [] # 当文件或目录被移动时触发此方法 def on_moved(self, event): # 判断是否是目录移动 if event.is_directory: # 打印目录移动的信息 print(f"目录从 {event.src_path} 移动到 {event.dest_path}") else: # 打印文件移动的信息 print(f"文件从 {event.src_path} 移动到 {event.dest_path}") # 当文件或目录被创建时触发此方法 def on_created(self, event): # 判断是否是目录创建 if event.is_directory: # 打印目录创建的信息 print(f"目录已创建: {event.src_path}") else: # 打印文件创建的信息 print(f"文件已创建: {event.src_path}") # 从文件的完整路径中提取文件名 file_full_name = str(event.src_path.split('/')[-1]) if file_full_name.endswith('.csv'): # 如果是csv文件，将其添加到文件名列表中 self.file_names.append(file_full_name) # 打印文件名列表 print(self.file_names) # 当文件或目录被删除时触发此方法 def on_deleted(self, event): # 判断是否是目录删除 if event.is_directory: # 打印目录删除的信息 print(f"目录已删除: {event.src_path}") else: # 打印文件删除的信息 print(f"文件已删除: {event.src_path}") # 当文件或目录被修改时触发此方法，注释代码表示不触发修改事件 # def on_modified(self, event): # pass if __name__ == "__main__": # 创建一个Observer对象，用于监控文件系统的变化 observer = Observer() # 创建一个自定义的文件事件处理对象 event_handler = FileEventHandler() # 定义要监控的目录 watch_directory = "demo_folder" # 将事件处理对象和要监控的目录添加到观察者中，并设置为递归监控 observer.schedule(event_handler, watch_directory, recursive=True) # 启动观察者，开始监控文件系统的变化 observer.start() try: # 当观察者处于活动状态时，持续循环 while observer.is_alive(): # 等待1秒，让观察者有时间处理文件系统事件 observer.join(1) finally: # 停止观察者，结束文件系统的监控 observer.stop() # 等待观察者线程结束 observer.join() 快照功能
采用基于定时器的批处理策略，合并并延迟频繁的文件变更事件，能够有效降低系统负载。具体做法是：当系统检测到文件首次变动时，启动可重置的计时器，在设定的延迟期内累积后续变更事件，待计时器超时后，再统一通过对比当前文件系统状态与上次缓存快照，可精准识别所有变更项。具体实现代码如下：
import time import os import threading from watchdog.observers import Observer from watchdog.events import * from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff class FileChangeHandler(FileSystemEventHandler): def __init__(self, monitored_path, delay=0.5): # 调用父类构造函数 super().__init__() # 被监控的目录路径 self.monitored_path = monitored_path self.delay = delay # 定时器对象，用于延迟检查 self.delay_timer = None # 快照用于记录指定目录在某个时间点的状态，包含了目录下所有文件和子目录的信息， # 记录目录的初始快照 self.initial_snapshot = DirectorySnapshot(self.monitored_path) def on_any_event(self, event): # 通过定时器减少不必要的检查 # 如果定时器已存在，取消它 if self.delay_timer: self.delay_timer.cancel() # 创建一个新的定时器，延迟后执行检查快照的操作 # 通过快照检测文件系统变化 self.delay_timer = threading.Timer(self.delay, self._check_directory_changes) self.delay_timer.start() def _check_directory_changes(self): # 获取当前目录的新快照 new_snapshot = DirectorySnapshot(self.monitored_path) # 计算新旧快照之间的差异 snapshot_difference = DirectorySnapshotDiff(self.initial_snapshot, new_snapshot) # 更新初始快照为新快照 self.initial_snapshot = new_snapshot # 清空定时器 self.delay_timer = None # 打印目录变化信息 self._print_changes(snapshot_difference) def _print_changes(self, diff): # 打印文件和目录的创建、删除、修改和移动信息 print("创建的文件:", diff.files_created) print("删除的文件:", diff.files_deleted) print("修改的文件:", diff.files_modified) print("移动的文件:", diff.files_moved) print("修改的目录:", diff.dirs_modified) print("移动的目录:", diff.dirs_moved) print("删除的目录:", diff.dirs_deleted) print("创建的目录:", diff.dirs_created) class DirectoryMonitor: def __init__(self, monitored_path): # 被监控的目录路径 self.monitored_path = monitored_path # 创建一个观察者对象 self.directory_observer = Observer() def start_monitoring(self): # 创建文件变化处理对象 change_handler = FileChangeHandler(self.monitored_path) # 安排观察者监控指定目录，递归地监控子目录 self.directory_observer.schedule(change_handler, self.monitored_path, recursive=True) # 启动观察者 self.directory_observer.start() def stop_monitoring(self): # 停止观察者 self.directory_observer.stop() # 等待观察者线程结束 self.directory_observer.join() if __name__ == "__main__": monitor = DirectoryMonitor("demo_folder") # 开始监控 monitor.start_monitoring() try: while True: time.sleep(1) except KeyboardInterrupt: monitor.stop_monitoring() 多文件夹监控
以下代码展示了通过schedule方法为每个目录注册事件处理程序，实现对多个目录的同步监控：
from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class CustomFileEventHandler(FileSystemEventHandler): def on_created(self, event): patterns = [“*.jpg”， “*.png”， “*.gif”] # 只监控图片文件 def on_modified(self， event)： print(f“图片文件有变化：{event.src_path}”) entity = "directory" if event.is_directory else "file" print(f"{entity} on_created: {event.src_path}") if __name__ == "__main__": file_observer = Observer() monitored_directories = ['demo_folder', 'demo_folder2'] file_event_handler = CustomFileEventHandler() for directory in monitored_directories: # 为每个目录注册事件处理程序 file_observer.schedule(file_event_handler, directory, recursive=True) file_observer.start() try: # 当观察者处于活动状态时，持续循环 while file_observer.is_alive(): # 等待1秒，让观察者有时间处理文件系统事件 file_observer.join(1) finally: # 停止观察者，结束文件系统的监控 file_observer.stop() # 等待观察者线程结束 file_observer.join() 
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/1.jpg&w=218&h=124&zc=1)
