影刀RPA 流程健康监控：自动检测机器人是否正常运行

作者：林焱

---

一、什么情况需要监控自动化流程本身

你花了几周时间搭建了一套自动化体系，每天定时运行20个流程。起初没问题，但某天业务系统悄悄改版了，一个流程失败了，但没有人知道——直到三天后有人发现数据没更新，才开始排查。

这就是"监控盲区"问题。RPA流程一旦部署就容易被遗忘，但它并不像人一样会主动喊"我坏了"。你需要搭建一套流程健康监控机制：主动检测流程是否按时运行、是否有错误、结果是否符合预期。

---

二、实战一：流程运行状态日报

每天早上7点，汇总前一天所有流程的运行结果。

Step 1：从影刀云服务器（或本地日志）读取运行记录

import glob
import json
import os

# 影刀日志目录（Windows路径）
log_dir = 'C:\\Users\\YingdaoRPA\\logs'
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')

# 读取昨天的所有流程日志
log_files = glob.glob(f'{log_dir}\\{yesterday}\\*.log')

flow_results = []

for log_file in log_files:
    flow_name = os.path.basename(log_file).replace('.log', '')
    
    with open(log_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 解析日志状态
    if '流程执行成功' in content or 'SUCCESS' in content:
        status = '✅ 成功'
    elif '流程执行失败' in content or 'ERROR' in content or 'Exception' in content:
        status = '❌ 失败'
        # 提取错误信息
        ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/d2f47139e0af4ba2b8a92102cb7d1753.png#pic_center)

        error_lines = [l for l in content.split('\n') if 'ERROR' in l or 'Exception' in l]
        error_msg = error_lines[0][:100] if error_lines else '未知错误'
    else:
        status = '⚠️ 状态未知'
        error_msg = '日志不完整'
    
    # 提取运行时长
    start_match = re.search(r'开始时间：(\d{2}:\d{2}:\d{2})', content)
    end_match = re.search(r'结束时间：(\d{2}:\d{2}:\d{2})', content)
    duration = '-'
    if start_match and end_match:
        duration = end_match.group(1)  # 简化，实际应计算差值
    
    flow_results.append({
        '流程名称': flow_name,
        '运行状态': status,
        '结束时间': duration,
        '错误信息': error_msg if '失败' in status else '-'
    })

df = pd.DataFrame(flow_results)
success_rate = len(df[df['运行状态'].str.contains('成功')]) / len(df) * 100

Step 2：生成日报并发送

# 生成文本汇总
report_lines = [
    f"📊 流程健康日报 {yesterday}",
    f"总流程数：{len(df)}  成功率：{success_rate:.0f}%",
    "",
    "失败流程：" if any('失败' in s for s in df['运行状态']) else "✅ 所有流程正常",
]

for _, row in df[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(f'C:\\监控\\日报_{yesterday}.xlsx', index=False)

---

三、实战二：流程结果合理性校验

不只是检查流程有没有报错，还要验证输出结果是否合理。

# 校验规则配置
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 = []

for flow_name, rules in validation_rules.items():
    try:
        df = pd.read_excel(rules['output_file'])
        
        # 检查行数
        if len(df) < rules.get('min_rows', 0):
            anomalies.append(f"⚠️ {flow_name}：输出数据太少（{len(df)}行，期望≥{rules['min_rows']}）")
        
        if len(df) > rules.get('max_rows', float('inf')):
            anomalies.append(f"⚠️ {flow_name}：输出数据异常多（{len(df)}行，期望≤{rules['max_rows']}）")
        
        # 检查必要列
        for col in rules.get('check_columns', []):
            if col not in df.columns:
                anomalies.append(f"⚠️ {flow_name}：缺少列 '{col}'")
        
    except FileNotFoundError:
        anomalies.append(f"❌ {flow_name}：输出文件不存在，流程可能未运行")
    except Exception as e:
        anomalies.append(f"❌ {flow_name}：文件读取失败 - {str(e)}")

if anomalies:
    alert_msg = "🚨 流程结果异常告警：\n" + '\n'.join(anomalies)
    yingdao.send_dingtalk_message(alert_msg, group='自动化运维群', at_all=True)

---

四、实战三：流程自愈机制

部分流程失败后，可以自动尝试重跑。

def run_with_retry(flow_func, max_retries=3, retry_interval=300):
    """
    带重试的流程执行器
    max_retries: 最大重试次数
    retry_interval: 重试间隔（秒）
    """
    for attempt in range(1, max_retries + 1):
        try:
            flow_func()
            print(f"✅ 流程执行成功（第{attempt}次尝试）")
            return True
        except Exception as e:
            print(f"❌ 第{attempt}次执行失败：{e}")
            if attempt < max_retries:
                print(f"⏳ {retry_interval}秒后重试...")
                time.sleep(retry_interval)
            else:
                print(f"🔴 已达到最大重试次数，放弃执行")
                # 发送告警
                yingdao.send_dingtalk_message(
                    f"🔴 流程彻底失败，已重试{max_retries}次：{str(e)[:100]}",
                    group='自动化运维群'
                )
                return False

# 使用示例
run_with_retry(lambda: yingdao.run_flow('每日数据采集'), max_retries=3, retry_interval=300)

---

五、踩坑记录

坑1：日志格式不统一
不同流程的日志格式千奇百怪，解析困难。从一开始就统一日志格式——[时间] [级别] [流程名] 消息，解析时用统一的正则。

坑2：监控流程本身也可能失败
监控流程是关键流程，也要被监控。解决方案：给监控流程加一个"心跳"——每天检查监控流程最后一次成功运行时间，超过25小时没运行则短信告警。

坑3：告警太多会被无视
如果每天发30条告警，运维人员会开始忽视。原则：只在状态从"正常"变"异常"时告警，稳定异常不重复告警，只在恢复正常时再通知一次。

坑4：文件时间戳不等于流程实际运行时间
判断流程是否今天运行，不要只看输出文件的修改时间——文件可能被意外复制或备份。应从流程运行日志里读取时间戳。

---

适用人群： RPA运维工程师、IT运营、自动化负责人

难度： ⭐⭐⭐（需要一定的运维思维）