什么是 watchdog？

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

它的核心优势在于：

• 高效：直接利用操作系统底层机制，而不是轮询检查文件状态,减少了资源消耗。
• 跨平台：一套代码可以在 Windows, macOS, Linux 上运行。
• 易用：提供了简洁的 API,可以快速上手。

安装

你需要安装 watchdog，打开你的终端或命令行,运行：

pip install watchdog

核心概念

理解 watchdog 的几个核心类是掌握它的关键：

• Observer：这是最重要的类，它是一个“观察者”或“监听器”的容器，你创建一个 Observer 实例，然后将一个或多个事件处理器（EventHandler）注册到它上面，并指定要监控的路径，你启动 Observer,它就会在后台线程中监听文件系统事件。
• EventHandler：这是一个抽象基类，定义了当文件系统事件发生时应该调用的方法，你不能直接使用它，而是需要继承它并重写其方法。
  • on_created(event): 当文件或目录被创建时调用。
  • on_modified(event): 当文件或目录被修改时调用。
  • on_deleted(event): 当文件或目录被删除时调用。
  • on_moved(event): 当文件或目录被移动或重命名时调用。
• FileSystemEvent：当一个事件发生时（如文件被修改），EventHandler 的方法会接收到一个 FileSystemEvent 对象，这个对象包含了事件的详细信息，
  • event.src_path: 事件源文件的路径（被修改的文件路径）。
  • event.event_type: 事件的类型，是 'created', 'modified', 'deleted', 或 'moved' 中的一个。
  • event.is_directory: 一个布尔值,指示事件源是否是一个目录。
  • event.dest_path (仅对 'moved' 事件): 文件被移动到的目标路径。

基础使用示例

下面是一个最简单的例子，它会在当前目录下监控文件和目录的创建、修改和删除事件。

1. 使用 FileSystemEventHandler (最常用)

我们创建一个自定义的处理器，继承自 watchdog.events.FileSystemEventHandler。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 1. 创建一个自定义的事件处理器
class MyEventHandler(FileSystemEventHandler):
    """
    自定义事件处理器，打印文件系统事件。
    """
    def on_created(self, event):
        if event.is_directory:
            print(f"目录被创建: {event.src_path}")
        else:
            print(f"文件被创建: {event.src_path}")
    def on_modified(self, event):
        if event.is_directory:
            print(f"目录被修改: {event.src_path}")
        else:
            print(f"文件被修改: {event.src_path}")
    def on_deleted(self, event):
        if event.is_directory:
            print(f"目录被删除: {event.src_path}")
        else:
            print(f"文件被删除: {event.src_path}")
    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}")
if __name__ == "__main__":
    # 2. 指定要监控的路径
    path = "."  # 监控当前目录
    # 3. 创建事件处理器实例
    event_handler = MyEventHandler()
    # 4. 创建 Observer 实例并注册事件处理器和路径
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)  # recursive=True 表示递归监控子目录
    # 5. 启动 observer
    print(f"开始监控路径: {path}")
    observer.start()
    try:
        # 6. 保持主线程运行，直到被中断
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 7. 当按下 Ctrl+C 时，停止 observer
        observer.stop()
        print("停止监控。")
    observer.join()

如何运行这个示例：

1. 将上述代码保存为 watcher.py。
2. 在终端中运行 python watcher.py。
3. 在另一个终端中，进入 watcher.py 所在的目录，执行一些操作，
   • 创建文件：touch new_file.txt
   • 修改文件：echo "hello" > new_file.txt
   • 创建目录：mkdir new_dir
   • 移动文件：mv new_file.txt renamed_file.txt
   • 删除文件：rm renamed_file.txt
   • 删除目录：rmdir new_dir

你会在第一个终端中看到对应的打印信息。

进阶使用示例

在实际应用中，我们通常不只是打印日志，而是要执行一些具体的操作，比如上传文件、重新加载配置等。

1. 监控特定类型的文件

假设我们只想监控 .py 文件的修改事件。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class PyFileEventHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 只处理文件，并且是 .py 后缀
        if not event.is_directory and event.src_path.endswith('.py'):
            print(f"检测到 Python 文件被修改: {event.src_path}")
            # 在这里可以添加你的逻辑，比如自动运行测试、重新加载服务等
            #  os.system("python my_app.py")
if __name__ == "__main__":
    path = "."
    event_handler = PyFileEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    print("开始监控 .py 文件的修改...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

2. 处理事件中的异常

在事件处理函数中，最好加上 try...except 块,以防止一个文件的处理失败影响到其他文件的监控。

class RobustEventHandler(FileSystemEventHandler):
    def on_modified(self, event):
        try:
            if not event.is_directory:
                print(f"处理文件修改: {event.src_path}")
                # 你的业务逻辑
                # process_file(event.src_path)
        except Exception as e:
            print(f"处理文件 {event.src_path} 时发生错误: {e}")
# ... 后续代码和之前一样 ...

常见问题与最佳实践

1. 递归监控 (recursive)

observer.schedule(event_handler, path, recursive=True) 中的 recursive 参数非常重要：

• True (默认): 监控指定路径下的所有子目录和文件,这是最常用的模式。
• False: 只监控指定的路径本身,不进入其子目录。

2. 避免无限循环

一个常见的陷阱是：你的监控脚本在检测到文件变化后，会去修改或写入那个文件，从而再次触发事件,形成一个无限循环。

示例错误：

def on_modified(self, event):
    # 错误的做法！
    with open(event.src_path, 'a') as f:
        f.write("This is a new line.\n") # 这会再次触发 on_modified 事件！

解决方案：

• 检查文件名：如果你的脚本会生成一个特定的文件（.processed），可以在事件处理函数中检查文件名,避免处理它。
• 使用锁：对于更复杂的场景,可以使用文件锁或线程锁来确保同一时间只有一个事件处理程序在运行。
• 添加时间戳检查：记录上次处理的时间，如果两次事件间隔太短（比如小于1秒）,则忽略它。

3. 性能考虑

• 监控的文件数量：监控成千上万个文件可能会消耗较多资源,尽量精确地指定要监控的目录。
• 事件处理速度：事件处理函数中的任务应该尽量轻量，如果需要执行耗时操作（如网络请求、文件拷贝），考虑将其放入一个线程池或异步任务队列中处理,以避免阻塞监控线程。

watchdog 是一个功能强大且易于使用的文件系统监控库,它的基本工作流程可以概括为：

1. 创建一个自定义的 EventHandler，继承 FileSystemEventHandler 并重写你感兴趣的事件方法（如 on_modified）。
2. 创建一个 Observer 实例。
3. 使用 observer.schedule() 将你的事件处理器和要监控的路径注册到观察者中。
4. 调用 observer.start() 启动监听。
5. （可选）让你的主程序保持运行（例如使用 while True 循环），并在程序退出时调用 observer.stop() 和 observer.join()。

通过掌握这些核心概念和示例，你就可以轻松地将 watchdog 集成到你的项目中，实现诸如自动同步、日志分析、开发工具热重载等多种功能。