需要监控文件夹的新增、修改、删除操作？想让脚本自动响应文件变化？今天教你用Python写一个专业的文件夹监控工具，支持实时监控、事件响应、日志记录等功能！

实战场景

• 监控下载文件夹自动处理新文件
• 备份系统实时同步文件变化
• 开发环境自动刷新浏览器
• 监控文件夹安全告警

核心实现

准备工作

pip install watchdog schedule

文件夹监控器

import os
import time
import hashlib
import json
from pathlib import Path
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileSystemEvent
from typing import Callable, Dict, List, Optional
import threading
import schedule

class FileMonitorHandler(FileSystemEventHandler):
    """文件监控系统处理类"""
    
    def __init__(self, monitor):
        """
        初始化处理类
        
        Args:
            monitor: FileMonitor实例
        """
        super().__init__()
        self.monitor = monitor
    
    def on_created(self, event: FileSystemEvent):
        """文件/文件夹创建时触发"""
        if not event.is_directory:
            self.monitor.handle_event('created', event.src_path)
    
    def on_deleted(self, event: FileSystemEvent):
        """文件/文件夹删除时触发"""
        if not event.is_directory:
            self.monitor.handle_event('deleted', event.src_path)
    
    def on_modified(self, event: FileSystemEvent):
        """文件/文件夹修改时触发"""
        if not event.is_directory:
            self.monitor.handle_event('modified', event.src_path)
    
    def on_moved(self, event: FileSystemEvent):
        """文件/文件夹移动时触发"""
        if not event.is_directory:
            self.monitor.handle_event('moved', event.src_path, event.dest_path)


