You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python使用watchdog监听器检测文件事件后自动复制文件

需求说明

现有基于watchdog编写的目录Excel文件监听代码可正常运行,需要新增以下逻辑:

  • 监听路径下检测到新增.xlsx格式文件时,自动将文件复制到名为today的文件夹中
  • 复制后的文件保留原始文件名与xlsx扩展名
  • 作为OOP初学者,需要明确新文件创建事件触发后的文件复制实现方式
原有代码(含1处笔误)

原有代码中KeyboardInterrupt分支下的ob.stop()存在变量名拼写错误,实际观察者实例名为obs,运行时触发中断会抛出NameError,修改时会一并修复:

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

class FileWatcher(PatternMatchingEventHandler):
    patterns = ["*.xlsx"] 

    def process(self, event):
        
       # event.src_path will be the full file path
       # event.event_type will be 'created', 'moved', etc.
       print('{} observed on {}'.format(event.event_type, event.src_path))

    def on_created(self, event):
        self.process(event)

if __name__ == '__main__':
    obs = Observer() 
    obs.schedule(FileWatcher(), path='path/')
    print("Monitoring started....")
    obs.start() # Start watching

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ob.stop() # 此处变量名拼写错误

    obs.join()
实现步骤

不需要额外安装第三方依赖,直接使用Python标准库即可完成复制逻辑:

  • 导入os模块处理路径拼接、目录存在性判断,导入shutil模块实现文件复制
  • 启动监听前先检查目标today文件夹是否存在,不存在则自动创建,避免复制时抛出路径不存在错误
  • 在事件处理方法中首先过滤目录创建事件(watchdog的on_created会同时响应文件和目录创建),仅处理xlsx文件
  • 拿到新增文件的原始文件名,拼接today文件夹路径作为复制目标路径,调用shutil.copy2完成复制(该方法会保留文件原始创建时间、修改时间等元数据,适合办公文档场景)
  • 修复原有代码的变量名笔误
修改后完整可运行代码
from watchdog.observers import Observer  
from watchdog.events import PatternMatchingEventHandler
import time
import os
import shutil

class FileWatcher(PatternMatchingEventHandler):
    patterns = ["*.xlsx"] 

    def __init__(self, target_dir):
        super().__init__()
        self.target_dir = target_dir

    def process(self, event):
        # 过滤目录创建事件,仅处理文件
        if event.is_directory:
            return
        print('{} observed on {}'.format(event.event_type, event.src_path))
        # 提取原始文件名
        file_name = os.path.basename(event.src_path)
        # 拼接目标文件路径
        target_path = os.path.join(self.target_dir, file_name)
        # 执行文件复制
        shutil.copy2(event.src_path, target_path)
        print(f'File copied to: {target_path}')

    def on_created(self, event):
        self.process(event)

if __name__ == '__main__':
    # 配置监听路径,替换为你的实际监听目录
    watch_path = 'path/'
    # 配置目标today文件夹路径
    today_dir = os.path.join(watch_path, 'today')
    # 若today文件夹不存在则自动创建
    os.makedirs(today_dir, exist_ok=True)

    obs = Observer() 
    obs.schedule(FileWatcher(target_dir=today_dir), path=watch_path)
    print("Monitoring started....")
    obs.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        obs.stop() # 修复原有变量名笔误

    obs.join()

注意:如果业务场景下存在大文件缓慢写入的情况(比如其他程序往监听目录传输大体积Excel,文件还没写完就触发created事件),可以在复制逻辑前加几秒的等待时间,避免复制到损坏的不完整文件。

内容的提问来源于stack exchange,提问作者biggboss2019

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.29 08:42:15