企业官网建设流程全解析

影刀RPA 文件监控：目录变化自动触发

作者：林焱

什么情况用

你的影刀流程需要监控某个文件夹——当有新文件到达时自动处理？你想在Excel文件被修改后自动重新生成报表？你需要监控下载目录，文件下载完成后自动移动到分类文件夹？

文件监控是自动化流程的触发器。不需要定时轮询，当文件系统发生变化时自动感知并执行相应操作。在影刀RPA中，通过Python的watchdog库可以实现实时文件监控。

本文讲清楚：怎么监控文件变化、怎么过滤有意义的变更、怎么避免频繁触发的抖动问题。

怎么做

一、轮询方式监控（简单方案）

不需要安装第三方库，适合简单场景：

importosimporttimefromdatetimeimportdatetimedefmonitor_directory(directory,check_interval=5,callback=None):""" 轮询监控目录变化 check_interval: 检查间隔（秒） callback: 发现新文件时的回调函数 """# 记录初始文件状态known_files=set()ifos.path.exists(directory):forfinos.listdir(directory):filepath=os.path.join(directory,f)ifos.path.isfile(filepath):known_files.add(filepath)print(f"开始监控:{directory}")print(f"初始文件数:{len(known_files)}")whileTrue:current_files=set()ifos.path.exists(directory):forfinos.listdir(directory):filepath=os.path.join(directory,f)ifos.path.isfile(filepath):current_files.add(filepath)# 新增的文件new_files=current_files-known_files# 删除的文件removed_files=known_files-current_filesforfilepathinnew_files:filename=os.path.basename(filepath)file_size=os.path.getsize(filepath)mtime=datetime.fromtimestamp(os.path.getmtime(filepath))print(f"[新增]{filename}({file_size}bytes,{mtime})")ifcallback:callback(filepath,"created")forfilepathinremoved_files:filename=os.path.basename(filepath)print(f"[删除]{filename}")ifcallback:callback(filepath,"deleted")known_files=current_files time.sleep(check_interval)defon_file_change(filepath,event):"""文件变化回调函数"""ifevent=="created":print(f" → 处理新文件:{filepath}")# 在这里添加处理逻辑# process_file(filepath)# 使用（注意：这是阻塞函数，在影刀中需要放在适当位置）# monitor_directory("D:/data/input", check_interval=5, callback=on_file_change)

二、watchdog库监控（实时方案）

# 需要安装：pip install watchdogfromwatchdog.observersimportObserverfromwatchdog.eventsimportFileSystemEventHandlerimporttimeimportosclassFileChangeHandler(FileSystemEventHandler):"""文件变化事件处理器"""def__init__(self,watch_extensions=None,callback=None):self.watch_extensions=watch_extensionsor[]# 只监控指定扩展名self.callback=callback self.last_modified={}# 记录文件最后修改时间（防抖动）defshould_process(self,filepath):"""判断是否应该处理该文件"""# 检查扩展名ifself.watch_extensions:ext=os.path.splitext(filepath)[1].lower()ifextnotinself.watch_extensions:returnFalse# 防抖动：同一个文件1秒内只处理一次current_time=time.time()iffilepathinself.last_modified:ifcurrent_time-self.last_modified[filepath]<1:returnFalseself.last_modified[filepath]=current_timereturnTruedefon_created(self,event):"""文件创建事件"""ifnotevent.is_directory:filepath=event.src_pathifself.should_process(filepath):filename=os.path.basename(filepath)print(f"[创建]{filename}")ifself.callback:self.callback(filepath,"created")defon_modified(self,event):"""文件修改事件"""ifnotevent.is_directory:filepath=event.src_pathifself.should_process(filepath):filename=os.path.basename(filepath)print(f"[修改]{filename}")ifself.callback:self.callback(filepath,"modified")defon_deleted(self,event):"""文件删除事件"""ifnotevent.is_directory:filepath=event.src_path filename=os.path.basename(filepath)print(f"[删除]{filename}")ifself.callback:self.callback(filepath,"deleted")defon_moved(self,event):"""文件移动/重命名事件"""ifnotevent.is_directory:old_name=os.path.basename(event.src_path)new_name=os.path.basename(event.dest_path)print(f"[重命名]{old_name}→{new_name}")ifself.callback:self.callback(event.dest_path,"moved")defstart_watching(directory,extensions=None,callback=None):"""启动文件监控"""observer=Observer()handler=FileChangeHandler(watch_extensions=extensions,callback=callback)observer.schedule(handler,directory,recursive=False)observer.start()print(f"开始监控:{directory}")ifextensions:print(f"监控扩展名:{extensions}")try:whileTrue:time.sleep(1)exceptKeyboardInterrupt:observer.stop()observer.join()# 使用defprocess_new_file(filepath,event):"""处理新文件的回调"""filename=os.path.basename(filepath)# 等待文件写入完成ifwait_file_ready(filepath,timeout=60):print(f" → 处理:{filename}")# 根据文件类型处理ext=os.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,timeout=60):"""等待文件写入完成"""importtime old_size=-1elapsed=0whileelapsed<timeout:try:current_size=os.path.getsize(filepath)ifcurrent_size==old_sizeandcurrent_size>0:# 大小不再变化，认为写入完成returnTrueold_size=current_sizeexceptOSError:passtime.sleep(1)elapsed+=1returnFalse# start_watching("D:/data/input", extensions=[".csv", ".xlsx"], callback=process_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_dir=download_dir self.sort_dir=sort_dir self.processing=set()# 正在处理的文件（防止重复处理）defon_created(self,event):ifevent.is_directory:returnfilepath=event.src_path filename=os.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,timeout=300):"""等待下载完成"""old_size=-1stable_count=0elapsed=0whileelapsed<timeout:try:current_size=os.path.getsize(filepath)ifcurrent_size==old_size:stable_count+=1ifstable_count>=3:# 连续3次大小不变，认为下载完成returnTrueelse:stable_count=0old_size=current_sizeexceptOSError:returnFalsetime.sleep(1)elapsed+=1returnFalsedef_sort_file(self,filepath):"""分类移动文件"""filename=os.path.basename(filepath)ext=os.path.splitext(filename)[1].lower()# 获取目标分类category=self.SORT_RULES.get(ext,"其他")# 创建分类目录target_dir=os.path.join(self.sort_dir,category)os.makedirs(target_dir,exist_ok=True)# 移动文件target_path=os.path.join(target_dir,filename)# 处理同名文件ifos.path.exists(target_path):name,ext2=os.path.splitext(filename)timestamp=time.strftime("%H%M%S")target_path=os.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", recursive=False)# observer.start()

