Python 文件监控利器Watchdog的使用方法

用户11081884

发布于 2026-07-20 17:34:01

510

Python Watchdog库是一个强大的文件系统监控工具，能够实时检测文件和目录的创建、修改、删除等操作，并触发自定义事件处理逻辑。它跨平台、高效可靠，适用于自动化备份、日志监控、文件同步等实际应用场景。本文将详细介绍Watchdog的使用方法、应用场景，并提供实用的代码示例。

Watchdog是一个跨平台的文件系统监控库，支持Linux、Windows和MacOS等多种操作系统。它采用高效的文件系统监控机制，性能优越且可靠性高，是Python生态中文件监控的首选工具。

主要特点：

实时监控文件和目录变化
支持多种事件类型（创建、修改、删除、移动）
可自定义事件处理逻辑
支持递归监控子目录
能够同时监控多个路径

安装Watchdog

安装Watchdog非常简单，只需使用pip命令：

pip install watchdog

使用方法

1. 基础监控示例

Watchdog的核心组件包括Observer和FileSystemEventHandler。Observer负责监控文件系统变化，FileSystemEventHandler则定义了事件发生时的处理逻辑。下面是一个最基本的文件监控示例，监控当前目录及其子目录中的所有文件变化：

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

class FileChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f'文件被修改: {event.src_path}')
    
    def on_created(self, event):
        print(f'文件被创建: {event.src_path}')
    
    def on_deleted(self, event):
        print(f'文件被删除: {event.src_path}')
    
    def on_moved(self, event):
        print(f'文件被移动: 从 {event.src_path} 到 {event.dest_path}')

def start_monitoring(path='.'):
    event_handler = FileChangeHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    start_monitoring('/path/to/monitor')

2. 监控特定文件类型

我们可以扩展处理器，只监控特定类型的文件，例如只关注.py和.txt文件：

class SpecificFileHandler(FileSystemEventHandler):
    def on_any_event(self, event):
        if event.is_directory:
            return
            
        if event.src_path.endswith(('.py', '.txt')):
            if event.event_type == 'modified':
                print(f"Python或文本文件被修改: {event.src_path}")
            elif event.event_type == 'created':
                print(f"新的Python或文本文件: {event.src_path}")
            elif event.event_type == 'deleted':
                print(f"Python或文本文件被删除: {event.src_path}")

应用场景

1. 自动化测试触发

在开发过程中，当测试文件或源代码发生变化时，自动运行测试：

import subprocess

class TestRunnerHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.py') and not event.is_directory:
            print(f"检测到变更: {event.src_path}")
            print("运行测试...")
            subprocess.run(['pytest'], check=True)

2. 实时日志分析

监控日志文件变化并实时分析错误：

class LogMonitorHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.log') and not event.is_directory:
            with open(event.src_path, 'r') as f:
                new_lines = f.readlines()[-10:]  # 读取最后10行
                for line in new_lines:
                    if 'ERROR' in line:
                        print(f"发现错误日志: {line.strip()}")

3. 自动备份系统

当重要文件发生变化时自动创建备份：

import shutil
from datetime import datetime

class AutoBackupHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory and event.src_path.endswith('.docx'):
            backup_dir = "backups"
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_file = f"{backup_dir}/{event.src_path}_{timestamp}.bak"
            shutil.copy2(event.src_path, backup_file)
            print(f"已创建备份: {backup_file}")

4. 图片自动处理

监控图片上传目录并自动处理：

from PIL import Image
import os

class ImageProcessorHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory and event.src_path.lower().endswith(('.png', '.jpg', '.jpeg')):
            print(f"新图片上传: {event.src_path}")
            try:
                img = Image.open(event.src_path)
                # 创建缩略图
                img.thumbnail((200, 200))
                thumb_path = os.path.join('thumbs', os.path.basename(event.src_path))
                img.save(thumb_path)
                print(f"缩略图已生成: {thumb_path}")
            except Exception as e:
                print(f"图片处理失败: {e}")

5. 多目录协同监控

同时监控多个目录并协同工作：

class MultiPathHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_events = {}
    
    def on_modified(self, event):
        if not event.is_directory:
            current_time = time.time()
            # 防止重复触发
            if event.src_path not in self.last_events or current_time - self.last_events[event.src_path] > 1:
                print(f"文件修改于 {event.src_path}")
                self.last_events[event.src_path] = current_time

if __name__ == "__main__":
    paths = ['/path/to/watch1', '/path/to/watch2', '/path/to/watch3']
    event_handler = MultiPathHandler()
    observer = Observer()
    
    for path in paths:
        if os.path.exists(path):
            observer.schedule(event_handler, path, recursive=True)
    
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

性能优化

1、事件去重：某些操作系统可能会为单个文件更改触发多个事件

class DebounceHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_event_time = {}
    
    def on_modified(self, event):
        current_time = time.time()
        if event.src_path in self.last_event_time:
            if current_time - self.last_event_time[event.src_path] < 1.0:  # 1秒内不重复处理
                return
        self.last_event_time[event.src_path] = current_time
        print(f"处理修改事件: {event.src_path}")

2、批量处理：对于高频事件，可以采用批量处理策略

from collections import defaultdict

class BatchHandler(FileSystemEventHandler):
    def __init__(self):
        self.pending_events = defaultdict(list)
        self.last_process_time = time.time()
    
    def on_modified(self, event):
        self.pending_events['modified'].append(event.src_path)
        self._process_if_needed()
    
    def _process_if_needed(self):
        current_time = time.time()
        if current_time - self.last_process_time > 5.0 or len(self.pending_events['modified']) > 10:
            self._process_events()
            self.last_process_time = current_time
    
    def _process_events(self):
        if self.pending_events['modified']:
            print(f"批量处理修改文件: {', '.join(self.pending_events['modified'])}")
            self.pending_events['modified'].clear()

3. 实时文件同步

import shutil
import os

class SyncHandler(FileSystemEventHandler):
    def __init__(self, dest_dir):
        self.dest_dir = dest_dir
        os.makedirs(dest_dir, exist_ok=True)
    
    def on_modified(self, event):
        if not event.is_directory:
            dest_path = os.path.join(self.dest_dir, os.path.basename(event.src_path))
            shutil.copy2(event.src_path, dest_path)
            print(f'同步文件: {event.src_path} -> {dest_path}')
    
    def on_created(self, event):
        if not event.is_directory:
            dest_path = os.path.join(self.dest_dir, os.path.basename(event.src_path))
            shutil.copy2(event.src_path, dest_path)
            print(f'同步新文件: {event.src_path} -> {dest_path}')

4. 自动化测试触发

class TestTriggerHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.py'):
            print(f'Python文件修改: {event.src_path}')
            # 在这里添加运行测试的逻辑
            # os.system('pytest')

5、多线程处理：将耗时操作放到单独线程中

from threading import Thread
from queue import Queue

class BatchHandler(FileSystemEventHandler):
    def __init__(self):
        self.queue = Queue()
        self.worker = Thread(target=self.process_queue)
        self.worker.daemon = True
        self.worker.start()
    
    def process_queue(self):
        while True:
            item = self.queue.get()
            # 处理事件
            print(f'处理事件: {item}')
            self.queue.task_done()
    
    def on_modified(self, event):
        self.queue.put(event)

Watchdog库为Python开发者提供了强大而灵活的文件系统监控能力，适用于各种自动化场景。无论是开发环境中的自动化测试、生产环境中的日志监控，还是日常工作中的文件同步和备份，Watchdog都能提供可靠的解决方案。

关键优势：