影刀RPA 文件监控目录变化自动触发作者林焱什么情况用你的影刀流程需要监控某个文件夹——当有新文件到达时自动处理你想在Excel文件被修改后自动重新生成报表你需要监控下载目录文件下载完成后自动移动到分类文件夹文件监控是自动化流程的触发器。不需要定时轮询当文件系统发生变化时自动感知并执行相应操作。在影刀RPA中通过Python的watchdog库可以实现实时文件监控。本文讲清楚怎么监控文件变化、怎么过滤有意义的变更、怎么避免频繁触发的抖动问题。怎么做一、轮询方式监控简单方案不需要安装第三方库适合简单场景店群矩阵自动化突破运营极限importosimporttimefromdatetimeimportdatetimedefmonitor_directory(directory,check_interval5,callbackNone): 轮询监控目录变化 check_interval: 检查间隔秒 callback: 发现新文件时的回调函数 # 记录初始文件状态known_filesset()ifos.path.exists(directory):forfinos.listdir(directory):filepathos.path.join(directory,f)ifos.path.isfile(filepath):known_files.add(filepath)print(f开始监控:{directory})print(f初始文件数:{len(known_files)})whileTrue:current_filesset()ifos.path.exists(directory):forfinos.listdir(directory):filepathos.path.join(directory,f)ifos.path.isfile(filepath):current_files.add(filepath)# 新增的文件new_filescurrent_files-known_files# 删除的文件removed_filesknown_files-current_filesforfilepathinnew_files:filenameos.path.basename(filepath)file_sizeos.path.getsize(filepath)mtimedatetime.fromtimestamp(os.path.getmtime(filepath))print(f[新增]{filename}({file_size}bytes,{mtime}))ifcallback:callback(filepath,created)forfilepathinremoved_files:filenameos.path.basename(filepath)print(f[删除]{filename})ifcallback:callback(filepath,deleted)known_filescurrent_files time.sleep(check_interval)defon_file_change(filepath,event):文件变化回调函数ifeventcreated:print(f → 处理新文件:{filepath})# 在这里添加处理逻辑# process_file(filepath)# 使用注意这是阻塞函数在影刀中需要放在适当位置# monitor_directory(D:/data/input, check_interval5, callbackon_file_change)二、watchdog库监控实时方案# 需要安装pip install watchdogfromwatchdog.observersimportObserverfromwatchdog.eventsimportFileSystemEventHandlerimporttimeimportosclassFileChangeHandler(FileSystemEventHandler):文件变化事件处理器def__init__(self,watch_extensionsNone,callbackNone):self.watch_extensionswatch_extensionsor[]# 只监控指定扩展名self.callbackcallback self.last_modified{}# 记录文件最后修改时间防抖动defshould_process(self,filepath):判断是否应该处理该文件# 检查扩展名ifself.watch_extensions:extos.path.splitext(filepath)[1].lower()ifextnotinself.watch_extensions:returnFalse# 防抖动同一个文件1秒内只处理一次current_timetime.time()iffilepathinself.last_modified:ifcurrent_time-self.last_modified[filepath]1:returnFalseself.last_modified[filepath]current_timereturnTruedefon_created(self,event):文件创建事件ifnotevent.is_directory:filepathevent.src_pathifself.should_process(filepath):filenameos.path.basename(filepath)print(f[创建]{filename})ifself.callback:self.callback(filepath,created)defon_modified(self,event):文件修改事件ifnotevent.is_directory:filepathevent.src_pathifself.should_process(filepath):filenameos.path.basename(filepath)print(f[修改]{filename})ifself.callback:self.callback(filepath,modified)defon_deleted(self,event):文件删除事件ifnotevent.is_directory:filepathevent.src_path filenameos.path.basename(filepath)print(f[删除]{filename})ifself.callback:self.callback(filepath,deleted)defon_moved(self,event):文件移动/重命名事件ifnotevent.is_directory:old_nameos.path.basename(event.src_path)new_nameos.path.basename(event.dest_path)print(f[重命名]{old_name}→{new_name})ifself.callback:self.callback(event.dest_path,moved)defstart_watching(directory,extensionsNone,callbackNone):启动文件监控observerObserver()handlerFileChangeHandler(watch_extensionsextensions,callbackcallback)observer.schedule(handler,directory,recursiveFalse)observer.start()print(f开始监控:{directory})ifextensions:print(f监控扩展名:{extensions})try:whileTrue:time.sleep(1)exceptKeyboardInterrupt:observer.stop()observer.join()# 使用defprocess_new_file(filepath,event):处理新文件的回调filenameos.path.basename(filepath)# 等待文件写入完成ifwait_file_ready(filepath,timeout60):print(f → 处理:{filename})# 根据文件类型处理extos.path.splitext(filename)[1].lower()ifext.csv:process_csv(filepath)elifext.xlsx:process_excel(filepath)elifext.json:process_json(filepath)else:print(f → 文件未就绪:{filename})defwait_file_ready(filepath,timeout60):等待文件写入完成importtime old_size-1elapsed0whileelapsedtimeout:try:current_sizeos.path.getsize(filepath)ifcurrent_sizeold_sizeandcurrent_size0:# 大小不再变化认为写入完成returnTrueold_sizecurrent_sizeexceptOSError:passtime.sleep(1)elapsed1returnFalse# start_watching(D:/data/input, extensions[.csv, .xlsx], callbackprocess_new_file)三、实战自动处理下载文件importosimporttimeimportshutilfromwatchdog.observersimportObserverfromwatchdog.eventsimportFileSystemEventHandlerclassDownloadAutoSorter(FileSystemEventHandler):下载目录自动分类处理器# 文件类型分类规则SORT_RULES{# 图片.jpg:图片,.jpeg:图片,.png:图片,.gif:图片,.bmp:图片,# 文档.doc:文档,.docx:文档,.pdf:文档,.txt:文档,.pptx:文档,# 表格.xls:表格,.xlsx:表格,.csv:表格,# 压缩包.zip:压缩包,.rar:压缩包,.7z:压缩包,# 视频.mp4:视频,.avi:视频,.mkv:视频,# 音频.mp3:音频,.wav:音频,.flac:音频,}def__init__(self,download_dir,sort_dir):self.download_dirdownload_dir self.sort_dirsort_dir self.processingset()# 正在处理的文件防止重复处理defon_created(self,event):ifevent.is_directory:returnfilepathevent.src_path filenameos.path.basename(filepath)# 跳过临时文件iffilename.startswith(~)orfilename.endswith(.tmp)orfilename.endswith(.crdownload):return# 跳过已处理的iffilepathinself.processing:returnself.processing.add(filepath)# 等待文件下载完成ifnotself._wait_download_complete(filepath):self.processing.discard(filepath)return# 分类移动文件self._sort_file(filepath)self.processing.discard(filepath)def_wait_download_complete(self,filepath,timeout300):等待下载完成old_size-1stable_count0elapsed0whileelapsedtimeout:try:current_sizeos.path.getsize(filepath)ifcurrent_sizeold_size:stable_count1ifstable_count3:# 连续3次大小不变认为下载完成returnTrueelse:stable_count0old_sizecurrent_sizeexceptOSError:returnFalsetime.sleep(1)elapsed1returnFalsedef_sort_file(self,filepath):分类移动文件filenameos.path.basename(filepath)extos.path.splitext(filename)[1].lower()# 获取目标分类categoryself.SORT_RULES.get(ext,其他)# 创建分类目录target_diros.path.join(self.sort_dir,category)os.makedirs(target_dir,exist_okTrue)# 移动文件target_pathos.path.join(target_dir,filename)# 处理同名文件ifos.path.exists(target_path):name,ext2os.path.splitext(filename)timestamptime.strftime(%H%M%S)target_pathos.path.join(target_dir,f{name}_{timestamp}{ext2})shutil.move(filepath,target_path)print(f[分类]{filename}→{category}/)# 使用# sorter DownloadAutoSorter(C:/Users/xxx/Downloads, D:/Sorted)# observer Observer()# observer.schedule(sorter, C:/Users/xxx/Downloads, recursiveFalse)# observer.start()四、监控文件修改触发操作classFileModifyTrigger(FileSystemEventHandler):文件修改时触发指定操作def__init__(self,watch_file,trigger_func,debounce_seconds5):self.watch_fileos.path.abspath(watch_file)self.trigger_functrigger_func self.debounce_secondsdebounce_seconds self.last_trigger0defon_modified(self,event):ifevent.is_directory:returnfilepathos.path.abspath(event.src_path)iffilepath!self.watch_file:return# 防抖动两次触发间隔至少debounce_seconds秒current_timetime.time()ifcurrent_time-self.last_triggerself.debounce_seconds:returnself.last_triggercurrent_timeprint(f[触发]{self.watch_file}被修改)try:self.trigger_func()exceptExceptionase:print(f[错误] 触发操作失败:{e})# 使用当data.xlsx被修改时自动重新生成报表defregenerate_report():重新生成报表print(开始重新生成报表...)# ... 报表生成逻辑 ...print(报表生成完成)# trigger FileModifyTrigger(D:/data/source.xlsx, regenerate_report, debounce_seconds10)# observer Observer()# observer.schedule(trigger, D:/data, recursiveFalse)# observer.start()有什么坑坑1文件未写入完成就处理现象文件刚创建就立即处理结果文件内容不完整或为空。原因文件创建事件在文件刚开始写入时就触发了此时文件内容还没写完。大文件尤其严重。解决文件创建后等待文件大小稳定——连续检查文件大小大小不再变化时认为写入完成。参考_wait_download_complete函数。一般等待2-3秒检查大小是否稳定即可。坑2同一事件触发多次现象复制一个文件到监控目录on_created被触发了好几次。原因文件系统在写入文件时可能触发多次事件创建、写入、修改。有些编辑器保存文件时先删除再创建也会触发多个事件。解决用防抖动机制——记录最近处理过的文件和时间同一文件在短时间如1秒内只处理一次。用processing集合标记正在处理的文件处理完后移除。坑3临时文件干扰现象监控目录出现了~$temp.xlsx、.crdownload等临时文件被误处理。temu店群自动化报活动案例原因Office编辑文件时会创建~$开头的临时文件。Chrome下载时会创建.crdownload临时文件。解决在事件处理器中过滤临时文件iffilename.startswith(~$)orfilename.endswith(.tmp)orfilename.endswith(.crdownload):return坑4watchdog在影刀中的阻塞问题现象watchdog的observer.start()后会阻塞影刀流程无法继续执行后续指令。原因observer.join()或while True循环会阻塞主线程。解决不要用observer.join()让observer在后台运行。在影刀中把文件监控放在一个单独的Python代码块中设置一个退出条件如检测到某个标志文件就停止监控。或者用轮询方式代替watchdog更容易控制生命周期。坑5网络驱动器监控不工作现象监控网络驱动器如\\server\share或映射的Z盘时文件变化事件不触发。原因watchdog依赖操作系统的文件系统通知APIWindows的ReadDirectoryChangesW网络驱动器可能不支持或延迟很大。解决对网络驱动器用轮询方式代替事件驱动。增加轮询频率到每2-3秒检查一次。如果性能允许检查文件修改时间而不是比较文件列表。