下面对这份 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 字段的覆盖与校验。