class FileMonitor:
    """文件夹监控器"""
    
    def __init__(self, watch_path: str, 
                 extensions: List[str] = None,
                 ignore_patterns: List[str] = None):
        """
        初始化监控器
        
        Args:
            watch_path: 监控路径
            extensions: 监控的文件扩展名，如 ['.txt', '.pdf']
            ignore_patterns: 忽略的模式，如 ['*.tmp', '__pycache__']
        """
        self.watch_path = watch_path
        self.extensions = extensions or []
        self.ignore_patterns = ignore_patterns or []
        
        self.observer = None
        self.handler = FileMonitorHandler(self)
        self.event_callbacks: Dict[str, List[Callable]] = {
            'created': [],
            'deleted': [],
            'modified': [],
            'moved': []
        }
        
        # 事件统计
        self.stats = {
            'created': 0,
            'deleted': 0,
            'modified': 0,
            'moved': 0
        }
        
        # 事件历史
        self.history: List[Dict] = []
        
        # 文件哈希缓存
        self.file_hashes: Dict[str, str] = {}
        
        # 锁
        self.lock = threading.Lock()
    
    def _should_process(self, file_path: str) -> bool:
        """判断文件是否应该处理"""
        path = Path(file_path)
        
        # 检查扩展名
        if self.extensions:
            if path.suffix.lower() not in [ext.lower() for ext in self.extensions]:
                return False
        
        # 检查忽略模式
        for pattern in self.ignore_patterns:
            if pattern.startswith('*'):
                if path.suffix.lower() == pattern[1:].lower():
                    return False
            elif pattern in str(path):
                return False
        
        return True
    
    def _get_file_hash(self, file_path: str) -> Optional[str]:
        """获取文件MD5哈希"""
        try:
            if not os.path.exists(file_path):
                return None
            
            md5 = hashlib.md5()
            with open(file_path, 'rb') as f:
                for chunk in iter(lambda: f.read(4096), b''):
                    md5.update(chunk)
            return md5.hexdigest()
        except:
            return None
    
    def handle_event(self, event_type: str, src_path: str, dest_path: str = None):
        """处理文件事件"""
        if not self._should_process(src_path):
            return
        
        # 创建事件记录
        event = {
            'type': event_type,
            'src_path': src_path,
            'dest_path': dest_path,
            'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'size': None
        }
        
        # 获取文件大小
        if event_type != 'deleted' and os.path.exists(src_path):
            try:
                event['size'] = os.path.getsize(src_path)
            except:
                pass
        
        # 记录事件
        with self.lock:
            self.history.append(event)
            self.stats[event_type] += 1
            
            # 保持历史记录在1000条以内
            if len(self.history) > 1000:
                self.history = self.history[-1000:]
        
        # 打印日志
        print(f"[{event['timestamp']}] {event_type.upper()}: {src_path}")
        if dest_path:
            print(f"   -> {dest_path}")
        
        # 调用回调函数
        callbacks = self.event_callbacks.get(event_type, [])
        for callback in callbacks:
            try:
                callback(event)
            except Exception as e:
                print(f"✗ 回调执行失败: {e}")
    
    def on_created(self, callback: Callable):
        """注册创建事件回调"""
        self.event_callbacks['created'].append(callback)
        return callback
    
    def on_deleted(self, callback: Callable):
        """注册删除事件回调"""
        self.event_callbacks['deleted'].append(callback)
        return callback
    
    def on_modified(self, callback: Callable):
        """注册修改事件回调"""
        self.event_callbacks['modified'].append(callback)
        return callback
    
    def on_moved(self, callback: Callable):
        """注册移动事件回调"""
        self.event_callbacks['moved'].append(callback)
        return callback
    
    def start(self):
        """启动监控"""
        if not os.path.exists(self.watch_path):
            raise FileNotFoundError(f"监控路径不存在: {self.watch_path}")
        
        self.observer = Observer()
        self.observer.schedule(self.handler, self.watch_path, recursive=True)
        self.observer.start()
        
        print(f"✓ 监控已启动: {self.watch_path}")
        print(f"  监控扩展名: {self.extensions or '所有文件'}")
        print(f"  忽略模式: {self.ignore_patterns or '无'}")
    
    def stop(self):
        """停止监控"""
        if self.observer:
            self.observer.stop()
            self.observer.join()
            print("✓ 监控已停止")
    
    def run(self, duration: int = None):
        """
        运行监控
        
        Args:
            duration: 运行时长（秒），None表示永久运行
        """
        self.start()
        
        try:
            if duration:
                print(f"⏱ 运行 {duration} 秒...")
                time.sleep(duration)
            else:
                print("🔄 持续监控中，按 Ctrl+C 停止")
                while True:
                    time.sleep(1)
        except KeyboardInterrupt:
            print("\n")
        finally:
            self.stop()
    
    def get_stats(self) -> Dict:
        """获取统计信息"""
        with self.lock:
            return dict(self.stats)
    
    def get_history(self, limit: int = 100) -> List[Dict]:
        """获取事件历史"""
        with self.lock:
            return self.history[-limit:]
    
    def clear_history(self):
        """清空历史记录"""
        with self.lock:
            self.history = []
            print("✓ 历史记录已清空")


class ScheduledTaskManager:
    """定时任务管理器（配合监控使用）"""
    
    def __init__(self):
        self.tasks: Dict[str, Callable] = {}
    
    def add_task(self, name: str, task: Callable, interval_seconds: int = 60):
        """
        添加定时任务
        
        Args:
            name: 任务名称
            task: 任务函数
            interval_seconds: 执行间隔（秒）
        """
        schedule.every(interval_seconds).seconds.do(task)
        self.tasks[name] = task
        print(f"✓ 定时任务已添加: {name} (每{interval_seconds}秒)")
    
    def run_continuously(self):
        """持续运行定时任务"""
        print("⏰ 定时任务运行中，按 Ctrl+C 停止")
        while True:
            schedule.run_pending()
            time.sleep(1)


