WeCom(企业微信)Bot 工具类逐段做详细说明

下面对这份 WeCom(企业微信)Bot 工具类逐段做详细说明,并给出常见坑位与可改进点。

总体概览

  • 这是一份封装了企业微信应用消息发送的 Python 工具类 WeChat。

  • 主要能力:

    • AccessToken 获取与自动续期

    • 媒体上传:临时图片 media_id、CDN 图片 URL

    • 发送消息:文本、图片、news(简单图文)、mpnews(富图文)

    • 模板卡片:text_notice、news_notice、button_interaction 的发送与更新

    • 可选 Redis 存储 task_id->command 的映射;可选将任务上下文落盘到本地临时文件

依赖与配置

  • 核心依赖:requests;可选 redis、grp、pwd(在 Linux 下用于 chown)。

  • 建议使用环境变量注入敏感信息:

    • WECHAT_CORP_ID

    • WECHAT_AGENT_SECRET

    • WECHAT_AGENT_ID

  • 代码里有默认 corpid/secret/agentid,仅用于演示,生产请务必删除硬编码。

初始化与 Redis

  • init

    • 从入参或环境变量读取 corpid、secret、agentid,创建 requests.Session。

    • token 初始为空;_token_expire_at 记录过期时间(秒级)。

    • 可选连接 Redis(默认 db=12)。用于存储 WeChatCommand_KEY 哈希表:task_id -> command_str。

    • 即使没有 Redis,类也能工作,相关功能自动跳过。

  • 注意:

    • 生产环境建议配置 requests 的代理、重试、连接池等(后文有建议)。

    • Redis 连接失败不抛异常,只打印提示。

Token 管理

  • refreshToken():

    • 调用企业微信 gettoken 接口,成功后缓存 access_token 和过期时间。

    • 过期时间提前 5 分钟刷新(expires_in - 300)。

  • ensureToken():

    • 无 token 或接近过期则刷新,失败返回 False。
  • 典型坑位:

    • Token 是企业级全局资源,多进程/多实例共享建议放到 Redis 缓存,并用分布式锁避免“惊群刷新”。

    • 强烈建议对 gettoken 做调用频率限制,避免被限流。

统一 POST 封装

  • _post_with_retry(url_tpl, json_body=None, files=None, timeout=15):

    • 确保 token 有效后请求 POST。

    • 若响应 JSON 的 errcode 为 token 无效/过期(40014/42001/40001),自动 refresh 后重试一次。

    • 返回 requests.Response(可能是 JSON,也可能不是,取决于接口)。

  • 注意:

    • 若第一次解析 JSON 失败,会直接返回 Response;上层调用需自己 r.json() 并处理异常。

    • 建议增强:统一解析与错误包装、网络级重试(指数退避)、日志打点。

媒体上传

  • uploadPic(file_name) -> Optional[str]:

    • 调用 media/upload?type=image(临时素材),返回 media_id。

    • 用于发送 image、或 mpnews 的 thumb_media_id。

    • 临时素材有效期(企业微信)一般是 3 天,过期后 media_id 不可用。

  • uploadImageGetUrl(file_name) -> Optional[str]:

    • 调用 media/uploadimg 上传图片到 WeCom CDN,返回 https 图片 URL。

    • 不占用素材配额,常用于 news.picurl。

  • 常见限制:

    • 图片大小与格式受限(常见如 JPG/PNG,大小 ≤ 2MB 等,具体以官方文档为准)。

    • uploadimg 返回的 URL 需为公网 https,客户端才能加载。

