影刀RPA 流程的暂停与恢复：手动确认与延时等待
影刀RPA 流程的暂停与恢复：手动确认与延时等待
作者：林焱
不是所有流程都需要从头跑到尾不停歇。有些场景需要流程暂停等待人工确认、有些需要等待某个条件满足后继续、有些需要在关键步骤前给用户一个检查的机会。但很多新手不知道怎么正确实现暂停，要么用死循环疯狂轮询占满CPU，要么暂停后就恢复不了了。
这篇文章把流程暂停与恢复的几种方案讲全。
一、延时等待
1.1 固定延时
最简单的暂停方式，等待固定时间后继续：
【延时等待】 等待时间：5秒Python中：
importtime time.sleep(5)# 暂停5秒time.sleep(0.5)# 暂停500毫秒1.2 随机延时
模拟人类操作节奏，避免被检测为机器人：
importtimeimportrandom# 随机等待1-3秒delay=random.uniform(1,3)time.sleep(delay)# 随机等待，带偏好delay=random.choice([1,2,2,3,3,3,5])# 更倾向于2-3秒time.sleep(delay)1.3 延时的适用场景
- 页面加载后等待渲染完成
- 操作之间避免太快触发频率限制
- 模拟人类操作节奏
店群矩阵自动化突破运营极限！
- 等待动画/过渡效果完成
坑：用延时代替等待元素
# 不推荐：固定等3秒，可能不够也可能浪费time.sleep(3)click_element("#button")# 推荐：等待元素出现wait_for_element("#button",timeout=10)click_element("#button")固定延时是最低效的等待方式。能用【等待元素】就用【等待元素】。
二、条件等待
2.1 轮询等待
等待某个条件满足后继续执行：
importtimedefwait_for_condition(check_func,timeout=60,interval=2):"""等待条件满足"""start=time.time()whiletime.time()-start<timeout:ifcheck_func():returnTruetime.sleep(interval)returnFalse# 超时# 使用：等待文件出现deffile_exists():importosreturnos.path.exists("D:\\data\\ready.flag")ifwait_for_condition(file_exists,timeout=300,interval=5):print("文件已出现，继续执行")else:print("等待超时，文件未出现")2.2 等待文件出现
场景：等待另一个流程生成数据文件后继续处理。
importosimporttimefromdatetimeimportdatetimedefwait_for_file(filepath,timeout=600,interval=5):"""等待文件出现"""print(f"等待文件：{filepath}")start=time.time()whiletime.time()-start<timeout:ifos.path.exists(filepath):# 检查文件是否写入完成（大小稳定）size1=os.path.getsize(filepath)time.sleep(1)size2=os.path.getsize(filepath)ifsize1==size2andsize1>0:print(f"文件已就绪：{filepath}（{size1}字节）")returnTruetime.sleep(interval)print(f"等待超时：{filepath}")returnFalse# 使用ifwait_for_file("D:\\data\\export.xlsx",timeout=300):# 处理文件process_file("D:\\data\\export.xlsx")else:print("文件未出现，跳过处理")2.3 等待网页加载
importtimedefwait_for_page_load(web_handle,timeout=30):"""等待网页加载完成"""start=time.time()whiletime.time()-start<timeout:# 检查document.readyStatestate=execute_js(web_handle,"return document.readyState")ifstate=="complete":returnTruetime.sleep(0.5)returnFalse三、人工确认暂停
3.1 弹窗确认
在关键步骤前暂停，弹窗让用户确认：
importctypesimportsysdefask_user_confirm(message):"""弹窗让用户确认"""result=ctypes.windll.user32.MessageBoxW(0,message,"RPA流程确认",4# 4 = MB_YESNO)returnresult==6# 6 = IDYES# 使用ifask_user_confirm("即将提交订单，确认继续？"):print("用户确认，继续执行")# 提交订单else:print("用户取消，停止执行")sys.exit(0)3.2 文件信号确认
场景：流程暂停，等待用户在指定位置创建一个文件来确认继续。
importosimporttimedefwait_for_user_confirmation(signal_file,timeout=3600,message=""):"""等待用户确认（创建信号文件）"""# 清除旧的信号文件ifos.path.exists(signal_file):os.remove(signal_file)print(f"\n{'='*50}")print(f"流程已暂停，等待人工确认")print(f"说明：{message}")print(f"确认方式：创建文件{signal_file}")print(f"超时时间：{timeout}秒")print(f"{'='*50}\n")start=time.time()whiletime.time()-start<timeout:ifos.path.exists(signal_file):print("收到确认信号，继续执行")os.remove(signal_file)# 清理信号文件returnTruetime.sleep(2)print("等待超时，流程终止")returnFalse# 使用ifwait_for_user_confirmation("D:\\confirm.flag",timeout=600,message="请检查采集数据是否正确，确认后创建此文件继续"):# 继续执行pass3.3 输入框暂停
场景：流程需要人工输入验证码或补充信息。
importtkinterastkfromtkinterimportsimpledialogdefget_user_input(prompt,title="请输入"):"""弹窗让用户输入"""root=tk.Tk()root.withdraw()# 隐藏主窗口user_input=simpledialog.askstring(title,prompt)root.destroy()returnuser_input# 使用：让用户输入验证码captcha=get_user_input("请输入验证码：")ifcaptcha:# 填写验证码并提交fill_input("#captcha",captcha)else:print("用户取消输入")四、暂停恢复的设计模式
4.1 断点续跑模式
流程记录当前执行到哪一步，暂停后可以从断点恢复：
importjsonimportosclassFlowCheckpoint:"""流程断点管理"""def__init__(self,checkpoint_file):self.file=checkpoint_file self.state=self._load()def_load(self):ifos.path.exists(self.file):withopen(self.file,"r")asf:returnjson.load(f)return{"completed_steps":[],"current_step":None,"data":{}}defsave_step(self,step_name):"""记录完成的步骤"""ifstep_namenotinself.state["completed_steps"]:self.state["completed_steps"].append(step_name)self.state["current_step"]=step_name self._save()defis_step_done(self,step_name):"""检查步骤是否已完成"""returnstep_nameinself.state["completed_steps"]defsave_data(self,key,value):"""保存数据"""self.state["data"][key]=value self._save()defget_data(self,key,default=None):"""获取数据"""returnself.state["data"].get(key,default)defclear(self):"""清除断点"""self.state={"completed_steps":[],"current_step":None,"data":{}}self._save()def_save(self):withopen(self.file,"w")asf:json.dump(self.state,f,ensure_ascii=False,indent=2)# 使用checkpoint=FlowCheckpoint("D:\\checkpoint.json")# 每个步骤前检查是否已完成ifnotcheckpoint.is_step_done("login"):login()checkpoint.save_step("login")ifnotcheckpoint.is_step_done("collect_page1"):data=collect_page(1)checkpoint.save_data("page1_data",data)checkpoint.save_step("collect_page1")ifnotcheckpoint.is_step_done("process_data"):data=checkpoint.get_data("page1_data")result=process(data)checkpoint.save_data("result",result)checkpoint.save_step("process_data")# 全部完成后清除断点checkpoint.clear()4.2 暂停标志模式
流程定期检查暂停标志，发现暂停就等待恢复：
importosimporttimedefcheck_pause():"""检查是否需要暂停"""pause_file="D:\\pause.flag"returnos.path.exists(pause_file)defwait_if_paused():"""如果暂停了就等待恢复"""whilecheck_pause():[video(video-i53zDdgW-1784482874853)(type-csdn)(url-https://live.csdn.net/v/embed/524992)(image-https://v-blog.csdnimg.cn/asset/b59aed2f01d4fe8583467562aaf4dcfd/cover/Cover0.jpg)(title-temu店群自动化报活动案例)]print("流程已暂停，等待恢复...")time.sleep(5)print("流程恢复")# 在关键位置调用foriteminitems:wait_if_paused()# 检查暂停process(item)用户创建D:\pause.flag文件暂停流程，删除文件恢复流程。
五、实战场景
5.1 采集前人工确认
# 1. 自动采集第一页数据preview_data=collect_page(1)# 2. 保存预览数据importpandasaspd df=pd.DataFrame(preview_data)df.to_excel("D:\\preview.xlsx",index=False)# 3. 暂停等待确认ifwait_for_user_confirmation("D:\\confirm.flag",timeout=300,message="请检查 D:\\preview.xlsx 中的数据格式是否正确"):# 4. 确认后继续采集所有页all_data=preview_data page=2whilehas_next_page():all_data.extend(collect_page(page))page+=1print(f"共采集{len(all_data)}条")else:print("用户未确认，流程终止")5.2 等待数据就绪
# 流程A：导出数据（另一个系统在做）# 流程B：等待导出完成后处理# 等待导出文件出现export_file="D:\\exports\\daily_report.xlsx"ifwait_for_file(export_file,timeout=1800,interval=10):# 文件就绪，开始处理df=pd.read_excel(export_file)process(df)else:# 超时send_notification("导出文件未就绪，流程超时终止")5.3 定时等待
fromdatetimeimportdatetimeimporttimedefwait_until(target_time_str):"""等待到指定时间"""now=datetime.now()target=datetime.strptime(f"{now.strftime('%Y-%m-%d')}{target_time_str}","%Y-%m-%d %H:%M:%S")# 如果目标时间已过，等到明天iftarget<now:fromdatetimeimporttimedelta target+=timedelta(days=1)wait_seconds=(target-now).total_seconds()print(f"等待{wait_seconds:.0f}秒到{target.strftime('%H:%M:%S')}")time.sleep(wait_seconds)# 等到9点再开始wait_until("09:00:00")print("开始执行")六、避坑清单
坑1：轮询间隔太短
# 错误：每0.1秒检查一次，CPU占用高whilenotcondition():time.sleep(0.1)# 推荐：根据场景设置合理间隔whilenotcondition():time.sleep(2)# 文件检查2-5秒# 网络检查可以更频繁坑2：暂停后忘记恢复
用户创建了暂停文件但忘记删除，流程一直暂停。设置超时机制。
坑3：暂停时资源未释放
流程暂停时浏览器一直开着，占用资源。长时间暂停考虑关闭浏览器，恢复时重新打开。
坑4：断点文件丢失
断点文件存在系统盘，系统崩溃后丢失。断点文件存在安全的位置，或者同步到云端。
坑5：弹窗阻塞流程
Windows弹窗是阻塞的，弹窗期间流程完全停止。如果需要非阻塞的通知，用文件信号或日志写入。