# 使用示例
if __name__ == "__main__":
    # 示例1: 基础监控
    monitor = FileMonitor(
        watch_path="./watch_folder",
        extensions=['.txt', '.pdf', '.docx']
    )
    
    # 示例2: 注册回调函数
    @monitor.on_created
    def handle_new_file(event):
        print(f"📄 新文件: {event['src_path']}")
        print(f"   大小: {event['size']} bytes")
    
    @monitor.on_modified
    def handle_modified(event):
        print(f"📝 修改文件: {event['src_path']}")
    
    @monitor.on_deleted
    def handle_deleted(event):
        print(f"🗑 删除文件: {event['src_path']}")
    
    # 示例3: 自动处理新文件
    @monitor.on_created
    def auto_process(event):
        """新文件自动处理示例"""
        file_path = event['src_path']
        ext = Path(file_path).suffix.lower()
        
        if ext == '.pdf':
            # PDF处理逻辑
            print(f"   正在处理PDF: {file_path}")
        
        elif ext == '.txt':
            # 文本处理逻辑
            print(f"   正在处理文本: {file_path}")
    
    # 示例4: 自动备份
    backup_monitor = FileMonitor(
        watch_path="./important_files",
        extensions=['.py', '.json', '.md']
    )
    
    backup_files = []
    
    @backup_monitor.on_modified
    def backup_modified(event):
        """备份修改的文件"""
        src = event['src_path']
        backup_path = f"./backup/{Path(src).name}.bak"
        
        try:
            import shutil
            shutil.copy2(src, backup_path)
            print(f"✓ 已备份: {backup_path}")
        except Exception as e:
            print(f"✗ 备份失败: {e}")
    
    # 示例5: 日志记录
    log_file = "./monitor_log.json"
    
    @monitor.on_created
    @monitor.on_modified
    @monitor.on_deleted
    def log_event(event):
        """记录所有事件到日志"""
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(event, ensure_ascii=False) + '\n')
    
    # 示例6: 统计分析
    @monitor.on_created
    def count_created(event):
        stats = monitor.get_stats()
        print(f"📊 统计: 创建 {stats['created']}, 修改 {stats['modified']}")
    
    # 示例7: 运行监控
    monitor.run(duration=60)  # 运行60秒
    
    # 示例8: 配合定时任务
    task_manager = ScheduledTaskManager()
    
    @task_manager.add_task("统计报告", lambda: print("生成统计报告..."), 300)
    def generate_report():
        stats = monitor.get_stats()
        print("\n========== 监控统计 ==========")
        print(f"创建: {stats['created']}")
        print(f"修改: {stats['modified']}")
        print(f"删除: {stats['deleted']}")
        print(f"移动: {stats['moved']}")
        print("===============================\n")
    
    # 示例9: 显示历史
    history = monitor.get_history(10)
    for event in history:
        print(f"{event['timestamp']} {event['type']}: {event['src_path']}")
    
    # 示例10: 命令行工具
    import sys
    
    def main():
        if len(sys.argv) < 2:
            print("""
╔════════════════════════════════════════╗
║       文件夹监控工具                     ║
╠════════════════════════════════════════╣
║  python monitor.py start <路径>         ║
║  python monitor.py status               ║
║  python monitor.py history              ║
║  python monitor.py stop                ║
╚════════════════════════════════════════╝
            """)
            return
        
        cmd = sys.argv[1]
        
        if cmd == "start":
            path = sys.argv[2] if len(sys.argv) > 2 else "./"
            m = FileMonitor(path)
            m.run()
        
        elif cmd == "status":
            stats = monitor.get_stats()
            print(f"创建: {stats['created']}")
            print(f"修改: {stats['modified']}")
            print(f"删除: {stats['deleted']}")
        
        elif cmd == "history":
            for event in monitor.get_history():
                print(f"{event['timestamp']} {event['type']}: {event['src_path']}")
        
        elif cmd == "clear":
            monitor.clear_history()
    
    if __name__ == "__main__":
        main()

进阶技巧

1. 递归监控：recursive=True可监控子文件夹

2. 性能优化：大文件夹可使用节流（throttle）

3. 多目录监控：创建多个Monitor实例

4. 事件去重：短时间内同一文件的多个修改合并处理

总结

这个文件夹监控工具可以帮你：

• ✅ 实时监控文件变化
• ✅ 灵活的事件回调
• ✅ 自动处理新文件
• ✅ 事件统计和历史记录

代码已经非常完善，直接复制使用即可！🙃