企业官网建设流程全解析

影刀RPA 流程灾难恢复：断网断电后的数据保护方案

作者：林焱

什么情况用

自动化流程最怕的不是逻辑bug，而是外部灾难：

• 半夜跑了3小时的数据导入流程，到第28000行时网络断了，前面28000行的结果也丢了
• 批量发送邮件的流程跑了200封，电脑自动更新重启了，完全不知道哪些已发哪些未发
• 数据处理到一半，影刀客户端崩溃了，中间状态全丢了，又得从头跑

这些场景的共同问题：流程没有「断点续跑」能力——一旦中断，全部推倒重来。

核心场景：长流程需要支持中断后从断点恢复，而不是从头开始。

怎么做

第一步：设计检查点机制

核心思路：每隔N条记录或每隔M分钟，把当前进度写到本地文件。中断后重新启动，先读检查点，跳过已完成的部分。

正常流程：step1 → step2 → [检查点] → step3 → step4 → [检查点] → step5 中断恢复：读取检查点 → 发现step4已完成 → 直接从step5开始

第二步：实现通用的检查点管理器

importjsonimportosimporttimefromdatetimeimportdatetimeclassCheckpointManager:""" 检查点管理器——支持断点续跑 每个流程实例对应一个检查点文件 """def__init__(self,flow_name,checkpoint_dir="./checkpoints"):self.flow_name=flow_name self.checkpoint_dir=checkpoint_dir# 确保目录存在os.makedirs(checkpoint_dir,exist_ok=True)# 检查点文件路径safe_name=flow_name.replace(" ","_").replace("/","_")self.checkpoint_file=os.path.join(checkpoint_dir,f"{safe_name}_checkpoint.json")defsave(self,data):"""保存检查点"""checkpoint={"flow_name":self.flow_name,"timestamp":datetime.now().isoformat(),"data":data}# 先写到临时文件，成功后再改名——防止写一半断电导致文件损坏temp_file=self.checkpoint_file+".tmp"withopen(temp_file,"w",encoding="utf-8")asf:json.dump(checkpoint,f,ensure_ascii=False,indent=2)os.replace(temp_file,self.checkpoint_file)# 原子操作defload(self):"""读取检查点"""ifnotos.path.exists(self.checkpoint_file):returnNonetry:withopen(self.checkpoint_file,"r",encoding="utf-8")asf:checkpoint=json.load(f)returncheckpoint.get("data")except(json.JSONDecodeError,IOError):print(f"检查点文件损坏，忽略")returnNonedefclear(self):"""清除检查点（流程正常结束后调用）"""ifos.path.exists(self.checkpoint_file):os.remove(self.checkpoint_file)defget_resume_info(self):"""获取断点恢复信息"""checkpoint=self.load()ifcheckpointisNone:return{"resumable":False,"message":"无检查点，需要从头执行"}return{"resumable":True,"last_position":checkpoint.get("last_index",0),"processed_count":checkpoint.get("processed_count",0),"timestamp":checkpoint.get("timestamp","未知")}

第三步：实战——批量数据处理的断点续跑

最经典的场景：读取10000行数据逐条处理，中间断了能继续。

importpandasaspdimporttimedefprocess_with_checkpoint(input_file,output_file,batch_size=100):""" 带断点续跑的批量数据处理 """cpm=CheckpointManager("batch_data_processing")# 1. 尝试恢复resume_info=cpm.get_resume_info()# 2. 读取全部数据df=pd.read_csv(input_file)total_rows=len(df)print(f"共{total_rows}行待处理")# 3. 确定起始位置ifresume_info["resumable"]:start_index=resume_info["last_position"]+1print(f"从检查点恢复，从第{start_index+1}行开始（已处理{start_index}行）")# 加载已有的处理结果ifos.path.exists(output_file):results=pd.read_csv(output_file).to_dict("records")else:results=[]else:start_index=0results=[]# 4. 逐条处理，每batch_size条存一次检查点try:foriinrange(start_index,total_rows):row=df.iloc[i]# ===== 实际业务处理逻辑 =====processed={"id":row.get("id",i),"name":row.get("name",""),"status":"processed","processed_at":datetime.now().isoformat()}# ===== 业务逻辑结束 =====results.append(processed)# 每batch_size条存一次检查点 + 保存结果if(i+1)%batch_size==0:cpm.save({"last_index":i,"processed_count":len(results),"total_count":total_rows})# 增量保存结果pd.DataFrame(results).to_csv(output_file,index=False)progress=(i+1)/total_rows*100print(f"进度：{progress:.1f}%，已处理{i+1}/{total_rows}")# 5. 全部完成，清除检查点cpm.clear()pd.DataFrame(results).to_csv(output_file,index=False)print(f"全部完成！共处理{total_rows}行")exceptExceptionase:print(f"处理中断于第{i+1}行：{e}")print(f"已保存检查点，下次运行将从第{i+1}行继续")# 保存当前进度cpm.save({"last_index":i-1,"processed_count":len(results)-1,"total_count":total_rows})pd.DataFrame(results).to_csv(output_file,index=False)raise# 重新抛出异常，让上层知道中断了

第四步：网络请求的重试+断点续跑

API调用的灾难恢复——指数退避重试 + 失败队列持久化。

