Python：watchdog 库高级用法举例和应用详解

from watchdog.observers import Observer  # 导入观察者类
from watchdog.events import FileSystemEventHandler  # 导入事件处理类
import time  # 导入时间模块

class MyHandler(FileSystemEventHandler):
    # 定义文件系统事件处理类，继承FileSystemEventHandler
    def on_modified(self, event):
        # 当文件被修改时触发该方法
        print(f'文件被修改: {event.src_path}')  # 打印修改的文件路径

    def on_created(self, event):
        # 当文件被创建时触发该方法
        print(f'文件创建: {event.src_path}')  # 打印创建的文件路径

event_handler = MyHandler()  # 创建事件处理实例
observer = Observer()  # 创建观察者实例
observer.schedule(event_handler, path='监控目录路径', recursive=True)  # 设置监控目录
observer.start()  # 开始监控

try:
    while True:  # 持续运行
        time.sleep(1)  # 每隔1秒检查一次
except KeyboardInterrupt:
    observer.stop()  # 当收到中断信号时停止监控
observer.join()  # 等待观察者线程结束

import logging  # 导入日志模块
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 配置日志输出
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')  # 设置日志格式

class LogHandler(FileSystemEventHandler):
    # 定义记录日志的事件处理类
    def on_modified(self, event):
        logging.info(f'文件被修改: {event.src_path}')  # 记录文件修改事件

    def on_deleted(self, event):
        logging.info(f'文件被删除: {event.src_path}')  # 记录文件删除事件

event_handler = LogHandler()  # 创建日志处理实例
observer = Observer()
observer.schedule(event_handler, path='监控目录路径', recursive=True)  # 监控指定目录
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()  # 停止监控
observer.join()

import shutil  # 导入shutil模块用于文件操作
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import os

class BackupHandler(FileSystemEventHandler):
    # 定义备份处理类
    def on_modified(self, event):
        if not event.is_directory:  # 仅处理文件事件，不处理目录
            src_file = event.src_path  # 源文件路径
            backup_file = f'备份目录/{os.path.basename(src_file)}'  # 定义备份文件路径
            shutil.copy(src_file, backup_file)  # 复制文件到备份目录
            print(f'已备份: {src_file} 至 {backup_file}')  # 打印备份信息

event_handler = BackupHandler()  # 创建备份处理类实例
observer = Observer()
observer.schedule(event_handler, path='监控目录路径', recursive=False)  # 监控指定目录
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()  # 停止监控
observer.join()