watchdog - Python文件系统监控库

watchdog是一个Python文件系统监控库，用于监控文件系统事件（创建、修改、删除、移动），支持跨平台实时响应文件变化

分类: thirdparty 难度: 中级 更新: 2025-01-21

文件监控 系统监控 事件处理 跨平台 实时监控 自动化

watchdog - Python文件系统监控库

📝 概述

watchdog 是一个 Python 库，用于监控文件系统事件（如文件创建、修改、删除、移动等）。它跨平台支持（Windows、Linux、macOS），广泛用于需要实时响应文件变化的场景，例如文件同步工具、自动构建系统等。watchdog 提供了事件驱动的文件系统监控机制，能够高效地处理大量文件变化事件。

🎯 学习目标

• 掌握 watchdog 库的安装和基本配置
• 理解文件系统事件监控的核心概念
• 学会使用 Observer 和 EventHandler 进行事件监控
• 掌握不同类型的文件系统事件处理
• 了解跨平台监控的差异和注意事项
• 学习高级功能：异步处理、轮询模式、命令行工具
• 掌握实际应用场景的开发技巧

📋 前置知识

• Python 基础语法和面向对象编程
• 文件系统基本概念（文件、目录、路径）
• 事件驱动编程基础
• 多线程和异步编程概念（高级功能需要）

🔍 详细内容

安装

通过 pip 安装 watchdog：

pip install watchdog

如果需要支持 macOS 的 fsevents（更高效的文件监控），可以安装带额外依赖的版本：

pip install "watchdog[watchmedo]"

依赖要求：Python 3.6+

核心组件

watchdog 主要由以下模块和类组成：

1. watchdog.observers 模块

• Observer：文件系统事件观察者，负责启动和停止监控
• PollingObserver：轮询模式的观察者，适用于不支持原生事件通知的系统

2. watchdog.events 模块

• FileSystemEventHandler：事件处理基类，用于定义事件回调
• 事件类型：
  • FileCreatedEvent：文件创建
  • FileModifiedEvent：文件修改
  • FileDeletedEvent：文件删除
  • FileMovedEvent：文件移动或重命名

3. watchdog.utils 模块

• 提供实用工具，如路径规范化

组件	类型	功能	说明
Observer	观察者	原生监控	高性能，依赖系统API
PollingObserver	观察者	轮询监控	兼容性好，CPU占用高
FileSystemEventHandler	处理器	事件处理基类	需要继承并重写方法

💡 实际应用

基本用法

使用 watchdog 监控文件系统的基本步骤：

1. 定义事件处理器：继承 FileSystemEventHandler 并重写方法来处理事件
2. 创建观察者：使用 Observer 或 PollingObserver，指定监控目录和事件处理器
3. 启动监控：调用 schedule 和 start 方法开始监控
4. 停止监控：调用 stop 和 join 方法停止观察者

简单示例

以下是一个监控指定目录的基本示例：

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

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        """文件或目录创建时触发"""
        print(f"Created: {event.src_path}")

    def on_modified(self, event):
        """文件或目录修改时触发"""
        print(f"Modified: {event.src_path}")

    def on_deleted(self, event):
        """文件或目录删除时触发"""
        print(f"Deleted: {event.src_path}")

    def on_moved(self, event):
        """文件或目录移动/重命名时触发"""
        print(f"Moved: {event.src_path} to {event.dest_path}")

if __name__ == "__main__":
    # 指定监控目录
    path = "./watch_dir"  # 替换为你的目录
    event_handler = MyHandler()
    
    # 创建观察者
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)  # recursive=True 表示监控子目录
    
    # 启动观察者
    observer.start()
    
    try:
        # 主线程保持运行
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 停止观察者
        observer.stop()
        observer.join()

运行步骤

1. 创建一个目录（如 watch_dir）
2. 运行脚本
3. 在 watch_dir 中创建、修改、删除文件或子目录，观察终端输出

输出示例

Created: ./watch_dir/test.txt
Modified: ./watch_dir/test.txt
Moved: ./watch_dir/test.txt to ./watch_dir/test_new.txt
Deleted: ./watch_dir/test_new.txt

事件处理器方法

FileSystemEventHandler 支持以下方法，可根据需要重写：

方法	触发条件	说明
on_created(event)	文件或目录创建	新建文件/目录时调用
on_modified(event)	文件或目录修改	内容或元数据变化时调用
on_deleted(event)	文件或目录删除	删除文件/目录时调用
on_moved(event)	文件或目录移动/重命名	移动或重命名时调用
on_any_event(event)	捕获所有事件	优先级低于具体事件方法

