影刀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_nameflow_name self.checkpoint_dircheckpoint_dir# 确保目录存在os.makedirs(checkpoint_dir,exist_okTrue)# 检查点文件路径safe_nameflow_name.replace( ,_).replace(/,_)self.checkpoint_fileos.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_fileself.checkpoint_file.tmpwithopen(temp_file,w,encodingutf-8)asf:json.dump(checkpoint,f,ensure_asciiFalse,indent2)os.replace(temp_file,self.checkpoint_file)# 原子操作defload(self):读取检查点ifnotos.path.exists(self.checkpoint_file):returnNonetry:withopen(self.checkpoint_file,r,encodingutf-8)asf:checkpointjson.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):获取断点恢复信息checkpointself.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_size100): 带断点续跑的批量数据处理 cpmCheckpointManager(batch_data_processing)# 1. 尝试恢复resume_infocpm.get_resume_info()# 2. 读取全部数据dfpd.read_csv(input_file)total_rowslen(df)print(f共{total_rows}行待处理)# 3. 确定起始位置ifresume_info[resumable]:start_indexresume_info[last_position]1print(f从检查点恢复从第{start_index1}行开始已处理{start_index}行)# 加载已有的处理结果ifos.path.exists(output_file):resultspd.read_csv(output_file).to_dict(records)else:results[]else:start_index0results[]# 4. 逐条处理每batch_size条存一次检查点try:foriinrange(start_index,total_rows):rowdf.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(i1)%batch_size0:cpm.save({last_index:i,processed_count:len(results),total_count:total_rows})# 增量保存结果pd.DataFrame(results).to_csv(output_file,indexFalse)progress(i1)/total_rows*100print(f进度{progress:.1f}%已处理{i1}/{total_rows})# 5. 全部完成清除检查点cpm.clear()pd.DataFrame(results).to_csv(output_file,indexFalse)print(f全部完成共处理{total_rows}行)exceptExceptionase:print(f处理中断于第{i1}行{e})print(f已保存检查点下次运行将从第{i1}行继续)# 保存当前进度cpm.save({last_index:i-1,processed_count:len(results)-1,total_count:total_rows})pd.DataFrame(results).to_csv(output_file,indexFalse)raise# 重新抛出异常让上层知道中断了第四步网络请求的重试断点续跑API调用的灾难恢复——指数退避重试 失败队列持久化。importrequestsimporttimeimportrandomclassResilientAPICaller:健壮的API调用器——自动重试失败持久化def__init__(self,checkpoint_dir./checkpoints):self.checkpoint_dircheckpoint_dir self.failed_queue_fileos.path.join(checkpoint_dir,failed_queue.json)os.makedirs(checkpoint_dir,exist_okTrue)defcall_with_retry(self,url,max_retries3,base_delay1,**kwargs): 带指数退避重试的API调用 base_delay: 基础延迟秒数每次重试翻倍 forattemptinrange(max_retries):try:resprequests.post(url,timeout30,**kwargs)resp.raise_for_status()returnresp.json()exceptrequests.exceptions.Timeout:delaybase_delay*(2**attempt)random.uniform(0,1)print(f超时第{attempt1}次重试等待{delay:.1f}秒...)time.sleep(delay)exceptrequests.exceptions.HTTPErrorase:# 4xx错误除了429不重试——重试也没用ifresp.status_code429:# 频率限制retry_afterint(resp.headers.get(Retry-After,30))print(f触发频率限制等待{retry_after}秒...)time.sleep(retry_after)continueelifresp.status_code500:# 服务端错误可以重试delaybase_delay*(2**attempt)print(f服务端错误{resp.status_code}第{attempt1}次重试...)time.sleep(delay)continueelse:raise# 4xx客户端错误不重试# 所有重试都失败raiseException(fAPI调用失败已重试{max_retries}次)defbatch_call(self,items,url,result_handlerNone): 批量API调用失败的进入失败队列 results[]failed[]fori,iteminenumerate(items):try:resultself.call_with_retry(url,jsonitem,max_retries3)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第{i1}条失败{e})# 每50条保存失败队列iflen(failed)0andlen(failed)%500: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_itemsjson.load(f)print(f重试{len(failed_items)}条失败的请求...)remaining[]foriteminfailed_items:try:resultself.call_with_retry(url,jsonitem[input],max_retries2)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_asciiFalse,indent2)第五步邮件群发的「已发送」追踪发200封邮件中间断了哪些发了哪些没发classEmailTracker:邮件发送追踪器def__init__(self,tracker_file./checkpoints/email_sent.json):self.tracker_filetracker_file self.sent_listself._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_asciiFalse,indent2)defclear(self):清空追踪完成后调用self.sent_list[]ifos.path.exists(self.tracker_file):os.remove(self.tracker_file)# 使用trackerEmailTracker()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条就标记文件操作每处理一个文件就标记。TEMU店群如何管理运营坑2检查点文件损坏如果在写入检查点文件的过程中断电文件可能写了一半下次读取时JSON解析失败整个流程无法恢复。解决方法先写临时文件再原子替换os.replace永远不要直接覆盖原文件。坑3数据处理的幂等性问题断点续跑意味着同一条数据可能被处理两遍。如果你的处理逻辑不是幂等的比如给金额1恢复后会出错。解决方法处理逻辑设计成幂等的——用唯一ID做去重或者用「先检查再处理」的模式。坑4业务数据的原子性你在处理一条数据库记录时断电了——记录改了一半。恢复后可能拿到一条脏数据。解决方法关键数据操作尽量用事务数据库事务、API的upsert语义要么全成功要么全回滚。总结长流程必须做灾难恢复设计否则跑一次挂一次。核心三件套检查点机制知道断在哪、幂等处理恢复不重复、失败队列失败的不丢。花一小时加好这些省下的是无数次头疼的重跑。