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

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

你的流程跑了一个月，某天早上你打开电脑发现——流程已经挂了三天，但谁都不知道。

这种"静默死亡"是RPA运维中最让人头疼的问题。这篇文章解决的就是它：如何搭建一套轻量级的监控告警系统，让流程出问题时第一时间通知你。

要监控什么

不是所有东西都需要监控。聚焦最关键的几个指标：

方案一：最简单的心跳检测（10分钟搞定）

如果只做一件事，就做心跳检测。

原理

每个流程在正常运行时，每隔一段时间向一个"心跳文件"写入当前时间戳。监控流程定期检查这个文件，如果超过一定时间没有更新，就发告警。

被监控的流程中（在关键步骤后写入心跳）

importjsonimportosfromdatetimeimportdatetimedefwrite_heartbeat(flow_name):"""写入心跳"""heartbeat_file=r"D:\rpa_monitor\heartbeat.json"os.makedirs(os.path.dirname(heartbeat_file),exist_ok=True)# 读取现有心跳数据heartbeats={}ifos.path.exists(heartbeat_file):withopen(heartbeat_file,'r',encoding='utf-8')asf:heartbeats=json.load(f)# 更新当前流程的心跳heartbeats[flow_name]={"last_beat":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"status":"running"}withopen(heartbeat_file,'w',encoding='utf-8')asf:json.dump(heartbeats,f,ensure_ascii=False,indent=2)

监控流程（独立运行，定时检查）

importjsonimportosfromdatetimeimportdatetime,timedeltadefcheck_heartbeats(alert_webhook,timeout_minutes=30):""" 检查心跳状态 timeout_minutes: 超过这个时间没有心跳就告警 """heartbeat_file=r"D:\rpa_monitor\heartbeat.json"ifnotos.path.exists(heartbeat_file):send_alert(alert_webhook,"心跳文件不存在，可能所有流程都未启动")return[]withopen(heartbeat_file,'r',encoding='utf-8')asf:heartbeats=json.load(f)alerts=[]now=datetime.now()forflow_name,infoinheartbeats.items():last_beat=datetime.strptime(info["last_beat"],"%Y-%m-%d %H:%M:%S")minutes_since=(now-last_beat).total_seconds()/60ifminutes_since>timeout_minutes:alert_msg=f"⚠️ 流程疑似停止：{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_dir=r"D:\rpa_logs"):self.flow_name=flow_name self.log_dir=log_dir os.makedirs(log_dir,exist_ok=True)self.log_file=os.path.join(log_dir,f"{flow_name}_{datetime.now().strftime('%Y%m%d')}.jsonl")deflog(self,level,message,extra=None):"""写入一条日志"""entry={"timestamp":datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],"flow":self.flow_name,![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/96de818205114e09955cab301688408c.png#pic_center)"level":level,# INFO / WARN / ERROR"message":message}ifextra:entry["extra"]=extrawithopen(self.log_file,'a',encoding='utf-8')asf:f.write(json.dumps(entry,ensure_ascii=False)+'\n')# 在流程中使用logger=FlowLogger("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_date=None):""" 分析指定日期的日志，生成健康报告 """iftarget_dateisNone:target_date=datetime.now().strftime("%Y%m%d")errors=[]warnings=[]flow_stats={}forfilenameinos.listdir(log_dir):ifnotfilename.endswith('.jsonl')ortarget_datenotinfilename:continueflow_name=filename.replace(f'_{target_date}.jsonl','')flow_errors=0withopen(os.path.join(log_dir,filename),'r',encoding='utf-8')asf:forlineinf:entry=json.loads(line.strip())ifentry['level']=='ERROR':flow_errors+=1errors.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_errors>0else"正常"}return{"date":target_date,"flow_stats":flow_stats,"error_count":len(errors),"errors":errors[:10],# 只取前10条"warning_count":len(warnings)}# 每日运行分析report=analyze_logs(r"D:\rpa_logs")ifreport["error_count"]>0:# 发送告警alert_content=f"今日共{report['error_count']}个错误\n"forerrinreport["errors"]:alert_content+=f" • [{err['flow']}]{err['message']}\n"send_alert(webhook,alert_content)

方案三：监控仪表盘（进阶）

如果团队有技术人员，可以搭一个简单的Web监控页面。

技术栈

• 后端：Python Flask（轻量Web框架）
• 前端：简单的HTML + 定时刷新
• 数据源：上面的JSONL日志文件

后端API（Flask，在云服务器上运行）

fromflaskimportFlask,jsonifyimportos,json app=Flask(__name__)@app.route('/api/status')defget_status():"""返回所有流程的状态"""log_dir=r"D:\rpa_logs"flows_status={}forfilenameinos.listdir(log_dir):ifnotfilename.endswith('.jsonl'):continueflow_name=filename.rsplit('_',1)[0]# 读取最后一行判断状态withopen(os.path.join(log_dir,filename),'r',encoding='utf-8')asf:lines=f.readlines()last_entry=json.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(host='0.0.0.0',port=5000)

前端页面（简单的状态面板）

<!DOCTYPEhtml><html><head><metacharset="UTF-8"><title>RPA监控面板</title><metahttp-equiv="refresh"content="30"><style>body{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></head><body><h1>📊 RPA流程监控面板</h1><divid="status"></div><script>fetch('/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(r=>r.json()).then(data=>{lethtml='';for(let[name,info]ofObject.entries(data)){letcls=info.is_running?'ok':'warn';html+=`<div class="card${cls}"> <strong>${name}</strong><br> 最后活动：${info.last_activity}<br>${info.last_message}</div>`;}document.getElementById('status').innerHTML=html;});</script></body></html>

告警渠道的选择

推荐组合：IM群消息（日常）+ 邮件（备份）+ 短信（只在凌晨的紧急告警）。

写在最后

监控告警的核心原则：宁可多报一千，不可漏报一次。

尤其是线上运行的流程，它不报错不代表它在正常工作。它可能在"假运行"——卡在某个循环里，不报错也不产出数据。

所以心跳检测 + 日志分析的组合是基础，监控仪表盘是锦上添花。从最简单的开始，先保证你能在30分钟内知道流程出了问题，再慢慢丰富监控维度。

内容标签：影刀RPA · 监控告警 · 心跳检测 · 日志分析 · 运维 · 仪表盘

作者：林焱

系列说明：本文是「影刀RPA新手到高手」系列教程之一，面向需要运维多个自动化流程的进阶用户。