影刀RPA新手教程：企业微信机器人完全指南——群消息推送、Markdown卡片与异常告警

作者：林焱 | 真实案例驱动，每篇覆盖12大核心模块，禁止空话。

案例背景：半夜告警救了我的命

去年一个大促夜，库存同步出了bug，有订单卖了负库存的商品。

如果我没在凌晨2点收到企业微信告警，第二天就是一场灾难。

这套告警系统，就是用影刀RPA + 企业微信机器人搭的。

今天从零拆给你看。

---

一、安装与准备工作

企业微信机器人的配置比飞书多一步：要先在企业微信管理后台创建应用。

登录企业微信管理后台 → 应用管理 → 自建 → 创建应用。

应用创建完后，会拿到两个关键东西：

• AgentId
• Secret

这两个加上企业的 CorpId，就是调用企业微信API的完整凭证。

我当时犯的第一个错：把Secret写到了配置文件里，提交代码时忘了删。

企业微信的Secret比飞书Webhook更敏感，拿到就能以企业身份发消息。

正确做法：存到影刀控制台的"应用参数"里， runtime动态读取。

安装影刀的步骤不再重复，重点说Python依赖：

企业微信API需要签名计算，用Python的 hashlib 和 time 库就够了，不需要额外安装第三方包。

---

二、元素定位：获取企业微信的推送凭证

企业微信不支持简单的Webhook，必须通过API发送消息。

流程是：先拿 access_token，再用 access_token 发消息。

获取 access_token 的HTTP请求：

GET https://qyapi.weixin.qq.com/cgi-bin/gettoken
参数：
  corpid = 企业ID
  corpsecret = 应用Secret

返回格式：

{
  "errcode": 0,
  "errmsg": "ok",
  "access_token": "xxxxxx",
  "expires_in": 7200
}

access_token 有效期7200秒，也就是2小时。

关键优化：不要把每次发消息都去拿一次token，太慢而且有频率限制。

正确做法：把token缓存到本地文件或影刀变量里，快过期时再刷新。

我当时没做缓存，每次发消息都先请求token，大促时并发一高，token接口直接限频，消息全部发送失败。

---

三、变量与数据类型：消息格式的全家桶

企业微信支持的消息格式比飞书多，最常用的是文本、Markdown、卡片。

文本消息格式：

{
  "touser": "@all",
  "msgtype": "text",
  "agentid": 1000002,
  "text": {
    "content": "告警：库存同步异常"
  }
}

Markdown消息格式：

{
  "touser": "@all",
  "msgtype": "markdown",
  "agentid": 1000002,
  "markdown": {
    "content": "## 库存告警\n> 同步失败，请立即处理\n**时间**：2024-11-11 02:00"
  }
}

在影刀里，用"配置参数"指令组装这些JSON。

企业微信的消息体比飞书的复杂，因为要指定接收人（touser）。

发给所有人用 @all，发给特定人用企业微信的 userid，多个用竖线分隔：

touser = "zhangsan|lisi|wangwu"

我当时踩过这个坑：userid 不是企业微信里的昵称，是管理后台里的"账号"字段。

一堆消息发不出去，报错"userid not found"，查了半天才发现用的是昵称不是账号。

---

四、流程控制：分级告警策略

真实生产环境的告警，不能一概而论。

我设计了三级告警：

• 一级：提示级，发群消息，不@人
• 二级：警告级，发群消息 + @相关负责人
• 三级：严重级，发群消息 + @所有人 + 发邮件 + 发短信

用影刀的"如果…否则"指令实现：

如果 错误级别 == "严重"
  touser = "@all"
  mention_users = "所有人"
  消息前缀 = "【紧急】"
否则如果 错误级别 == "警告"
  touser = "组长userid"
  mention_users = "组长"
  消息前缀 = "【警告】"
否则
  touser = ""
  mention_users = ""
  消息前缀 = "【提示】"
结束如果

循环用在批量通知场景。

比如一个告警要同时通知3个群，把群的 chatid 存到列表里，遍历发送。

企业微信的群消息是通过"群机器人"Webhook发送的，和API发送是两条路。

群机器人Webhook格式和飞书类似，但域名不同：

https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx

---

五、网页自动化：从监控页面抓取异常信息

告警的触发，来源于对业务页面的定时检查。

用影刀的"启动新浏览器"指令，打开内部监控后台。

XPath + CSS 双保险定位：

先试XPath：

//table[@id='error-log']/tbody/tr[1]/td[2]

XPath失效时用CSS选择器兜底：

table#error-log tbody tr:first-child td:nth-child(2)

页面用了iframe的情况，要先切换iframe再定位：

在影刀里用"切换到iframe"指令，参数是iframe的id或XPath。

我当时踩过这个坑：监控后台用了嵌套iframe，我只切换了一层，元素一直定位不到。

