影刀RPA 文件夹监控：新文件到达自动触发处理

影刀RPA 文件夹监控新文件到达自动触发处理作者林焱有些场景需要来一个文件处理一个——比如客户通过FTP上传了订单文件、或者某个共享文件夹里有新的PDF需要识别。影刀的定时任务可以周期性检查但间隔太长不够实时、太短又浪费资源。文件夹监控就是解决这个问题的。watchdog库实现文件监控# pip install watchdogimporttimefromwatchdog.observersimportObserverfromwatchdog.eventsimportFileSystemEventHandlerclassFileHandler(FileSystemEventHandler):defon_created(self,event):文件创建时触发ifevent.is_directory:returnfilepathevent.src_path filenameos.path.basename(filepath)print(f发现新文件:{filename})# 处理逻辑iffilename.endswith(.xlsx):process_excel(filepath)eliffilename.endswith(.pdf):process_pdf(filepath)eliffilename.endswith(.csv):process_csv(filepath)defon_modified(self,event):文件修改时触发处理大文件时等写入完成再处理ifevent.is_directory:return# 有些程序写文件时会多次触发modified事件# 这里可以加延时判断文件是否写入完成# 启动监控watch_dirD:/待处理文件/event_handlerFileHandler()observerObserver()observer.schedule(event_handler,watch_dir,recursiveFalse)observer.start()try:whileTrue:time.sleep(1)exceptKeyboardInterrupt:observer.stop()observer.join()大文件防抖处理店群矩阵自动化突破运营极限大文件写入时可能触发多次modified事件。需要防抖——等文件确实不再变化了再处理importosimporttimeclassDebouncedHandler(FileSystemEventHandler):def__init__(self):self.processingset()# 正在处理的文件self.file_sizes{}# 记录文件大小defon_created(self,event):ifevent.is_directory:returnfilepathevent.src_path# 延迟1秒等文件完全写入time.sleep(1)iffilepathinself.processing:returnself.processing.add(filepath)try:# 确认文件大小不再变化防抖stableFalsefor_inrange(5):# 最多等5秒sizeos.path.getsize(filepath)iffilepathinself.file_sizesandsizeself.file_sizes[filepath]:stableTruebreakself.file_sizes[filepath]size time.sleep(1)ifstable:process_file(filepath)# 处理完移到归档目录archive_pathos.path.join(D:/已处理/,os.path.basename(filepath))os.rename(filepath,archive_path)finally:self.processing.discard(filepath)按文件类型分流defprocess_file(filepath):extos.path.splitext(filepath)[1].lower()handlers{.xlsx:handle_excel,.xls:handle_excel,.csv:handle_csv,.pdf:handle_pdf,.png:handle_image,.jpg:handle_image,}handlerhandlers.get(ext)ifhandler:try:handler(filepath)log(f✓ 已处理:{filepath})exceptExceptionase:log(f✗ 处理失败:{filepath}, 错误:{e})move_to_error(filepath)else:log(f⚠️ 不支持的文件类型:{ext})move_to_unknown(filepath)影刀集成方式watchdog是持续运行的适合独立部署。如果只在影刀流程中临时使用可以改用定时轮询importosimporttime WATCH_DIRD:/待处理/PROCESSED_DIRD:/已处理/defpoll_and_process():影刀每30秒调用一次filesos.listdir(WATCH_DIR)forfilenameinfiles:filepathos.path.join(WATCH_DIR,filename)# 跳过正在写入的文件大小还在变化try:size1os.path.getsize(filepath)time.sleep(1)size2os.path.getsize(filepath)ifsize1!size2:continue# 还在写入中except:continueprocess_file(filepath)os.rename(filepath,os.path.join(PROCESSED_DIR,filename))# 影刀定时任务每30秒调一次poll_and_process()踩坑实录坑1watchdog在Windows上的路径Windows路径用反斜杠watchdog内部可能处理不好。统一用正斜杠或os.path.normpath()watch_diros.path.normpath(D:/待处理文件/)temu店群自动化报活动案例坑2某些程序写文件会创建临时文件Word保存文件时先写一个~$文件名.docx的临时文件再重命名。你的监控要忽略~$开头的临时文件ifos.path.basename(filepath).startswith(~$):return# 忽略临时文件坑3observer线程和影刀主流程冲突如果把watchdog启动在影刀的Python代码块里observer.start()后代码就卡住了。解决办法在单独的Python脚本中运行影刀通过文件/数据库与之通信。坑4文件权限刚创建的文件可能被其他程序锁定比如Excel正在打开此时重命名或删除会失败。加try/except处理文件占用try:os.rename(filepath,archive_path)exceptPermissionError:print(f文件被占用稍后重试:{filepath})写在最后文件夹监控让影刀从定时拉取变成事件驱动——不用每分钟查询一次文件一来就处理。用轮询方案简单稳定适合影刀内置集成用watchdog方案适合高实时性独立运行场景。