importrequestsimporttimeimportrandomclassResilientAPICaller:"""健壮的API调用器——自动重试+失败持久化"""def__init__(self,checkpoint_dir="./checkpoints"):self.checkpoint_dir=checkpoint_dir self.failed_queue_file=os.path.join(checkpoint_dir,"failed_queue.json")os.makedirs(checkpoint_dir,exist_ok=True)defcall_with_retry(self,url,max_retries=3,base_delay=1,**kwargs):""" 带指数退避重试的API调用 base_delay: 基础延迟秒数，每次重试翻倍 """forattemptinrange(max_retries):try:resp=requests.post(url,timeout=30,**kwargs)resp.raise_for_status()returnresp.json()exceptrequests.exceptions.Timeout:delay=base_delay*(2**attempt)+random.uniform(0,1)print(f"超时，第{attempt+1}次重试，等待{delay:.1f}秒...")time.sleep(delay)exceptrequests.exceptions.HTTPErrorase:# 4xx错误（除了429）不重试——重试也没用ifresp.status_code==429:# 频率限制retry_after=int(resp.headers.get("Retry-After",30))print(f"触发频率限制，等待{retry_after}秒...")time.sleep(retry_after)continueelifresp.status_code>=500:# 服务端错误，可以重试delay=base_delay*(2**attempt)print(f"服务端错误{resp.status_code}，第{attempt+1}次重试...")time.sleep(delay)continueelse:raise# 4xx客户端错误，不重试# 所有重试都失败raiseException(f"API调用失败，已重试{max_retries}次")defbatch_call(self,items,url,result_handler=None):""" 批量API调用，失败的进入失败队列 """results=[]failed=[]fori,iteminenumerate(items):try:result=self.call_with_retry(url,json=item,max_retries=3)results.append({"input":item,"output":result,"status":"success"})ifresult_handler:result_handler(item,result)exceptExceptionase:failed.append({"input":item,"error":str(e),"index":i})print(f"第{i+1}条失败：{e}")# 每50条保存失败队列iflen(failed)>0andlen(failed)%50==0:self._save_failed_queue(failed)# 最终保存iffailed:self._save_failed_queue(failed)print(f"共{len(failed)}条失败，已保存到失败队列")returnresults,faileddefretry_failed_queue(self,url):"""重试失败队列中的请求"""ifnotos.path.exists(self.failed_queue_file):print("没有待重试的请求")returnwithopen(self.failed_queue_file,"r")asf:failed_items=json.load(f)print(f"重试{len(failed_items)}条失败的请求...")remaining=[]foriteminfailed_items:try:result=self.call_with_retry(url,json=item["input"],max_retries=2)print(f"重试成功：第{item['index']+1}条")exceptExceptionase:remaining.append(item)print(f"仍失败：第{item['index']+1}条 -{e}")self._save_failed_queue(remaining)print(f"重试完成，{len(remaining)}条仍需处理")def_save_failed_queue(self,items):withopen(self.failed_queue_file,"w")asf:json.dump(items,f,ensure_ascii=False,indent=2)

第五步：邮件群发的「已发送」追踪

发200封邮件，中间断了，哪些发了哪些没发？

classEmailTracker:"""邮件发送追踪器"""def__init__(self,tracker_file="./checkpoints/email_sent.json"):self.tracker_file=tracker_file self.sent_list=self._load_sent()def_load_sent(self):ifos.path.exists(self.tracker_file):withopen(self.tracker_file,"r")asf:returnjson.load(f)return[]defis_sent(self,email):"""检查邮件是否已发送"""returnemailinself.sent_listdefmark_sent(self,email):"""标记为已发送"""self.sent_list.append(email)self._save()def_save(self):withopen(self.tracker_file,"w")asf:json.dump(self.sent_list,f,ensure_ascii=False,indent=2)defclear(self):"""清空追踪（完成后调用）"""self.sent_list=[]ifos.path.exists(self.tracker_file):os.remove(self.tracker_file)# 使用tracker=EmailTracker()forrecipientinrecipient_list:# 跳过已发送的iftracker.is_sent(recipient):print(f"跳过已发送：{recipient}")continuetry:send_email(recipient,subject,body)tracker.mark_sent(recipient)print(f"已发送：{recipient}")exceptExceptionase:print(f"发送失败：{recipient}-{e}")# 已成功发送的不会被丢

有什么坑

坑1：检查点频率设计不当

存太频繁（每1条存一次）→ IO开销大，流程变慢。存太稀疏（每10000条存一次）→ 断了损失大。

经验值：数据批量处理，每100-500条存一次；邮件发送，每发1条就标记；文件操作，每处理一个文件就标记。

坑2：检查点文件损坏

如果在写入检查点文件的过程中断电，文件可能写了一半，下次读取时JSON解析失败，整个流程无法恢复。

解决方法：先写临时文件，再原子替换（os.replace），永远不要直接覆盖原文件。

坑3：数据处理的幂等性问题

断点续跑意味着同一条数据可能被处理两遍。如果你的处理逻辑不是幂等的（比如给金额+1），恢复后会出错。

解决方法：处理逻辑设计成幂等的——用唯一ID做去重，或者用「先检查再处理」的模式。

坑4：业务数据的原子性

你在处理一条数据库记录时断电了——记录改了一半。恢复后可能拿到一条脏数据。

解决方法：关键数据操作尽量用事务（数据库事务、API的upsert语义），要么全成功，要么全回滚。

总结：长流程必须做灾难恢复设计，否则跑一次挂一次。核心三件套：检查点机制（知道断在哪）、幂等处理（恢复不重复）、失败队列（失败的不丢）。花一小时加好这些，省下的是无数次头疼的重跑。