基础消息发送

  • sendMsg(content, to_user="@all") -> bool:

    • 发送文本消息,自动追加时间戳。

    • payload:

      • touser 支持具体 userids(以 | 分隔)、@all、或可扩展 toparty/totag(当前类未封装)。

      • agentid:应用 ID。

      • text.content:建议不超过文档限制(常见为 2048 字节左右)。

      • safe:0/1,1 表示保密消息(仅在客户端展示)。

  • sendPictureFile(file_name, to_user="@all")/sendPicture(media_id, to_user="@all"):

    • 先上传取 media_id 后发图片,或直接用已有 media_id。
  • sendNews(articles, to_user="@all"):

    • 简单图文 news:

      • 每个 article 包含 title/description/url/picurl/btntxt。

      • picurl 必须为公网 https(可用 uploadImageGetUrl)。

  • sendMpNews(articles, to_user="@all"):

    • 富图文 mpnews:

      • 需 thumb_media_id(由 uploadPic 返回),content 支持 HTML。

      • digest 为摘要,show_cover_pic = 1 显示封面。

  • 返回值:

    • 以上方法统一返回布尔值,且会打印 HTTP 状态码与响应内容。生产建议改为日志并返回结构化结果。

模板卡片发送与更新

  • sendTaskCard(...):

    • 发送 button_interaction 类型模板卡片。

    • 自动生成 task_id(或自传),支持:

      • main_title:title/desc

      • horizontal_content_list:展示的键值对

      • button_list:按钮配置(text/style/key)

      • 可选 button_selection:数量/选项选择器

    • 额外行为:

      • 若传入 command_str 且 Redis 可用,保存到 Hash “WeChatCommand_KEY”:task_id -> command。

      • 将任务上下文写入本地 task_tmp/ 目录的文件,文件名为 task_id(JSON 内容)。若系统支持尝试 chown 给 jupyter 用户组(非必要,失败忽略)。

  • sendTemplateCard(card_type, ...):

    • 通用卡片封装,支持:

      • text_notice:强调内容、引用区、跳转、水平内容列等

      • news_notice:卡片大图、纵向/横向内容、跳转

      • button_interaction:与 sendTaskCard 类似

  • updateTemplateCard(replace_card: Dict) -> bool:

    • 用于“按钮点击回调”后更新卡片内容(例如换成结果态)。

    • 关键字段:

      • agentid:应用 ID

      • response_code:来自企业微信回调事件(用户点击按钮后 WeCom 推送给你的回调数据里带的 response_code)

      • replace_card:用于替换的卡片结构(如 text_notice)

    • 注意:

      • 本代码未实现“接收回调”的服务端逻辑。要完成交互闭环,需部署回调服务(接收企业微信的事件推送,解析出 response_code、task_id、按钮 key、选项等),然后调用 updateTemplateCard。

      • response_code 与具体这条消息强绑定,只能更新对应的那一条。

main 示例流程解读

  • wc = WeChat():从环境变量或默认值读取配置,准备好会话。

  • 发送文本消息:wc.sendMsg("测试消息")

  • 上传 baoxiang.png 得到 media_id 并发送图片。

  • 发送 news:

    • 若 uploadImageGetUrl 成功,使用其返回 URL 作为 picurl;否则使用一个备用 https 图片地址。
  • 发送 mpnews:

    • 先上传 jinzhuan.png 得到 thumb_media_id,再发送支持 HTML 内容的富图文。
  • 发送按钮交互模板卡片:

    • 提供按钮与一个选择器(1张/3张),并将 command_str 写入 Redis。
  • 更新模板卡片:

    • 示例 replace 中的 response_code 是占位;真实值需从回调事件中获取。

    • replace_card 用 text_notice 显示“任务完成”。

