什么是 watchdog？

watchdog 是一个跨平台的 Python 库，它通过监听操作系统的原生 API（如 Linux 的 inotify, macOS 的 FSEvents, Windows 的 ReadDirectoryChangesW）来高效地检测文件系统事件。

核心优势：

• 高效：直接利用操作系统内核机制，比轮询（每隔一段时间检查文件是否变化）性能高得多。
• 跨平台：代码无需修改即可在 Windows, macOS, Linux 上运行。
• 易用：提供了简洁的 API,上手非常快。

安装 watchdog

在使用之前，你需要先安装它，打开你的终端或命令行,运行：

pip install watchdog

核心概念

要使用 watchdog,你需要理解三个核心类：

1. Observer (观察者)：这是事件监控的“引擎”，它会启动一个后台线程，持续监听指定路径下的文件系统事件，你可以把它想象成一个“监控摄像头”。
2. FileSystemEventHandler (文件系统事件处理器)：这是一个“事件处理器”的基类，你可以继承这个类，并重写你感兴趣的事件方法（如 on_created, on_modified 等）。
3. Event (事件)：当文件系统发生某个动作（如创建、修改）时，Observer 会捕获到一个 Event 对象，并将其传递给相应的事件处理方法，这个对象包含了事件的详细信息，如文件路径、事件类型等。

工作流程：

1. 创建一个 FileSystemEventHandler 的子类,并重写需要处理的事件方法。
2. 创建一个 Observer 实例。
3. 将你的事件处理器和要监控的路径（以及可选的递归监控设置）注册到 Observer 中。
4. 启动 Observer。
5. 当文件系统事件发生时，Observer 会调用你定义的事件处理方法。
6. 当不再需要监控时，停止并 join Observer。

基础使用示例

下面是一个最简单的例子,监控当前目录下的文件创建和修改事件。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 1. 创建一个自定义的事件处理器
class MyEventHandler(FileSystemEventHandler):
    """
    自定义事件处理器，处理文件创建和修改事件。
    """
    def on_created(self, event):
        """
        当文件/目录被创建时被调用。
        """
        # event.is_directory 判断事件是否是目录
        print(f"检测到创建事件: {event.src_path}")
    def on_modified(self, event):
        """
        当文件/目录被修改时被调用。
        """
        if not event.is_directory:
            print(f"检测到修改事件: {event.src_path}")
# 2. 设置要监控的路径
path_to_watch = '.'  # 当前目录
# 3. 创建 Observer 实例
event_handler = MyEventHandler()
observer = Observer()
# 4. 将事件处理器和路径注册到 Observer
# recursive=True 表示递归监控子目录
observer.schedule(event_handler, path_to_watch, recursive=True)
# 5. 启动 Observer
print(f"开始监控路径: {path_to_watch}")
observer.start()
try:
    # 6. 保持主线程运行，防止程序退出
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    # 7. 当按下 Ctrl+C 时，停止 Observer
    print("\n停止监控...")
    observer.stop()
# 8. 等待 Observer 线程完全结束
observer.join()
print("监控已停止。")

如何运行这个示例？

1. 将上述代码保存为 watcher.py。
2. 在终端中运行 python watcher.py。
3. 打开另一个终端，在 watcher.py 所在的目录下创建或修改一个文件，touch new_file.txt 或 echo "hello" >> existing_file.txt。
4. 你会看到第一个终端的输出中出现了相应的事件信息。

监控更多事件类型

FileSystemEventHandler 提供了多种事件方法,你可以根据需要重写它们：

• on_any_event(self, event): 所有事件都会触发。
• on_created(self, event): 文件/目录被创建。
• on_deleted(self, event): 文件/目录被删除。
• on_modified(self, event): 文件/目录被修改。
• on_moved(self, event): 文件/目录被移动或重命名。

on_moved 事件特别说明： 当文件被移动或重命名时,会产生两个事件：

1. 一个 MOVED_FROM 事件,表示文件从原路径离开。
2. 一个 MOVED_TO 事件,表示文件到达新路径。