解决方法是用"获取页面结构"指令，先看清楚iframe的层级关系，再决定切几层。

---

六、数据处理：把异常信息格式化成告警内容

从页面抓到的原始数据，要处理成适合告警的格式。

企业微信Markdown支持的颜色有限，用粗体和引用符号来做视觉区分：

content = "### 库存同步异常告警\n" +
          "> 错误类型：" + error_type + "\n" +
          "> 影响订单：" + order_count + "笔\n" +
          "**请立即登录后台处理**"

数据里有可能包含特殊字符，发JSON前要转义：

在Python里处理：

import json

def escape_markdown(text):
    # 企业微信Markdown需要转义的特殊字符
    escape_chars = ['\\', '`', '*', '_', '#']
    for char in escape_chars:
        text = text.replace(char, '\\' + char)
    return text

我当时遇到过一个bug：错误描述里有一段SQL语句，包含反引号和星号，发出去的消息格式全乱了。

后来所有动态内容在拼到消息体之前，都先过一遍转义处理。

---

七、鼠标键盘图像：截图附到告警里

纯文字告警有时候不够直观，尤其是界面类的异常。

用影刀的"截取整个页面截图"指令，把异常页面截下来。

企业微信发送图片，要先上传到临时素材，拿到 media_id，再用 media_id 发消息。

上传临时素材的HTTP请求：

POST https://qyapi.weixin.qq.com/cgi-bin/media/upload
参数：
  access_token = xxxxxx
  type = image
Body：
  文件二进制流

用Python处理文件上传：

import requests

def upload_image(token, image_path):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={token}&type=image"
    with open(image_path, 'rb') as f:
        files = {'media': f}
        resp = requests.post(url, files=files, timeout=30)
    result = resp.json()
    return result.get('media_id')

拿到 media_id 后，发图片消息：

{
  "touser": "@all",
  "msgtype": "image",
  "agentid": 1000002,
  "image": {
    "media_id": "xxxxxx"
  }
}

我当时踩过这个坑：临时素材的 media_id 有效期只有3天，但我存的缓存没设过期时间，3天后图片消息全部发不出去。

---

八、进阶技能：卡片消息与企业微信动作按钮

企业微信的卡片消息比飞书的更强大，支持按钮触发自定义动作。

卡片消息格式（文本卡片）：

{
  "touser": "@all",
  "msgtype": "textcard",
  "agentid": 1000002,
  "textcard": {
    "title": "库存同步异常",
    "description": "错误类型：API超时\n影响订单：23笔",
    "url": "https://ops.example.com/issue/12345",
    "btntxt": "查看详情"
  }
}

点击按钮后跳转到 url 指定的页面。

如果要实现点击按钮后触发一个API（比如"确认已处理"），要用企业微信的"模板卡片"消息。

模板卡片的配置比文本卡片复杂得多，需要用"模板卡片消息"API，而且只有认证企业才能用。

我当时评估过，对于告警场景，文本卡片 + 跳转链接已经够用了，没必要上模板卡片。

---

九、平台实战：在影刀配置异常触发任务

告警任务不能只靠定时触发，要能做到实时触发。

方案：业务系统发生异常时，调用影刀的开放API，触发影刀流程执行。

影刀开放API的调用方式（参考官方文档）：

POST https://api.winrobot360.com/oapi/dispatch/v2/job/start
Header：
  Authorization: Bearer {accessToken}
Body：
  {
    "appUuid": "应用UUID",
    "robotClientUuid": "机器人UUID",
    "params": {
      "error_msg": "具体错误信息"
    }
  }

在影刀控制台 → 应用管理 → 开放API，开启API调用权限，拿到 accessToken。

业务系统的异常处理逻辑里，加上调影刀API的代码，就能做到异常发生即触发推送。

我当时踩过这个坑：开放API的 accessToken 有效期是24小时，我没做自动刷新，token过期后所有告警都触发失败了。

正确做法：用影刀的"获取访问令牌"指令，每次调用前先刷新token。

---

十、系统联动：多渠道告警互不依赖

企业微信消息有可能因为网络问题发送失败，要设计降级方案。

我的方案：企业微信为主，邮件为辅，短信为最后防线。

在影刀流程里，用"尝试捕获"包住企业微信发送逻辑：

尝试
  调用企业微信API发送消息
  如果 返回码 != 200
    抛出异常 "企业微信发送失败"
  结束如果
捕获异常
  发送邮件告警
  如果 错误级别 == "严重"
    调用短信API发送短信告警
  结束如果
结束尝试

短信告警用阿里云短信服务，后面第4篇文章会详细讲。

系统联动的核心是：任何一个通道失败，不能影响其他通道的执行。

用影刀的"继续即使出错"指令，或者把每个通道放在独立的"尝试捕获"块里。