常见坑位与注意事项

  • 凭证与安全

    • 不要把 corpid/secret/agentid 硬编码到代码;使用环境变量或安全配置中心。

    • 发送消息接口的返回 errcode 要记录/告警,方便定位权限、可见范围、目标成员不存在等问题。

  • 对象范围与发送目标

    • touser、toparty、totag 三者可组合,但至少有一个。当前类仅封装了 touser,若要发部门或标签消息需扩展。

    • @all 需要确保应用对全员可见且有权限。

  • 频率与限流

    • gettoken、message/send 均有频率限制,务必缓存 token 并进行失败重试与退避策略。
  • 媒体与内容限制

    • 临时素材(media_id)有效期短(3天),过期需重新上传。

    • 内容长度(text、标题、摘要、卡片字段)都有字节限制;过长会报错或被截断。

    • news.picurl 必须是公网 https,否则客户端不显示图片。

  • 模板卡片交互闭环

    • 仅发送按钮卡片不够,还需要:

      • 搭建回调接收服务(企业微信“事件回调”),校验消息签名与解密。

      • 解析点击事件,拿到 response_code、task_id、按钮 key、用户信息等。

      • 执行业务逻辑(例如根据 task_id 在 Redis 找到 command_str 并执行)。

      • 调用 updateTemplateCard 替换成“处理中/完成/失败”等结果态。

  • 可靠性

    • _post_with_retry 仅在 token 失效时重试一次,网络瞬断/超时不会重试。建议加网络级重试(幂等性注意)。

    • 当前使用 print,生产建议使用标准 logging 并带有 request_id、msgid、耗时与响应体摘要。

可改进建议

  • 配置与安全

    • 删除默认 corpid/secret,强制从环境/配置文件读取。

    • 支持从 Redis 共享 access_token(setex + 分布式锁),多实例统一缓存。

  • 稳定性与可观测性

    • 使用 requests.adapters.HTTPAdapter 配置重试(对可安全重试的接口);增加超时和连接池参数;支持代理。

    • 标准化错误返回:让 sendXxx 返回结构化对象(成功/失败、errcode、errmsg、request_id)。

    • 使用 logging 替代 print,并提供 debug/info/warn/error 级别。

  • API 覆盖

    • 增加 markdown 消息(msgtype=markdown)、文件消息、文本卡片等接口。

    • 支持 toparty、totag 字段。

    • message/send 的 enable_duplicate_check 与 duplicate_check_interval 支持,降低重复消息风险。

  • 线程/进程安全

    • ensureToken 增加线程锁;多进程场景用 Redis 分布式锁。
  • 结构化建模

    • 为 Articles、TemplateCard 等定义数据类(dataclass/TypedDict),增强类型检查与 IDE 体验。

    • 为 replace_card 定义校验函数,确保结构满足文档要求。

  • 本地文件与权限

    • task_tmp 的 chown 在非 root 下会失败(已捕获),建议改为可选行为或通过环境配置关闭。

    • 对临时文件增加清理策略(按时间或数量上限)。

小提示:updateTemplateCard 的典型请求体

  • 你在回调里拿到 response_code 后,构造如下 payload 调用更新接口:

    • {

      "agentid": 1000002,

      "response_code": "RSP_xxx_from_callback",

      "replace_card": {

      "card_type": "text_notice",
      "main_title": {"title": "任务完成", "desc": "备份成功"},
      "sub_title_text": "用时 12s",
      "horizontal_content_list": [
        {"keyname": "执行人", "value": "ZhangSan"},
        {"keyname": "任务ID", "value": "20250811_123456"}
      ]
      

      }

      }

  • replace_card 的结构需符合模板卡片规范(与 send 时的结构一致,只是通过 response_code 指定要替换哪一条)。

如何快速运行与排错

  • 运行前:

    • pip install requests redis

    • 导出环境变量 WECHAT_CORP_ID/WECHAT_AGENT_SECRET/WECHAT_AGENT_ID

    • 保证 baoxiang.png、jinzhuan.png 在当前目录或调整路径。

  • 首次跑建议仅调用 wc.refreshToken() 或 wc.sendMsg("hello") 验证鉴权。

  • 若报错:

    • 关注打印的 HTTP 状态码与 errcode/errmsg。

    • 检查应用是否对目标用户可见、用户是否在同一个企业、agentid 是否正确。

    • 图片是否满足要求、picurl 是否为公网 https、网络是否可达。

总结

  • 这份代码已具备“可直接用于小型场景”的完整度:会话与 token 管理、常见消息类型、模板卡片交互的发送与更新接口形态。

  • 生产落地建议重点完善:token/请求重试策略、错误与日志、回调服务闭环、限流与安全配置,以及对更多 API 字段的覆盖与校验。