企业官网建设流程全解析

影刀RPA 流程中的回滚与补偿：中间失败后的数据恢复

作者：林焱

什么情况用这个

你有一个流程：读取Excel中的100条数据→逐条提交到网页→把提交结果写回Excel。跑到第67条的时候网断了，网页提交失败，流程中断。Excel里前66条的结果已经写了，但第67条的状态到底是"已提交"还是"未提交"？你根本不知道。

更糟的是，有些操作是不可逆的——你已经删除了原始文件、已经发送了通知邮件、已经扣减了库存。流程中断后留下一地鸡毛，你手动恢复可能要花比写流程更长的时间。

这就是为什么需要"回滚"和"补偿"机制——让流程在出错后，要么回到出错前的状态（回滚），要么把出错的步骤补上（补偿）。

怎么做

第一步：判断你的操作属于哪种类型

不是所有操作都能回滚。先分类：

| 外部通知 | 不可能 | 发送邮件、企微消息 | 记录→手动处理 |
| API调用 | 看情况 | 创建订单、删除记录 | 调取状态→补偿 |

核心原则：可回滚的操作先备份再执行；不可回滚的操作先检查再执行。

第二步：本地文件的回滚——备份机制

importshutilimportosfromdatetimeimportdatetimeclassFileBackup:"""文件操作的备份与回滚"""def__init__(self):self.backups={}# {原文件路径: 备份路径}defbackup(self,file_path):"""执行操作前备份文件"""ifnotos.path.exists(file_path):returnNone# 生成备份文件名timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")backup_dir=os.path.join(os.path.dirname(file_path),"rpa_backups")os.makedirs(backup_dir,exist_ok=True)backup_path=os.path.join(backup_dir,f"{os.path.basename(file_path)}.{timestamp}.bak")shutil.copy2(file_path,backup_path)self.backups[file_path]=backup_pathreturnbackup_pathdefrollback(self,file_path):"""回滚：恢复备份"""iffile_pathinself.backups:backup_path=self.backups[file_path]ifos.path.exists(backup_path):shutil.copy2(backup_path,file_path)returnTruereturnFalsedefcommit(self,file_path):"""提交：确认操作成功，删除备份"""iffile_pathinself.backups:backup_path=self.backups[file_path]ifos.path.exists(backup_path):os.remove(backup_path)delself.backups[file_path]# ===== 使用模式 =====backup_manager=FileBackup()# 1. 操作前备份excel_path="D:/data/库存表.xlsx"backup_manager.backup(excel_path)try:# 2. 执行修改操作modify_excel(excel_path,new_data)# 3. 验证修改结果ifverify_excel(excel_path):backup_manager.commit(excel_path)# 成功 → 清理备份else:backup_manager.rollback(excel_path)# 验证失败 → 恢复exceptExceptionase:backup_manager.rollback(excel_path)# 异常 → 恢复raise

第三步：数据库操作的回滚——事务机制

importsqlite3# 或 MySQL: import pymysqldefexecute_with_transaction(db_path,operations):""" 在事务中执行一组数据库操作 任何一步失败，全部回滚 """conn=sqlite3.connect(db_path)cursor=conn.cursor()try:# 开启事务conn.execute("BEGIN TRANSACTION")results=[]foropinoperations:cursor.execute(op["sql"],op.get("params",()))ifop.get("expect_result"):result=cursor.fetchall()results.append(result)# 全部成功，提交conn.commit()return{"success":True,"results":results}exceptExceptionase:# 任何一步失败，回滚全部conn.rollback()return{"success":False,"error":str(e)}finally:conn.close()# 使用示例result=execute_with_transaction("D:/data/business.db",[{"sql":"UPDATE inventory SET stock = stock - ? WHERE sku = ?","params":(10,"SKU001")},{"sql":"INSERT INTO orders (order_id, sku, qty, time) VALUES (?, ?, ?, ?)","params":("ORD001","SKU001",10,"2025-03-15")},{"sql":"INSERT INTO order_log (order_id, action) VALUES (?, ?)","params":("ORD001","created")},])ifnotresult["success"]:print(f"操作失败，已全部回滚:{result['error']}")

第四步：网页操作的补偿机制

网页操作通常不能回滚（你无法"撤回"已提交的表单），需要补偿：

