在 Python 中，watchdog 通常指的是一个用于 监控文件系统变化 的第三方库。它可以监听文件或目录的创建、修改、删除等事件，并触发相应的回调函数。

下面是 watchdog 的核心概念和用法：

1️⃣ 安装

pip install watchdog

2️⃣ 基本使用

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

# 自定义事件处理器
class MyHandler(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}")

# 设置监控目录
path = "./test_folder"
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()

3️⃣ 核心类说明

4️⃣ 小提示

• recursive=True 可以监控子目录，否则只监控指定目录。
• event.src_path 可以获取触发事件的文件或目录路径。
• 可以结合 多线程 或 异步处理 做更复杂的自动化任务。

自动处理新文件的 Watchdog 示例

这个示例会实现：

• 监控一个目录（例如 ./watch_folder）
• 当有新文件被创建时自动移动到另一个目录（例如 ./processed）
• 并打印日志

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

# 1️⃣ 自定义事件处理器
class FileHandler(FileSystemEventHandler):
    def __init__(self, dest_folder):
        self.dest_folder = dest_folder
        os.makedirs(dest_folder, exist_ok=True)

    def on_created(self, event):
        # 忽略目录，只处理文件
        if event.is_directory:
            return
        src_path = event.src_path
        filename = os.path.basename(src_path)
        dest_path = os.path.join(self.dest_folder, filename)

        # 移动文件到目标文件夹
        try:
            shutil.move(src_path, dest_path)
            print(f"[INFO] 文件已处理: {filename} -> {self.dest_folder}")
        except Exception as e:
            print(f"[ERROR] 移动文件失败: {filename}, {e}")

# 2️⃣ 配置监控目录
watch_folder = "./watch_folder"
processed_folder = "./processed"

# 3️⃣ 创建 Observer 并绑定事件
event_handler = FileHandler(processed_folder)
observer = Observer()
observer.schedule(event_handler, watch_folder, recursive=False)

# 4️⃣ 启动监控
observer.start()
print(f"[INFO] 正在监控目录: {watch_folder}")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

✅ 功能说明

1. 自动创建目录：如果目标目录不存在，会自动创建。
2. 只处理文件：不会对目录触发事件。
3. 异常处理：移动文件失败会打印错误日志。
4. 可扩展：你可以在 on_created 中加处理逻辑，比如读取 CSV、生成报告、上传到云端等。