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
相关产品推荐
相关产品推荐

