影刀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_retry3,delay2):安全点击带重试forattemptinrange(1,max_retry1):try:# 在影刀中用【点击】指令# click_element(element_selector)returnTrueexceptExceptionase:ifattemptmax_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,timeout30):等待元素出现并点击starttime.time()whiletime.time()-starttimeout: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# 策略2JS点击try:execute_js(fdocument.querySelector({selector}).click())returnTrueexceptException:pass# 策略3坐标点击try:# get_element_position(selector) → x, y# click_coordinate(x, y)returnTrueexceptException:passreturnFalse2.4 操作级异常处理的适用场景网页元素偶尔加载慢按钮偶尔点击不到网络偶尔不稳定元素偶尔被遮挡三、步骤级异常处理步骤级处理一个完整步骤的异常比如登录“采集一页数据”“处理一个文件”。3.1 步骤级Try-Catchdeflogin_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,timeout10)return{success:True,message:登录成功}exceptElementNotFoundError:return{success:False,message:登录元素未找到可能页面改版}exceptTimeoutError:return{success:False,message:登录超时可能网络问题}exceptExceptionase:return{success:False,message:f登录未知错误{e}}# 使用resultlogin_step()ifnotresult[success]:print(f登录失败{result[message]})# 决定是否重试或跳过3.2 步骤重试defstep_with_retry(step_func,max_retry3,delay10):步骤级重试forattemptinrange(1,max_retry1):resultstep_func()ifresult[success]:returnresultifattemptmax_retry:print(f步骤失败第{attempt}次{result[message]}{delay}秒后重试)time.sleep(delay)else:print(f步骤失败已达最大重试次数)returnresultreturnresult# 使用login_resultstep_with_retry(login_step,max_retry3,delay10)3.3 步骤级降级defcollect_data_step():数据采集步骤带降级方案try:# 主方案直接采集datacollect_from_page()return{success:True,data:data,method:direct}exceptExceptionase:print(f直接采集失败{e}尝试降级方案)try:# 降级方案1通过API获取datacollect_from_api()return{success:True,data:data,method:api}exceptExceptionase2:print(fAPI采集也失败{e2})# 降级方案2使用缓存数据cachedload_cache()ifcached:return{success:True,data:cached,method:cache,warning:使用缓存数据}return{success:False,message:f所有采集方式失败}四、子流程级异常处理子流程级是最粗粒度的异常处理负责整个子流程的容错和恢复。4.1 子流程级Try-Catchdefcollect_site_data(site_config):采集一个网站的完整子流程site_namesite_config[name]try:# 1. 打开网站open_webpage(site_config[url])# 2. 搜索关键词search(site_config[keyword])# 3. 采集数据datacollect_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:[]}# 主流程中遍历多个网站sitesget_variable(site_list)all_results[]forsiteinsites:resultcollect_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_handleNonefile_handleNonetry:web_handleopen_webpage(config[url])file_handleopen(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():完整的三层异常处理流程sitesget_variable(site_list)all_data[]error_log[]# 子流程级遍历多个网站 forsiteinsites:site_namesite[name]site_data[]site_successTruetry:# 步骤级登录 login_resultstep_with_retry(lambda:login_step(site),max_retry3,delay10)ifnotlogin_result[success]:site_successFalseerror_log.append({site:site_name,step:login,error:login_result[message]})continue# 跳过这个网站继续下一个# 步骤级翻页采集 page1whilepage50:# 安全阀try:# 操作级采集单页数据 page_datasafe_collect_page(page)ifnotpage_data:breaksite_data.extend(page_data)page1exceptExceptionase:# 操作级异常跳过这一页继续下一页error_log.append({site:site_name,step:fpage_{page},error:str(e)})page1continue# 步骤级登出 try:logout_step(site)except:pass# 登出失败不影响数据exceptExceptionase:# 子流程级异常整个网站失败site_successFalseerror_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:resultdo_something()except:pass# 出错了也不知道# 好至少记录日志try:resultdo_something()exceptExceptionase:print(f操作失败{e})resultNone# 提供默认值6.3 异常时截图保存现场try:click_element(#submit)exceptExceptionase:# 截图保存timestampdatetime.now().strftime(%Y%m%d_%H%M%S)screenshot_pathfD:\\error_screenshots\\error_{timestamp}.pngtake_screenshot(screenshot_path)# 记录详细错误error_info{time:timestamp,error:str(e),screenshot:screenshot_path,traceback:traceback.format_exc()}log_error(error_info)七、避坑清单坑1except子句顺序错误# 错误Exception在前后面的具体异常永远捕获不到try:...exceptException:...exceptValueError:# 永远不会执行ValueError是Exception的子类...# 正确具体异常在前try:...exceptValueError:...exceptException:...坑2裸except捕获所有异常# 不推荐try:...except:# 捕获所有异常包括KeyboardInterruptpass# 推荐try:...exceptException:# 只捕获普通异常pass坑3finally中的异常掩盖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)# 等待后重试
