影刀RPA完全指南_自动化流程的监控告警系统搭建出了问题第一时间知道

影刀RPA完全指南自动化流程的监控告警系统搭建——出了问题第一时间知道你的流程跑了一个月某天早上你打开电脑发现——流程已经挂了三天但谁都不知道。这种静默死亡是RPA运维中最让人头疼的问题。这篇文章解决的就是它如何搭建一套轻量级的监控告警系统让流程出问题时第一时间通知你。要监控什么不是所有东西都需要监控。聚焦最关键的几个指标监控项重要性原因流程是否在运行★★★★★流程崩了等于全白干最后成功时间★★★★★能判断是否假运行处理数据量★★★★数据少了一半说明有问题错误日志★★★★看出错类型和频率运行耗时★★★突然变慢通常是大问题的前兆磁盘空间★★日志把磁盘写满了方案一最简单的心跳检测10分钟搞定如果只做一件事就做心跳检测。原理每个流程在正常运行时每隔一段时间向一个心跳文件写入当前时间戳。监控流程定期检查这个文件如果超过一定时间没有更新就发告警。店群矩阵自动化突破运营极限被监控的流程中在关键步骤后写入心跳importjsonimportosfromdatetimeimportdatetimedefwrite_heartbeat(flow_name):写入心跳heartbeat_filerD:\rpa_monitor\heartbeat.jsonos.makedirs(os.path.dirname(heartbeat_file),exist_okTrue)# 读取现有心跳数据heartbeats{}ifos.path.exists(heartbeat_file):withopen(heartbeat_file,r,encodingutf-8)asf:heartbeatsjson.load(f)# 更新当前流程的心跳heartbeats[flow_name]{last_beat:datetime.now().strftime(%Y-%m-%d %H:%M:%S),status:running}withopen(heartbeat_file,w,encodingutf-8)asf:json.dump(heartbeats,f,ensure_asciiFalse,indent2)监控流程独立运行定时检查importjsonimportosfromdatetimeimportdatetime,timedeltadefcheck_heartbeats(alert_webhook,timeout_minutes30): 检查心跳状态 timeout_minutes: 超过这个时间没有心跳就告警 heartbeat_filerD:\rpa_monitor\heartbeat.jsonifnotos.path.exists(heartbeat_file):send_alert(alert_webhook,心跳文件不存在可能所有流程都未启动)return[]withopen(heartbeat_file,r,encodingutf-8)asf:heartbeatsjson.load(f)alerts[]nowdatetime.now()forflow_name,infoinheartbeats.items():last_beatdatetime.strptime(info[last_beat],%Y-%m-%d %H:%M:%S)minutes_since(now-last_beat).total_seconds()/60ifminutes_sincetimeout_minutes:alert_msgf⚠️ 流程疑似停止{flow_name}上次心跳{last_beat}已超过{minutes_since:.0f}分钟alerts.append(alert_msg)ifalerts:full_msg【RPA心跳告警】\n\n.join(alerts)send_alert(alert_webhook,full_msg)returnalerts方案二结构化日志 自动分析心跳只能告诉流程死了但不能告诉为什么死。统一的日志格式所有流程使用同一个日志格式importjsonimportosfromdatetimeimportdatetimeclassFlowLogger:统一的流程日志记录器def__init__(self,flow_name,log_dirrD:\rpa_logs):self.flow_nameflow_name self.log_dirlog_dir os.makedirs(log_dir,exist_okTrue)self.log_fileos.path.join(log_dir,f{flow_name}_{datetime.now().strftime(%Y%m%d)}.jsonl)deflog(self,level,message,extraNone):写入一条日志entry{timestamp:datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)[:-3],flow:self.flow_name,level:level,# INFO / WARN / ERRORmessage:message}ifextra:entry[extra]extrawithopen(self.log_file,a,encodingutf-8)asf:f.write(json.dumps(entry,ensure_asciiFalse)\n)# 在流程中使用loggerFlowLogger(pdd_price_monitor)# 流程各节点记录logger.log(INFO,流程启动)logger.log(INFO,登录成功,{account:tb_user_001})logger.log(INFO,开始采集商品列表,{page:1})logger.log(WARN,第3页加载超时重试中,{retry_count:2})logger.log(INFO,采集完成,{total_items:450,duration_sec:320})logger.log(ERROR,写入Excel失败,{error:Permission denied})日志分析脚本监控流程中每日运行importjsonimportosfromdatetimeimportdatetime,timedeltafromcollectionsimportCounterdefanalyze_logs(log_dir,target_dateNone): 分析指定日期的日志生成健康报告 iftarget_dateisNone:target_datedatetime.now().strftime(%Y%m%d)errors[]warnings[]flow_stats{}forfilenameinos.listdir(log_dir):ifnotfilename.endswith(.jsonl)ortarget_datenotinfilename:continueflow_namefilename.replace(f_{target_date}.jsonl,)flow_errors0withopen(os.path.join(log_dir,filename),r,encodingutf-8)asf:forlineinf:entryjson.loads(line.strip())ifentry[level]ERROR:flow_errors1errors.append({flow:flow_name,time:entry[timestamp],message:entry[message]})elifentry[level]WARN:warnings.append({flow:flow_name,message:entry[message]})flow_stats[flow_name]{errors:flow_errors,status:异常ifflow_errors0else正常}return{date:target_date,flow_stats:flow_stats,error_count:len(errors),errors:errors[:10],# 只取前10条warning_count:len(warnings)}# 每日运行分析reportanalyze_logs(rD:\rpa_logs)ifreport[error_count]0:# 发送告警alert_contentf今日共{report[error_count]}个错误\nforerrinreport[errors]:alert_contentf • [{err[flow]}]{err[message]}\nsend_alert(webhook,alert_content)方案三监控仪表盘进阶如果团队有技术人员可以搭一个简单的Web监控页面。技术栈后端Python Flask轻量Web框架前端简单的HTML 定时刷新数据源上面的JSONL日志文件后端APIFlask在云服务器上运行fromflaskimportFlask,jsonifyimportos,json appFlask(__name__)app.route(/api/status)defget_status():返回所有流程的状态log_dirrD:\rpa_logsflows_status{}forfilenameinos.listdir(log_dir):ifnotfilename.endswith(.jsonl):continueflow_namefilename.rsplit(_,1)[0]# 读取最后一行判断状态withopen(os.path.join(log_dir,filename),r,encodingutf-8)asf:linesf.readlines()last_entryjson.loads(lines[-1])iflineselseNoneflows_status[flow_name]{last_activity:last_entry[timestamp]iflast_entryelse未知,last_message:last_entry[message]iflast_entryelse无记录,is_running:完成in(last_entry[message]iflast_entryelse)}returnjsonify(flows_status)if__name____main__:app.run(host0.0.0.0,port5000)前端页面简单的状态面板!DOCTYPEhtmlhtmlheadmetacharsetUTF-8titleRPA监控面板/titlemetahttp-equivrefreshcontent30stylebody{font-family:sans-serif;padding:20px;background:#f5f5f5;}.card{background:white;padding:15px;margin:10px 0;border-radius:8px;box-shadow:0 1px 3pxrgba(0,0,0,0.1);}.ok{border-left:4px solid #4CAF50;}.warn{border-left:4px solid #FF9800;}.error{border-left:4px solid #F44336;}/style/headbodyh1 RPA流程监控面板/h1dividstatus/divscriptfetch(/api/status)[video(video-VoJs8fu6-1781253084616)(type-csdn)(url-https://live.csdn.net/v/embed/524992)(image-https://v-blog.csdnimg.cn/asset/b59aed2f01d4fe8583467562aaf4dcfd/cover/Cover0.jpg)(title-temu店群自动化报活动案例)].then(rr.json()).then(data{lethtml;for(let[name,info]ofObject.entries(data)){letclsinfo.is_running?ok:warn;htmldiv classcard${cls} strong${name}/strongbr 最后活动${info.last_activity}br${info.last_message}/div;}document.getElementById(status).innerHTMLhtml;});/script/body/html告警渠道的选择渠道适用场景延迟企业微信群机器人日常通知、日报秒级钉钉群机器人同上秒级飞书群机器人同上秒级邮件重要告警备份分钟级短信深夜紧急告警秒级需要短信API推荐组合IM群消息日常 邮件备份 短信只在凌晨的紧急告警。写在最后监控告警的核心原则宁可多报一千不可漏报一次。尤其是线上运行的流程它不报错不代表它在正常工作。它可能在假运行——卡在某个循环里不报错也不产出数据。所以心跳检测 日志分析的组合是基础监控仪表盘是锦上添花。从最简单的开始先保证你能在30分钟内知道流程出了问题再慢慢丰富监控维度。内容标签影刀RPA · 监控告警 · 心跳检测 · 日志分析 · 运维 · 仪表盘作者林焱系列说明本文是「影刀RPA新手到高手」系列教程之一面向需要运维多个自动化流程的进阶用户。