事件对象属性

属性	类型	说明
event.src_path	str	事件源路径
event.dest_path	str	目标路径（仅 on_moved 有）
event.is_directory	bool	是否为目录
event.event_type	str	事件类型字符串（如 "created"、"modified"）

高级用法

使用 PollingObserver

如果系统不支持原生事件通知（某些网络文件系统），可以使用轮询模式：

from watchdog.observers.polling import PollingObserver

observer = PollingObserver()
observer.schedule(event_handler, path, recursive=True)
observer.start()

监控多个目录

paths = ["./dir1", "./dir2"]
observer = Observer()
for path in paths:
    observer.schedule(event_handler, path, recursive=True)
observer.start()

结合异步操作

与 asyncio 集成，异步处理事件：

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

class AsyncHandler(FileSystemEventHandler):
    async def on_modified(self, event):
        print(f"Modified: {event.src_path}")
        await asyncio.sleep(1)  # 模拟异步操作
        print(f"Processed: {event.src_path}")

async def run_observer():
    path = "./watch_dir"
    event_handler = AsyncHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    asyncio.run(run_observer())

实用工具：watchmedo

watchdog 提供命令行工具 watchmedo，可快速监控并执行命令。

安装

pip install "watchdog[watchmedo]"

示例：监控并运行脚本

watchmedo shell-command \
    --patterns="*.py;*.txt" \
    --recursive \
    --command='echo "${watch_src_path} was changed"' \
    ./watch_dir

输出

./watch_dir/test.py was changed

实际案例

文件同步工具

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

class SyncHandler(FileSystemEventHandler):
    def __init__(self, source_dir, target_dir):
        self.source_dir = source_dir
        self.target_dir = target_dir

    def on_created(self, event):
        if not event.is_directory:
            src = event.src_path
            dst = src.replace(self.source_dir, self.target_dir)
            os.makedirs(os.path.dirname(dst), exist_ok=True)
            shutil.copy2(src, dst)
            print(f"Copied {src} to {dst}")

    def on_modified(self, event):
        if not event.is_directory:
            src = event.src_path
            dst = src.replace(self.source_dir, self.target_dir)
            shutil.copy2(src, dst)
            print(f"Updated {src} to {dst}")

    def on_deleted(self, event):
        if not event.is_directory:
            dst = event.src_path.replace(self.source_dir, self.target_dir)
            if os.path.exists(dst):
                os.remove(dst)
                print(f"Deleted {dst}")

if __name__ == "__main__":
    source_dir = "./source"
    target_dir = "./target"
    os.makedirs(source_dir, exist_ok=True)
    os.makedirs(target_dir, exist_ok=True)

    event_handler = SyncHandler(source_dir, target_dir)
    observer = Observer()
    observer.schedule(event_handler, source_dir, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        observer.join()

功能：

• 监控 source 目录，将文件同步到 target 目录
• 支持创建、修改、删除操作

⚠️ 注意事项

性能考虑

• 原生观察者（Observer）依赖系统事件通知，性能较高
• 轮询观察者（PollingObserver）会增加 CPU 使用，适合特殊场景

跨平台差异

操作系统	底层API	特点
Windows	ReadDirectoryChangesW	高性能，支持递归监控
macOS	FSEvents	需安装 watchdog[watchmedo]
Linux	inotify	原生支持，性能优秀

事件重复

• 某些操作（如保存文件）可能触发多次 on_modified，需在处理器中去重
• 建议使用时间戳或文件哈希进行去重判断

异常处理

• 在事件处理中添加 try/except，避免观察者崩溃
• 监控大量文件时，注意系统资源限制

class SafeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        try:
            # 处理文件修改事件
            self.process_file(event.src_path)
        except Exception as e:
            print(f"处理文件 {event.src_path} 时出错: {e}")

🔗 相关内容

• os 模块 - 操作系统接口
• pathlib 模块 - 面向对象的路径处理
• time 模块 - 时间处理
• logging 模块 - 日志记录
• psutil - 系统监控与进程管理库

📚 扩展阅读

• watchdog 官方文档
• 文件系统监控最佳实践
• 跨平台文件监控技术对比

🏷️ 标签

文件监控 系统监控 事件处理 跨平台 实时监控 自动化

---

最后更新: 2025-01-21
作者: Python技术文档工程师
版本: 1.0

讨论与反馈

欢迎在下方留言讨论，分享你的学习心得或提出问题。评论基于GitHub Issues，需要GitHub账号。