---

十一、工程化规范：告警系统的可维护性

告警系统最怕的是"告警风暴"——一个bug触发几千条告警，把所有人的手机炸了。

防护策略：

频率限制：同一个告警类型，5分钟内最多发1次。

在影刀里用"配置文件"指令，记录上次发送时间，每次发送前检查间隔：

读取 上次发送时间
如果 当前时间 - 上次发送时间 < 300秒
  跳过发送
  记录日志 "频率限制，跳过发送"
  结束流程
  ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/ab1b8c263eed4f529897e44b70fe1c9c.png#pic_center)

结束如果
写入 上次发送时间 = 当前时间

告警升级：同一个问题持续告警超过3次，自动升级为"严重"级别，加发短信。

用影刀的"配置文件"记录同一个告警的连续触发次数，每次发送后加1，恢复后清零。

我当时没做频率限制，有一个定时任务配置错了cron表达式，每分钟触发一次告警，连续发了2000多条。

企业微信把我们的应用限流了，所有消息被拦截，真正的严重告警也收不到了。

---

十二、速查表与常见报错

报错1：40014 invalid access_token

原因：token过期或无效。

解决：重新获取token，检查token有效期，做好缓存和刷新逻辑。

报错2：81013 userid not found

原因：touser 字段里的userid不存在。

解决：去企业微信管理后台确认用户的"账号"字段，不是昵称。

报错3：45009 api freq out of limit

原因：接口调用超过频率限制。

解决：加上token缓存，控制调用频率，必要时申请提高限额。

报错4：93000 应用未被授权使用

原因：应用没有给对应部门或人员授权。

解决：去企业微信管理后台，给应用添加可见范围。

企业微信消息类型速查：

类型	msgtype值	适用场景
文本	text	简单通知
Markdown	markdown	格式化通知
图片	image	截图告警
文本卡片	textcard	带按钮的通知
文件	file	日志文件发送

---

完整案例代码参考

获取access_token（Python，在影刀"执行Python代码"指令里调用）：

import requests
import time
import json

TOKEN_CACHE = {}

def get_access_token(corpid, corpsecret):
    cache_key = f"{corpid}_{corpsecret}"
    now = time.time()
    
    # 检查缓存
    if cache_key in TOKEN_CACHE:
        token, expire_time = TOKEN_CACHE[cache_key]
        if now < expire_time - 300:  # 提前5分钟刷新
            return token
    
    # 重新获取
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    params = {"corpid": corpid, "corpsecret": corpsecret}
    resp = requests.get(url, params=params, timeout=10).json()
    
    if resp.get('errcode') == 0:
        token = resp['access_token']
        TOKEN_CACHE[cache_key] = (token, now + resp['expires_in'])
        return token
    else:
        raise Exception(f"获取token失败：{resp}")

# 发送文本消息
def send_text_message(token, agentid, touser, content):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    data = {
        "touser": touser,
        "msgtype": "text",
        "agentid": agentid,
        "text": {"content": content}
    }
    resp = requests.post(url, json=data, timeout=10).json()
    return resp

XPath元素定位参考（监控后台错误日志）：

# 最新一条错误日志
//div[@class='error-list']/div[1]//text()

# 错误类型标签
//span[contains(@class,'error-tag')]/text()

# 动态加载的日志（等待出现）
//div[@id='logs']//tr[contains(@class,'new')]

CSS选择器备选：

/* 错误列表第一条 */
div.error-list > div:first-child

/* 包含"异常"文本的单元格 */
td:contains("异常")

/* 最近5分钟的错误（有特定class） */
tr.error-row.recent

---

我当时踩过这个坑：三个最深刻的教训

教训一：access_token 没做线程安全。

早期版本用的是全局变量存token，多个流程同时跑的时候，token被覆盖，一半的请求用了过期token。

后来改成每个流程独立拿token，或者用文件锁保证同一时间只有一个流程在刷新token。

教训二：告警消息里带了敏感信息。

有一次告警消息里包含了用户的手机号，被群里的人截图发到了外部群。

从那以后，所有推送到群里的消息，都先过一遍敏感信息过滤，手机号、身份证号、地址全部打码。

教训三：企业微信应用的可视范围没配好。

应用创建完默认只有创建者自己能看到，发出的消息其他人都收不到。

记得去"应用管理 → 可见范围"里，把需要接收消息的部门或人员加进去。

---

延伸：home.linyan.cloud 上的更多资源

企业微信机器人的完整配置模板，包括获取token的Python封装、所有消息类型的发送函数，我都整理到了 home.linyan.cloud。

还有一套"告警降频算法"的实现代码，能有效防止告警风暴，需要的可以去下载。

---

#影刀RPA #RPA教程 #企业微信机器人 #消息推送 #异常告警 #自动化 #林焱

作者：林焱