watchdog 会将这两个事件组合成一个 FileMovedEvent 对象，并传递给 on_moved 方法,这个对象有两个重要属性：

• event.src_path: 原路径。
• event.dest_path: 新路径。

示例：监控移动/重命名事件

from watchdog.events import FileSystemEventHandler
class MoveHandler(FileSystemEventHandler):
    def on_moved(self, event):
        if not event.is_directory:
            print(f"文件被移动/重命名: 从 {event.src_path} 到 {event.dest_path}")

高级用法

1 使用 PatternMatchingEventHandler

如果你只想监控特定类型的文件（例如所有 .log 文件），使用 PatternMatchingEventHandler 会更方便。

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
# 只监控 .log 文件，忽略 .tmp 文件
event_handler = PatternMatchingEventHandler(
    patterns=["*.log"],      # 要匹配的文件模式
    ignore_patterns=["*.tmp"], # 要忽略的文件模式
    ignore_directories=False,  # 不忽略目录
    case_sensitive=True      # 是否区分大小写
)
event_handler.on_modified = lambda event: print(f"日志文件被修改: {event.src_path}")
path = '.'
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

2 使用 RegexMatchingEventHandler

如果你需要更复杂的匹配规则,可以使用正则表达式。

from watchdog.events import RegexMatchingEventHandler
event_handler = RegexMatchingEventHandler(regexes=[r".*\.txt$"]) # 匹配所有 .txt 文件
event_handler.on_created = lambda event: print(f"检测到新的文本文件: {event.src_path}")

完整的实用示例：实时同步目录

下面是一个稍微复杂但非常实用的例子：实现一个简单的“实时同步”工具，它会监控一个源目录,并将任何新创建或修改的文件同步到目标目录。

import time
import os
import shutil
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
    def __init__(self, src_dir, dest_dir):
        self.src_dir = src_dir
        self.dest_dir = dest_dir
        # 确保目标目录存在
        os.makedirs(self.dest_dir, exist_ok=True)
    def on_created(self, event):
        if not event.is_directory:
            self.sync_file(event.src_path)
    def on_modified(self, event):
        if not event.is_directory:
            self.sync_file(event.src_path)
    def sync_file(self, src_path):
        """
        将源文件同步到目标目录。
        """
        # 计算相对于源目录的相对路径
        relative_path = os.path.relpath(src_path, self.src_dir)
        dest_path = os.path.join(self.dest_dir, relative_path)
        print(f"同步文件: {src_path} -> {dest_path}")
        # 确保目标文件的目录存在
        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
        # 复制文件
        shutil.copy2(src_path, dest_path)
if __name__ == "__main__":
    source_directory = "./source"  # 源目录
    destination_directory = "./destination" # 目标目录
    # 检查源目录是否存在
    if not os.path.exists(source_directory):
        os.makedirs(source_directory)
        print(f"已创建源目录: {source_directory}")
    print(f"开始监控源目录: {source_directory}")
    print(f"同步到目标目录: {destination_directory}")
    event_handler = SyncHandler(source_directory, destination_directory)
    observer = Observer()
    observer.schedule(event_handler, source_directory, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
    print("同步工具已停止。")

如何测试这个同步工具？

1. 创建 source 目录。
2. 运行脚本 python sync_tool.py。
3. 在 source 目录下创建文件、修改文件内容、创建子目录等，你会看到 destination 目录下实时同步了这些变化。

注意事项

1. 性能开销：监控大量文件或非常频繁的文件系统操作可能会对性能产生影响,确保你的事件处理逻辑尽可能高效。
2. 事件风暴：某些应用程序（如文本编辑器）在保存文件时可能会产生多个连续的 on_modified 事件，你可能需要加入一些逻辑（如延迟处理）来避免重复操作。
3. 网络文件系统：watchdog 在网络文件系统（如 NFS, SMB）上的表现可能不如本地文件系统稳定,因为网络延迟和文件系统语义的差异。
4. 权限问题：确保运行脚本的用户有权限访问和监控你指定的目录。

希望这份详细的指南能帮助你熟练地使用 watchdog！