四、监控文件修改触发操作

classFileModifyTrigger(FileSystemEventHandler):"""文件修改时触发指定操作"""def__init__(self,watch_file,trigger_func,debounce_seconds=5):self.watch_file=os.path.abspath(watch_file)self.trigger_func=trigger_func self.debounce_seconds=debounce_seconds self.last_trigger=0defon_modified(self,event):ifevent.is_directory:returnfilepath=os.path.abspath(event.src_path)iffilepath!=self.watch_file:return# 防抖动：两次触发间隔至少debounce_seconds秒current_time=time.time()ifcurrent_time-self.last_trigger<self.debounce_seconds:returnself.last_trigger=current_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_seconds=10)# observer = Observer()# observer.schedule(trigger, "D:/data", recursive=False)# observer.start()

有什么坑

坑1：文件未写入完成就处理

现象：文件刚创建就立即处理，结果文件内容不完整或为空。

原因：文件创建事件在文件刚开始写入时就触发了，此时文件内容还没写完。大文件尤其严重。

解决：文件创建后等待文件大小稳定——连续检查文件大小，大小不再变化时认为写入完成。参考_wait_download_complete函数。一般等待2-3秒检查大小是否稳定即可。

坑2：同一事件触发多次

现象：复制一个文件到监控目录，on_created被触发了好几次。

原因：文件系统在写入文件时可能触发多次事件（创建、写入、修改）。有些编辑器保存文件时先删除再创建，也会触发多个事件。

解决：用防抖动机制——记录最近处理过的文件和时间，同一文件在短时间（如1秒）内只处理一次。用processing集合标记正在处理的文件，处理完后移除。

坑3：临时文件干扰

现象：监控目录出现了~$temp.xlsx、.crdownload等临时文件，被误处理。

原因：Office编辑文件时会创建~$开头的临时文件。Chrome下载时会创建.crdownload临时文件。

解决：在事件处理器中过滤临时文件：

iffilename.startswith("~$")orfilename.endswith(".tmp")orfilename.endswith(".crdownload"):return

坑4：watchdog在影刀中的阻塞问题

现象：watchdog的observer.start()后会阻塞，影刀流程无法继续执行后续指令。

原因：observer.join()或while True循环会阻塞主线程。

解决：不要用observer.join()，让observer在后台运行。在影刀中，把文件监控放在一个单独的Python代码块中，设置一个退出条件（如检测到某个标志文件就停止监控）。或者用轮询方式代替watchdog，更容易控制生命周期。

坑5：网络驱动器监控不工作

现象：监控网络驱动器（如\\server\share或映射的Z盘）时，文件变化事件不触发。

原因：watchdog依赖操作系统的文件系统通知API（Windows的ReadDirectoryChangesW），网络驱动器可能不支持或延迟很大。

解决：对网络驱动器用轮询方式代替事件驱动。增加轮询频率到每2-3秒检查一次。如果性能允许，检查文件修改时间而不是比较文件列表。