classWebCompensation:"""网页操作的补偿处理器"""def__init__(self):self.compensation_queue=[]# 需要补偿的操作记录defsafe_submit_form(self,data):"""安全提交表单——记录状态以便补偿"""# 1. 提交前记录record={"action":"submit_form","data":data,"status":"pending",# pending → submitted → verified"timestamp":datetime.now().isoformat(),"result_id":None}self.compensation_queue.append(record)try:# 2. 执行提交result_id=submit_to_website(data)record["status"]="submitted"record["result_id"]=result_id# 3. 验证提交成功（回查确认）ifverify_submission(result_id):record["status"]="verified"return{"success":True,"id":result_id}else:record["status"]="unverified"return{"success":False,"error":"提交后验证失败"}exceptExceptionase:record["status"]="failed"record["error"]=str(e)return{"success":False,"error":str(e)}defget_pending_compensations(self):"""获取所有需要补偿的操作"""return[rforrinself.compensation_queueifr["status"]in["pending","failed","unverified"]]defcompensate(self):"""执行补偿：对失败的操作进行重试或撤销"""forrecordinself.get_pending_compensations():ifrecord["status"]=="unverified":# 可能已提交但没验证成功 → 检查状态status=check_submission_status(record["result_id"])ifstatus=="success":record["status"]="verified"elifstatus=="not_found":# 没提交成功，可以重试retry_result=submit_to_website(record["data"])

第五步：流程级别的断点续跑

在流程层面实现"断点续跑"，比每条操作都做精细回滚更实用：

importjsonclassCheckpointManager:"""断点管理器——记录流程执行到哪里了"""def__init__(self,checkpoint_file="D:/rpa_checkpoint.json"):self.checkpoint_file=checkpoint_file self.state=self._load()def_load(self):"""加载上次的检查点"""ifos.path.exists(self.checkpoint_file):withopen(self.checkpoint_file,"r",encoding="utf-8")asf:returnjson.load(f)return{"step":0,"data":{},"logs":[]}defsave(self,step,**data):"""保存当前进度"""self.state["step"]=step self.state["data"].update(data)self.state["logs"].append({"step":step,"time":datetime.now().isoformat(),"data":data})withopen(self.checkpoint_file,"w",encoding="utf-8")asf:json.dump(self.state,f,ensure_ascii=False,indent=2)defget_resume_point(self):"""获取续跑起点"""returnself.state["step"]defis_fresh_run(self):"""是否从头开始"""returnself.state["step"]==0# ===== 在影刀流程中使用 =====ckpt=CheckpointManager()# 流程开始ifckpt.is_fresh_run():print("全新开始")start_index=0else:start_index=ckpt.get_resume_point()print(f"从第{start_index}条继续")ckpt.save(start_index,status="resuming")# 逐条处理items=load_items()foriinrange(start_index,len(items)):try:process_item(items[i])ckpt.save(i+1,last_processed=items[i]["id"])exceptExceptionase:print(f"第{i}条处理失败，已记录断点")ckpt.save(i,error=str(e),failed_item=items[i]["id"])break

有什么坑

坑一：备份文件占用大量磁盘空间

现象：每次操作前都备份完整文件，一个月下来备份目录几十GB。

解决：

• 只备份被修改的文件，不备份整个目录
• 操作成功后清理备份（commit）
• 设置备份保留天数，定期清理

坑二：事务不支持的数据库操作

现象：MySQL的某些DDL操作（ALTER TABLE、TRUNCATE）不能回滚。

解决：DDL操作前先手动备份数据结构（导出建表语句），而不是依赖事务回滚。

坑三：回滚本身也可能失败

现象：备份文件损坏或路径不存在，回滚失败。

解决：备份成功后验证备份文件完整性（检查文件大小和MD5），回滚前检查备份是否存在。

坑四：补偿操作造成双重提交

现象：网页提交可能已经成功了（但网络原因没收到响应），补偿重试导致提交了两条相同数据。

解决：补偿前先查重——用唯一业务ID查询是否已存在：

defsafe_retry_submit(data,unique_key):"""重试前先查重，避免重复提交"""existing=query_by_unique_key(unique_key)ifexisting:return{"success":True,"message":"数据已存在，跳过","id":existing["id"]}returnsubmit_to_website(data)

坑五：断点记录与实际执行不一致

现象：断点记录说处理到第68条，但第68条其实没处理完，从第68条继续会漏掉。

解决：断点记录的是"已完成的最大索引"，而不是"当前处理的索引"。只有确认处理完成后才更新断点。

总结：回滚和补偿不需要一步到位。先做最简单的——流程开始时备份Excel，流程结束时验证结果。这一步就能解决50%的数据安全问题。然后再逐步加入数据库事务和断点续跑。你的流程抗风险能力是一点一点垒起来的。