影刀RPA 流程中的回滚与补偿中间失败后的数据恢复作者林焱什么情况用这个你有一个流程读取Excel中的100条数据→逐条提交到网页→把提交结果写回Excel。跑到第67条的时候网断了网页提交失败流程中断。Excel里前66条的结果已经写了但第67条的状态到底是已提交还是未提交你根本不知道。更糟的是有些操作是不可逆的——你已经删除了原始文件、已经发送了通知邮件、已经扣减了库存。流程中断后留下一地鸡毛你手动恢复可能要花比写流程更长的时间。这就是为什么需要回滚和补偿机制——让流程在出错后要么回到出错前的状态回滚要么把出错的步骤补上补偿。怎么做第一步判断你的操作属于哪种类型不是所有操作都能回滚。先分类操作类型能否回滚示例出错后处理本地文件操作容易修改Excel、创建文件备份→恢复数据库操作容易事务INSERT、UPDATEROLLBACK网页操作困难提交表单、发送数据检查状态→补偿拼多多店群自动化上架方案| 外部通知 | 不可能 | 发送邮件、企微消息 | 记录→手动处理 || API调用 | 看情况 | 创建订单、删除记录 | 调取状态→补偿 |核心原则可回滚的操作先备份再执行不可回滚的操作先检查再执行。第二步本地文件的回滚——备份机制importshutilimportosfromdatetimeimportdatetimeclassFileBackup:文件操作的备份与回滚def__init__(self):self.backups{}# {原文件路径: 备份路径}defbackup(self,file_path):执行操作前备份文件ifnotos.path.exists(file_path):returnNone# 生成备份文件名timestampdatetime.now().strftime(%Y%m%d_%H%M%S)backup_diros.path.join(os.path.dirname(file_path),rpa_backups)os.makedirs(backup_dir,exist_okTrue)backup_pathos.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_pathself.backups[file_path]ifos.path.exists(backup_path):shutil.copy2(backup_path,file_path)returnTruereturnFalsedefcommit(self,file_path):提交确认操作成功删除备份iffile_pathinself.backups:backup_pathself.backups[file_path]ifos.path.exists(backup_path):os.remove(backup_path)delself.backups[file_path]# 使用模式 backup_managerFileBackup()# 1. 操作前备份excel_pathD:/data/库存表.xlsxbackup_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): 在事务中执行一组数据库操作 任何一步失败全部回滚 connsqlite3.connect(db_path)cursorconn.cursor()try:# 开启事务conn.execute(BEGIN TRANSACTION)results[]foropinoperations:cursor.execute(op[sql],op.get(params,()))ifop.get(expect_result):resultcursor.fetchall()results.append(result)# 全部成功提交conn.commit()return{success:True,results:results}exceptExceptionase:# 任何一步失败回滚全部conn.rollback()return{success:False,error:str(e)}finally:conn.close()# 使用示例resultexecute_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 → verifiedtimestamp:datetime.now().isoformat(),result_id:None}self.compensation_queue.append(record)try:# 2. 执行提交result_idsubmit_to_website(data)record[status]submittedrecord[result_id]result_id# 3. 验证提交成功回查确认ifverify_submission(result_id):record[status]verifiedreturn{success:True,id:result_id}else:record[status]unverifiedreturn{success:False,error:提交后验证失败}exceptExceptionase:record[status]failedrecord[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:# 可能已提交但没验证成功 → 检查状态statuscheck_submission_status(record[result_id])ifstatussuccess:record[status]verifiedelifstatusnot_found:# 没提交成功可以重试retry_resultsubmit_to_website(record[data])第五步流程级别的断点续跑在流程层面实现断点续跑比每条操作都做精细回滚更实用importjsonclassCheckpointManager:断点管理器——记录流程执行到哪里了def__init__(self,checkpoint_fileD:/rpa_checkpoint.json):self.checkpoint_filecheckpoint_file self.stateself._load()def_load(self):加载上次的检查点ifos.path.exists(self.checkpoint_file):withopen(self.checkpoint_file,r,encodingutf-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,encodingutf-8)asf:json.dump(self.state,f,ensure_asciiFalse,indent2)defget_resume_point(self):获取续跑起点returnself.state[step]defis_fresh_run(self):是否从头开始returnself.state[step]0# 在影刀流程中使用 ckptCheckpointManager()# 流程开始ifckpt.is_fresh_run():print(全新开始)start_index0else:start_indexckpt.get_resume_point()print(f从第{start_index}条继续)ckpt.save(start_index,statusresuming)# 逐条处理itemsload_items()foriinrange(start_index,len(items)):try:process_item(items[i])ckpt.save(i1,last_processeditems[i][id])exceptExceptionase:print(f第{i}条处理失败已记录断点)ckpt.save(i,errorstr(e),failed_itemitems[i][id])break有什么坑坑一备份文件占用大量磁盘空间现象每次操作前都备份完整文件一个月下来备份目录几十GB。解决只备份被修改的文件不备份整个目录操作成功后清理备份commit设置备份保留天数定期清理坑二事务不支持的数据库操作现象MySQL的某些DDL操作ALTER TABLE、TRUNCATE不能回滚。TEMU店群如何管理运营解决DDL操作前先手动备份数据结构导出建表语句而不是依赖事务回滚。坑三回滚本身也可能失败现象备份文件损坏或路径不存在回滚失败。解决备份成功后验证备份文件完整性检查文件大小和MD5回滚前检查备份是否存在。坑四补偿操作造成双重提交现象网页提交可能已经成功了但网络原因没收到响应补偿重试导致提交了两条相同数据。解决补偿前先查重——用唯一业务ID查询是否已存在defsafe_retry_submit(data,unique_key):重试前先查重避免重复提交existingquery_by_unique_key(unique_key)ifexisting:return{success:True,message:数据已存在跳过,id:existing[id]}returnsubmit_to_website(data)坑五断点记录与实际执行不一致现象断点记录说处理到第68条但第68条其实没处理完从第68条继续会漏掉。解决断点记录的是已完成的最大索引而不是当前处理的索引。只有确认处理完成后才更新断点。总结回滚和补偿不需要一步到位。先做最简单的——流程开始时备份Excel流程结束时验证结果。这一步就能解决50%的数据安全问题。然后再逐步加入数据库事务和断点续跑。你的流程抗风险能力是一点一点垒起来的。
