影刀RPA 流程健康监控自动检测机器人是否正常运行作者林焱一、什么情况需要监控自动化流程本身你花了几周时间搭建了一套自动化体系每天定时运行20个流程。起初没问题但某天业务系统悄悄改版了一个流程失败了但没有人知道——直到三天后有人发现数据没更新才开始排查。这就是监控盲区问题。RPA流程一旦部署就容易被遗忘但它并不像人一样会主动喊我坏了。你需要搭建一套流程健康监控机制主动检测流程是否按时运行、是否有错误、结果是否符合预期。拼多多店群自动化上架方案二、实战一流程运行状态日报每天早上7点汇总前一天所有流程的运行结果。Step 1从影刀云服务器或本地日志读取运行记录importglobimportjsonimportos# 影刀日志目录Windows路径log_dirC:\\Users\\YingdaoRPA\\logsyesterday(datetime.now()-timedelta(days1)).strftime(%Y-%m-%d)# 读取昨天的所有流程日志log_filesglob.glob(f{log_dir}\\{yesterday}\\*.log)flow_results[]forlog_fileinlog_files:flow_nameos.path.basename(log_file).replace(.log,)withopen(log_file,r,encodingutf-8)asf:contentf.read()# 解析日志状态if流程执行成功incontentorSUCCESSincontent:status✅ 成功elif流程执行失败incontentorERRORincontentorExceptionincontent:status❌ 失败# 提取错误信息error_lines[lforlincontent.split(\n)ifERRORinlorExceptioninl]error_msgerror_lines[0][:100]iferror_lineselse未知错误else:status⚠️ 状态未知error_msg日志不完整# 提取运行时长start_matchre.search(r开始时间(\d{2}:\d{2}:\d{2}),content)end_matchre.search(r结束时间(\d{2}:\d{2}:\d{2}),content)duration-ifstart_matchandend_match:durationend_match.group(1)# 简化实际应计算差值flow_results.append({流程名称:flow_name,运行状态:status,结束时间:duration,错误信息:error_msgif失败instatuselse-})dfpd.DataFrame(flow_results)success_ratelen(df[df[运行状态].str.contains(成功)])/len(df)*100Step 2生成日报并发送# 生成文本汇总report_lines[f 流程健康日报{yesterday},f总流程数{len(df)}成功率{success_rate:.0f}%,,失败流程ifany(失败insforsindf[运行状态])else✅ 所有流程正常,]for_,rowindf[df[运行状态].str.contains(失败)].iterrows():report_lines.append(f ❌{row[流程名称]}:{row[错误信息]})report_text\n.join(report_lines)yingdao.send_dingtalk_message(report_text,group自动化运维群)df.to_excel(fC:\\监控\\日报_{yesterday}.xlsx,indexFalse)三、实战二流程结果合理性校验不只是检查流程有没有报错还要验证输出结果是否合理。# 校验规则配置validation_rules{每日数据采集:{output_file:C:\\数据\\今日采集.xlsx,min_rows:100,# 至少100行数据max_rows:5000,# 不超过5000行check_columns:[日期,商品名称,价格],# 必须有这些列},发票处理流程:{output_file:C:\\发票\\今日已处理.xlsx,min_rows:1,required_value:{处理状态:已完成},# 所有行的某列必须是特定值}}anomalies[]forflow_name,rulesinvalidation_rules.items():try:dfpd.read_excel(rules[output_file])# 检查行数iflen(df)rules.get(min_rows,0):anomalies.append(f⚠️{flow_name}输出数据太少{len(df)}行期望≥{rules[min_rows]})iflen(df)rules.get(max_rows,float(inf)):anomalies.append(f⚠️{flow_name}输出数据异常多{len(df)}行期望≤{rules[max_rows]})# 检查必要列forcolinrules.get(check_columns,[]):ifcolnotindf.columns:anomalies.append(f⚠️{flow_name}缺少列 {col})exceptFileNotFoundError:anomalies.append(f❌{flow_name}输出文件不存在流程可能未运行)exceptExceptionase:anomalies.append(f❌{flow_name}文件读取失败 -{str(e)})ifanomalies:alert_msg 流程结果异常告警\n\n.join(anomalies)yingdao.send_dingtalk_message(alert_msg,group自动化运维群,at_allTrue)四、实战三流程自愈机制部分流程失败后可以自动尝试重跑。TEMU店群如何管理运营defrun_with_retry(flow_func,max_retries3,retry_interval300): 带重试的流程执行器 max_retries: 最大重试次数 retry_interval: 重试间隔秒 forattemptinrange(1,max_retries1):try:flow_func()print(f✅ 流程执行成功第{attempt}次尝试)returnTrueexceptExceptionase:print(f❌ 第{attempt}次执行失败{e})ifattemptmax_retries:print(f⏳{retry_interval}秒后重试...)time.sleep(retry_interval)else:print(f 已达到最大重试次数放弃执行)# 发送告警yingdao.send_dingtalk_message(f 流程彻底失败已重试{max_retries}次{str(e)[:100]},group自动化运维群)returnFalse# 使用示例run_with_retry(lambda:yingdao.run_flow(每日数据采集),max_retries3,retry_interval300)五、踩坑记录坑1日志格式不统一不同流程的日志格式千奇百怪解析困难。从一开始就统一日志格式——[时间] [级别] [流程名] 消息解析时用统一的正则。坑2监控流程本身也可能失败监控流程是关键流程也要被监控。解决方案给监控流程加一个心跳——每天检查监控流程最后一次成功运行时间超过25小时没运行则短信告警。坑3告警太多会被无视如果每天发30条告警运维人员会开始忽视。原则只在状态从正常变异常时告警稳定异常不重复告警只在恢复正常时再通知一次。坑4文件时间戳不等于流程实际运行时间判断流程是否今天运行不要只看输出文件的修改时间——文件可能被意外复制或备份。应从流程运行日志里读取时间戳。适用人群RPA运维工程师、IT运营、自动化负责人难度⭐⭐⭐需要一定的运维思维