如何实现文件移入文件夹后立即自动处理？Python Watchdog/OS模块实践

你只需要自定义事件处理器替代默认的LoggingEventHandler即可，watchdog触发的所有事件对象都自带src_path属性，直接就能拿到监测到的文件完整路径，基于这个路径就能做读取、处理、移动等所有操作。

完整代码示例 #

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

# 自定义事件处理器
class CustomFileHandler(FileSystemEventHandler):
    # 重写文件创建触发的方法
    def on_created(self, event):
        # 跳过文件夹变更事件，仅处理文件
        if event.is_directory:
            return
        
        file_full_path = event.src_path
        file_name = os.path.basename(file_full_path)
        
        # 过滤无效文件：临时文件、隐藏文件等
        if file_name == ".ipynb_checkpoints" or file_name.startswith("~$") or file_name.startswith("."):
            return
        
        # 匹配你需要处理的文件后缀
        file_suffix = os.path.splitext(file_name)[-1].lower()
        target_suffix = [".xlsx", ".xls", ".xlsm", ".zip"]
        if file_suffix not in target_suffix:
            return
        
        # 避免处理未完全传输/拷贝的文件，等待文件解除占用
        while True:
            try:
                with open(file_full_path, "rb") as f:
                    break
            except PermissionError:
                time.sleep(0.5)
        
        # ---------------------- 以下写你的文件处理逻辑 ----------------------
        print(f"开始处理文件：{file_full_path}")
        # 示例：读取Excel内容
        # import pandas as pd
        # df = pd.read_excel(file_full_path)
        # 你的业务处理逻辑...
        
        # 处理完成后移动文件
        target_dir = "你要移动到的目标文件夹绝对路径"
        os.makedirs(target_dir, exist_ok=True)
        shutil.move(file_full_path, os.path.join(target_dir, file_name))
        print(f"文件处理完成，已移动至：{target_dir}")

if __name__ == "__main__":
    monitor_path = 'G:\\Actuarial\\Pricing\\IBOTT\\Team\\NP\\Projects\\Claims Automation\\Processed Files'
    event_handler = CustomFileHandler()
    observer = Observer()
    observer.schedule(event_handler, monitor_path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

补充说明 #

• 如果需要监测被移动到目标文件夹的文件，可以额外重写on_moved方法，通过event.dest_path获取移动后的文件路径，处理逻辑和on_created保持一致即可
• 等待文件解除占用的逻辑可以根据你实际的文件大小调整等待间隔，避免大文件拷贝未完成就触发处理报错

内容的提问来源于stack exchange，提问作者Nhyi