影刀RPA 异常处理的分层设计：子流程级、步骤级、操作级的三层防线
作者：林焱
异常处理是RPA流程稳定性的最后一道防线。没有异常处理的流程，一个报错就全盘崩溃；异常处理太粗的流程，错误信息被吞掉，出问题了都不知道哪里出的问题。好的异常处理应该分层设计——子流程级、步骤级、操作级，每一层负责不同粒度的错误捕获和恢复。
这篇文章把三层异常防线的设计思路和实现方法讲清楚。
一、为什么需要分层异常处理
1.1 单层异常处理的问题
很多新手的做法是：在整个流程最外层套一个大Try-Catch。这样做的问题：
# 粗放的异常处理try:open_website()login()collect_data()process_data()save_result()send_email()exceptExceptionase:print(f"出错了：{e}")问题：
- 不知道哪一步出错
- 出错后没有恢复机制，整个流程白跑
- 错误信息太笼统，排查困难
- 一步失败导致后续所有步骤都不执行
1.2 分层异常处理的优势
子流程级：采集数据子流程失败 → 跳过这个网站，继续下一个 ↓ 步骤级：登录步骤失败 → 重试3次，仍失败则跳过 ↓ 操作级：点击按钮失败 → 等待重试，或用JS点击每一层处理不同级别的异常，流程能恢复的就恢复，不能恢复的优雅降级。
二、操作级异常处理
拼多多店群自动化报活动上架！
操作级是最细粒度的异常处理，针对单个操作（如点击、填写、读取）做容错。
2.1 单操作重试
importtimedefsafe_click(element_selector,max_retry=3,delay=2):"""安全点击，带重试"""forattemptinrange(1,max_retry+1):try:# 在影刀中用【点击】指令# click_element(element_selector)returnTrueexceptExceptionase:ifattempt<max_retry:print(f"点击失败（第{attempt}次）：{e}，{delay}秒后重试")time.sleep(delay)else:print(f"点击失败，已达最大重试次数{max_retry}")returnFalse# 使用ifnotsafe_click("#submit-btn"):# 点击失败后的降级方案safe_click("#fallback-btn")2.2 等待元素出现的容错
defwait_and_click(selector,timeout=30):"""等待元素出现并点击"""start=time.time()whiletime.time()-start<timeout:try:ifcheck_element_exists(selector):click_element(selector)returnTrueexceptException:passtime.sleep(1)print(f"等待元素超时：{selector}")returnFalse2.3 多种操作方式降级
defrobust_click(selector):"""多策略点击"""# 策略1：直接点击try:click_element(selector)returnTrueexceptException:pass# 策略2：JS点击try:execute_js(f"document.querySelector('{selector}').click()")returnTrueexceptException:pass# 策略3：坐标点击try:# get_element_position(selector) → x, y# click_coordinate(x, y)returnTrueexceptException:passreturnFalse2.4 操作级异常处理的适用场景
网页元素偶尔加载慢
按钮偶尔点击不到
网络偶尔不稳定
元素偶尔被遮挡
三、步骤级异常处理
步骤级处理一个完整步骤的异常，比如"登录"“采集一页数据”“处理一个文件”。
3.1 步骤级Try-Catch
deflogin_step():"""登录步骤，带完整异常处理"""try:# 1. 打开登录页open_webpage("https://example.com/login")# 2. 填写用户名fill_input("#username",get_variable("username"))# 3. 填写密码fill_input("#password",get_variable("password"))# 4. 点击登录click_element("#login-btn")# 5. 验证登录成功wait_for_element(".user-avatar",timeout=10)return{"success":True,"message":"登录成功"}exceptElementNotFoundError:return{"success":False,"message":"登录元素未找到，可能页面改版"}exceptTimeoutError:return{"success":False,"message":"登录超时，可能网络问题"}exceptExceptionase:return{"success":False,"message":f"登录未知错误：{e}"}# 使用result=login_step()ifnotresult["success"]:print(f"登录失败：{result['message']}")# 决定是否重试或跳过3.2 步骤重试
defstep_with_retry(step_func,max_retry=3,delay=10):"""步骤级重试"""forattemptinrange(1,max_retry+1):result=step_func()ifresult["success"]:returnresultifattempt<max_retry:print(f"步骤失败（第{attempt}次）：{result['message']}，{delay}秒后重试")time.sleep(delay)else:print(f"步骤失败，已达最大重试次数")returnresultreturnresult# 使用login_result=step_with_retry(login_step,max_retry=3,delay=10)3.3 步骤级降级
defcollect_data_step():"""数据采集步骤，带降级方案"""try:# 主方案：直接采集data=collect_from_page()return{"success":True,"data":data,"method":"direct"}exceptExceptionase:print(f"直接采集失败：{e}，尝试降级方案")try:# 降级方案1：通过API获取data=collect_from_api()return{"success":True,"data":data,"method":"api"}exceptExceptionase2:print(f"API采集也失败：{e2}")# 降级方案2：使用缓存数据cached=load_cache()ifcached:return{"success":True,"data":cached,"method":"cache","warning":"使用缓存数据"}return{"success":False,"message":f"所有采集方式失败"}四、子流程级异常处理
子流程级是最粗粒度的异常处理，负责整个子流程的容错和恢复。
4.1 子流程级Try-Catch
defcollect_site_data(site_config):"""采集一个网站的完整子流程"""site_name=site_config["name"]try:# 1. 打开网站open_webpage(site_config["url"])# 2. 搜索关键词search(site_config["keyword"])# 3. 采集数据data=collect_results()# 4. 关闭网页close_webpage()return{"site":site_name,"success":True,"data_count":len(data),"data":data}exceptExceptionase:# 子流程级异常：整个网站采集失败print(f"网站{site_name}采集失败：{e}")# 确保关闭网页try:close_webpage()except:passreturn{"site":site_name,"success":False,"error":str(e),"data_count":0,"data":[]}# 主流程中遍历多个网站sites=get_variable("site_list")all_results=[]forsiteinsites:result=collect_site_data(site)all_results.append(result)ifresult["success"]:print(f"✓{result['site']}：采集{result['data_count']}条")else:print(f"✗{result['site']}：{result['error']}")# 即使某个网站失败，其他网站的数据仍然可用4.2 子流程级异常的关键原则
原则1：隔离故障
一个子流程的异常不应该影响其他子流程。用Try-Catch包裹每个子流程，确保异常被捕获。
原则2：记录足够信息
异常发生时记录：哪个子流程出错、什么错误、当时的变量值、截图。这些信息是排查问题的关键。
原则3：清理资源
异常发生后确保释放资源（关闭网页、关闭文件、断开连接）。用finally块或Try-Catch嵌套。
defsafe_subprocess(config):web_handle=Nonefile_handle=Nonetry:web_handle=open_webpage(config["url"])file_handle=open(config["output"],"w")# ... 处理逻辑 ...return{"success":True}exceptExceptionase:return{"success":False,"error":str(e)}finally:# 无论成功失败都清理资源ifweb_handle:try:close_webpage(web_handle)except:passiffile_handle:try:file_handle.close()except:pass五、三层异常处理的完整模板
TEMU店群矩阵自动化运营核价报活动
importtimeimporttracebackfromdatetimeimportdatetimedefrun_collection_pipeline():"""完整的三层异常处理流程"""sites=get_variable("site_list")all_data=[]error_log=[]# ===== 子流程级：遍历多个网站 =====forsiteinsites:site_name=site["name"]site_data=[]site_success=Truetry:# ===== 步骤级：登录 =====login_result=step_with_retry(lambda:login_step(site),max_retry=3,delay=10)ifnotlogin_result["success"]:site_success=Falseerror_log.append({"site":site_name,"step":"login","error":login_result["message"]})continue# 跳过这个网站，继续下一个# ===== 步骤级：翻页采集 =====page=1whilepage<=50:# 安全阀try:# ===== 操作级：采集单页数据 =====page_data=safe_collect_page(page)ifnotpage_data:breaksite_data.extend(page_data)page+=1exceptExceptionase:# 操作级异常：跳过这一页，继续下一页error_log.append({"site":site_name,"step":f"page_{page}","error":str(e)})page+=1continue# ===== 步骤级：登出 =====try:logout_step(site)except:pass# 登出失败不影响数据exceptExceptionase:# 子流程级异常：整个网站失败site_success=Falseerror_log.append({"site":site_name,"step":"subprocess","error":str(e),"traceback":traceback.format_exc()})finally:# 清理：确保关闭网页try:close_all_webpages()except:passall_data.extend(site_data)status="✓"ifsite_successelse"✗"print(f"{status}{site_name}:{len(site_data)}条")# 汇总print(f"\n总计采集{len(all_data)}条数据")iferror_log:print(f"错误{len(error_log)}个：")forerrinerror_log:print(f"{err['site']}-{err['step']}:{err['error']}")set_variable("all_data",all_data)set_variable("error_log",error_log)set_variable("error_count",len(error_log))returnall_data六、异常处理的最佳实践
6.1 捕获具体异常
# 不推荐：捕获所有异常try:...exceptException:pass# 吞掉所有错误# 推荐：捕获具体异常try:...exceptFileNotFoundError:print("文件不存在")exceptPermissionError:print("权限不足")exceptValueError:print("数据格式错误")exceptExceptionase:print(f"未知错误：{e}")raise# 重新抛出，让上层处理6.2 不要吞掉异常
# 坏：完全吞掉异常try:result=do_something()except:pass# 出错了也不知道# 好：至少记录日志try:result=do_something()exceptExceptionase:print(f"操作失败：{e}")result=None# 提供默认值6.3 异常时截图保存现场
try:click_element("#submit")exceptExceptionase:# 截图保存timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")screenshot_path=f"D:\\error_screenshots\\error_{timestamp}.png"take_screenshot(screenshot_path)# 记录详细错误error_info={"time":timestamp,"error":str(e),"screenshot":screenshot_path,"traceback":traceback.format_exc()}log_error(error_info)七、避坑清单
坑1：except子句顺序错误
# 错误：Exception在前，后面的具体异常永远捕获不到try:...exceptException:...exceptValueError:# 永远不会执行（ValueError是Exception的子类）...# 正确：具体异常在前try:...exceptValueError:...exceptException:...坑2：裸except捕获所有异常
# 不推荐try:...except:# 捕获所有异常包括KeyboardInterruptpass# 推荐try:...exceptException:# 只捕获普通异常pass坑3：finally中的异常掩盖try中的异常
try:1/0# ZeroDivisionErrorfinally:do_cleanup()# 如果这里也抛异常，原来的ZeroDivisionError被掩盖# 安全的finallytry:1/0finally:try:do_cleanup()exceptException:pass坑4：异常处理范围太大
# 不推荐：整个流程一个大trytry:step1()step2()step3()step4()exceptException:pass# 不知道哪步出错# 推荐：每步单独tryforstepin[step1,step2,step3,step4]:try:step()exceptExceptionase:print(f"{step.__name__}失败：{e}")坑5：重试不等待
# 错误：立即重试，可能还是失败foriinrange(3):try:do_something()breakexcept:continue# 立即重试# 正确：重试前等待foriinrange(3):try:do_something()breakexcept:time.sleep(5)# 等待后重试
 函数进阶用法与性能优化（头歌云课）)
 agent和组件)