简介:面向需要快速打通企业微信消息通道的 Python 开发者和运维人员,这份小型代码示例包以企业微信机器人为核心,用极轻量的方式解决文本、Markdown 与图文消息的自动推送问题,可覆盖内部通知、告警提醒、运营报表等常见场景。压缩包内仅有 1 个 Python 脚本,大小仅 2KB,却完整封装了三种发送逻辑:send_text_msg 处理普通文本,send_markdown_msg 按标题与正文构造 Markdown 负载,send_image_text_msg 完成图文卡片组装,调用时只需传入 webhook_url 和对应的消息参数即可。开发者可直接将脚本嵌入监控报警、定时任务或数据分析流程,也能基于它二次封装成更完善的消息服务;由于代码量小、结构清晰,非常适合初学者对照企业微信官方 API 文档熟悉请求格式,同时为有经验的工程师提供一份精炼的集成范例。目前该示例已有 7519 人学习,对于希望低成本验证机器人消息能力、快速落地自动通知的团队或个人,是一份实用性很强的参考资源。1. 这个资源能解决什么:一条 Python 脚本把企业微信机器人消息送到群里深夜收到线上告警时,最让人烦躁的不是问题本身,而是还要手动复制文案、打开企业微信、粘贴到群里。这个资源就是用 Python 实现企业微信机器人消息发送的完整工程包——它把从“添加自定义机器人”到“群里弹出消息”之间所有环节都封装好了,你只需要填上机器人 webhook 地址和加签密钥,就能用一条命令把文本或 Markdown 消息推送到目标群聊。它特别适合做告警通知、CI/CD 构建结果推送、定时报表下发的开发者,解压后填 key 就能跑,省掉重新踩一遍签名和限流的坑。2. 企业微信机器人消息发送:从 Webhook 到群消息落地的完整链路企业微信机器人看着像个黑匣子,其实它只是把 HTTP 协议上的能力包装成了一个可以配置的接口。把这条链路从发送端到接收端看透,后面读核心代码才会顺畅。2.1 机器人消息的触发原理与 Webhook 地址是怎么来的企业微信的机器人不是某种常驻进程,而是在群聊里由管理员手动添加的“自定义机器人”。添加完成后,企业微信会生成一条形如这样的 URL:https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx这里面 key 就是机器人的身份凭证。我们做的所有工作,本质上就是组装一条 HTTP POST 请求打到这个 URL 上。这条链路里没有长连接、没有消息队列,你的程序永远是客户端,企业微信服务端永远是接收端。所以它也决定了这个功能可以被任何语言调用,Python 只是其中一个可选实现。调用链路上有几个关键点值得先记住。第一,key 标识的是“哪个群里的哪个机器人”,而不是你个人。第二,如果创建机器人时开启了“加签”,那么请求 URL 必须附带 timestamp 和 sign 参数,sign 是由密钥与时间戳计算出的动态串。第三,POST 之后响应体是一个 JSON,里面有个 errcode 字段,0 表示成功,其他数字表示不同的失败原因。把这三点理解到位,后面二次开发基本不会走偏。典型响应体长这样:{ "errcode": 0, "errmsg": "ok" }注意 errmsg 只是给人类看的提示,真正的判断依据永远是 errcode。这个习惯请从一开始就养好,否则你会被“HTTP 200 就代表成功”这个直觉坑好一阵子。2.2 解压后的工程结构:各文件职责与配置约定解压这套资源后,你大概率能看到下面这样的目录布局(不同版本可能略有差异,以 README.md 为准):wecom_bot_demo/ ├── send_msg.py # 命令行入口,支持传内容参数 ├── wecom_bot.py # 机器人核心模块,封装签名、构造、发送 ├── config.ini # 配置文件,存放 webhook key 与 secret ├── requirements.txt # 第三方依赖,一般只有 requests └── README.md # 使用说明与示例send_msg.py 的职责是让用户最快跑通。它读取配置,创建机器人对象,把命令行传入的字符串作为消息内容发出。它的代码量很小,因为你真正要学的是 wecom_bot.py。wecom_bot.py 里包含签名函数、消息构造方法、底层 POST 方法、错误码判断与重试,是所有改动最频繁的地方。config.ini 的内容一般长这样:[bot] webhook_key = your_key_here secret = your_secret_here配置这块我想提醒一个最容易产生后悔药的地方:不要把这个文件提交到公开仓库。webhook key 一旦泄露,任何人都可以往你的群里发任意内容。我见过有人把 key 写在 send_msg.py 里发给同事,结果同事随手贴到技术论坛求助,过了几分钟他们的工作群就开始刷屏。让配置与代码分离不是讲究,是底线。2.3 三种消息类型与它们的 payload 结构企业微信机器人最常用的是文本、Markdown、图片三种消息类型,它们的发送参数对企业微信服务端来说都是裸的 JSON 结构。文本消息最小,只需要一个 content 字段:{ "msgtype": "text", "text": { "content": "这是一条来自机器人的消息" } }Markdown 消息用 markdown 字段承载,内容里的换行和颜色标签需要自己控制:{ "msgtype": "markdown", "markdown": { "content": "# 标题\n **加粗** 或者 <font color=\"info\">绿色</font>" } }图片消息则要先读取图片文件,计算 base64 与 md5 后提交:{ "msgtype": "image", "image": { "base64": "图片base64字符串", "md5": "图片内容md5" } }三种类型在字段设计上一个比一个重,所以核心模块里通常会把“构造各种 payload”做成分散的公开方法,而把“最终发送”收敛成一个私有方法 _send_payload。这样一来,以后你想发图文卡片,只需要新写一个构造方法,发送链路完全不用动。把三种消息放在一起对比,你会发现它们最主要的差别只在 content 字段的位置不同,HTTP 请求本身毫无差别。这也是为什么这个场景非常适合用 Python 做封装——语言本身的 dict 操作足够灵活,构造 payload 的过程几乎不需要额外依赖。2.4 频率限制与管理入口如果使用量上来,你会遇到频控问题。企业微信机器人的发送频率有配额限制,虽然官方文档给出的数字不一定很明确,但以我实际跑下来的经验,单个机器人每分钟大约能发送 20 条消息左右,超过就会收到一条错误码。这个限制对所有调用方共享,所以如果你有多个脚本同时往同一个群发消息,额度是共用的。这个细节会在后面的避坑章节展开。另外,机器人管理入口在群聊的“群机器人”设置里,这里可以关闭机器人、重新生成 webhook key,但也会导致旧 key 失效。设计系统时应该把机器人管理纳入日常权限管理,而不是等出了事故再去找。2.5 用 Python 而不是 curl:脚本化带来的复用价值也许你会问,既然本质是 HTTP POST,那直接用 curl 不就行了?确实,一条 curl 命令能完成发送,但项目里需要的是“可复用的能力”:同一个机器人要发文本、发 Markdown,还要在出错时重试;告警脚本要把日志内容动态填进模板;运维平台要接多个机器人。这些场景如果用 curl,就得每次手写一大串参数,维护成本直线上升。用 Python 封装后,调用方只需关心业务内容,网络细节全部收敛在类里,这就是这份资源比一条 curl 命令更值得下载的原因。3. 核心模块拆解:一个请求从时间戳到群聊消息的完整执行第2章画了一张链路图,现在把链路拆开逐行看代码。如果目标只是发一条消息,可以跳到第4章;如果要做二次开发,这部分才是真正值钱的地方。3.1 签名生成:动态密钥不是防伪造那么简单企业微信机器人开启“加签”后,服务端要求每个 POST 地址带上 timestamp 和 sign 参数。sign 的计算方法是把当前时间戳、换行符、密钥拼接成字符串,做 SHA256 的十六进制摘要。代码实现是下面这个样子:import hashlib import time def generate_sign(secret: str, timestamp: int) -> str: string_to_sign = f"{timestamp}\n{secret}" hash_obj = hashlib.sha256(string_to_sign.encode("utf-8")) return hash_obj.hexdigest()调用方式是在每次发送前取出当前时间戳,拼进 URL 里:timestamp = int(time.time()) sign = generate_sign(secret, timestamp) url = f"{base_url}?key={webhook_key}×tamp={timestamp}&sign={sign}"这里有两个最容易写错的地方。第一,timestamp 必须是秒级整数,如果写成 float 或者毫秒级,服务端验签时直接拒绝。第二,拼接顺序是 timestamp + "\n" + secret,换行符不能丢,顺序也不能调换。市面上相当一部分教学代码在这里就写反了,表面上看密钥没错,实际上一条都发不出去。这份资源如果实现正确,你在本地测试时能明显感觉到“一次通过”。那么为什么需要签名呢?在企业微信服务端看来,密钥是一把只有你和服务端共享的钥匙,而时间戳用于防止重放攻击。把时间戳和密钥一起做摘要,服务端可以用同样的算法验证请求是否来自密钥持有者。这也能解释为什么密钥永远不应该出现在 URL 明文里,而只能出现在 Hash 的计算过程中。3.2 消息发送入口与底层 POST 封装文本与 Markdown 的构造逻辑非常相似,唯一区别是字段名。常见的工程做法是用一个类把所有逻辑聚在一起,对外暴露语义清晰的方法,而不是让调用方自己拼 dict。下面是一段符合这类资源风格的代码:import requests class WecomBot: def __init__(self, webhook_key: str, secret: str = ""): self.base_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send" self.webhook_key = webhook_key self.secret = secret def send_text(self, content: str) -> dict: payload = { "msgtype": "text", "text": {"content": content} } return self._post_payload(payload) def send_markdown(self, content: str) -> dict: payload = { "msgtype": "markdown", "markdown": {"content": content} } return self._post_payload(payload) def _post_payload(self, payload: dict) -> dict: url = self._build_url() try: resp = requests.post(url, json=payload, timeout=5) data = resp.json() except Exception as exc: return {"errcode": -1, "errmsg": str(exc)} return data def _build_url(self) -> str: timestamp = int(time.time()) url = f"{self.base_url}?key={self.webhook_key}" if self.secret: sign = generate_sign(self.secret, timestamp) url += f"×tamp={timestamp}&sign={sign}" return url这个类设计有几个关键点。第一,构造函数的两个参数把调用与配置解耦,调用方不需要知道签名怎么算、URL 怎么拼,只需要传入业务内容。第二,send_text 和 send_markdown 只负责把业务数据翻译成企业微信定义的 payload,发送逻辑收进 _post_payload,这样每增加一种消息类型,你不用复制一遍网络代码。第三,异常被捕获后返回一个统一结构的 dict,上层就能从返回值里判断成功与否,而不会因为一个网络超时把整个调度流程打断。至于为什么需要从 JSON 响应里再取一次 errcode,是因为 HTTP 本身的 200 只代表“请求到达服务端”,不代表业务成功。举个例子,签名错误时服务端照样返回 HTTP 200,但 body 里的 errcode 会是某个业务错误码。如果只检查 requests.post 的返回状态,你根本不知道消息有没有真正发出去。这是我每次写类似脚本时都要提醒自己的习惯。3.3 超时配置与失败重试:消息必达的最后一道保障发送消息这类通知任务,网络抖动是常态,所以给 requests 加上超时和重试才算完整。超时参数建议分开设置连接和读取时间:requests.post(url, json=payload, timeout=(3, 5))timeout=(3, 5) 的含义是连接阶段最多等 3 秒,读取响应最多等 5 秒,这样可以避免服务端一直不返回数据时脚本被挂死。普通的单值 timeout=5 对两个阶段强加同一个时间,不够精细。重试可以参考下面的模式:import time def _post_with_retry(self, url: str, payload: dict, max_retry: int = 3) -> dict: for attempt in range(max_retry): try: resp = requests.post(url, json=payload, timeout=(3, 5)) data = resp.json() if data.get("errcode") == 0: return data if data.get("errcode") == 45009: # 限流 time.sleep(2 ** attempt) except requests.exceptions.RequestException: time.sleep(2 ** attempt) return {"errcode": -1, "errmsg": "retry exhausted"}重试间隔 1、2、4 秒,指数退避防止在限流时立刻再次冲击服务端。如果连续三次都失败,返回一个自定义错误结构,交给上层决定是记录日志还是换渠道告警。到这里,一次请求的生命周期才算完整:签名、构造、发送、校验、重试。整个过程没有黑匣子,任何一步出错都可以通过返回值定位到具体原因。4. 本地跑通:解压后从零到发出一条消息现在到了动手环节。假设你已经把 .rar 解压,进入目录。下面按顺序操作,每一步我都说明它在做什么,这样出了问题你知道去哪排查。4.1 环境准备:确认 Python 版本与安装依赖先确认本地 Python 解释器版本:python --version建议至少使用 3.8。这个资源的代码如果用了 f-string、类型注解,那么 3.8 以下是跑不动的。如果你机器上同时有 Python 2 和 Python 3,请用 python3 命令,否则可能出现语法不兼容的报错。然后安装依赖:pip install -r requirements.txtrequirements.txt 里通常只会写 requests。如果里面带版本号,如 requests==2.31.0,建议直接安装锁定版本,没必要为了追新把行为改了。如果你从不使用虚拟环境,那么全局安装也没关系,这个小项目不涉及复杂的依赖冲突;但如果想保持环境干净,可以这样:python -m venv .venv source .venv/bin/activate # Windows 用户用 .venv\Scripts\activate pip install -r requirements.txt虚拟环境的主要价值是隔离。等你后面同时维护好几个 Python 项目,就知道这么做能省下多少互相踩依赖的问题。这里顺手一提,不是强制要求。4.2 配置文件:webhook key 与加签密钥怎么填拿到企微机器人的 webhook 地址后,把 key 填到 config.ini 中。如果机器人开启了加签,把加签密钥写入 secret 字段。这里推荐一种更灵活的读取方式,让程序既能读配置文件也能读环境变量:import configparser import os config = configparser.ConfigParser() config.read("config.ini", encoding="utf-8") webhook_key = os.getenv("WECOM_KEY") or config.get("bot", "webhook_key") secret = os.getenv("WECOM_SECRET") or config.get("bot", "secret", fallback="")这样本地开发用 config.ini,部署到服务器时改用环境变量,代码不用动。很多临时脚本为了图省事会把密钥硬编码在文件里,但只要你换一次环境,就知道这种省事其实设了一个以后要还的债。验证配置是否读对,可以直接跑一行 Python:python -c "import configparser; c=configparser.ConfigParser(); c.read('config.ini'); print(c.get('bot','webhook_key'))"能打印出那一串 key 就说明文件格式没问题。如果报 NoSectionError,说明 config.ini 里缺少 [bot] 节点,手动补上即可。4.3 执行发送脚本与验证结果在目录里执行:python send_msg.py "这是一条测试消息"如果一切正常,终端会输出类似这样的 JSON:{"errcode": 0, "errmsg": "ok"}然后去群里看,你应该能看到“这是一条测试消息”。如果返回的不是 0,而是某个错误码,优先去查该错误码的含义;一般 45009 是限流,40048 是签名错误,92000 是 key 无效。这些错误码在 README 里一般都有表格。如果返回 0 但群里没有消息,问题大概率不在代码,而在机器人本身:可能 key 是旧的,可能机器人被移出群聊,也可能发到了另一个群。这时候不要盲目改代码,先回到群设置里核对一下机器人信息,往往比在代码里翻半天更有效。还有个常见误区是以为 errcode 非 0 一定是代码问题,实际上很多业务错误码直接指向配置或权限,排查的第一步永远是看错误码本身。测试通过后,建议先停一停,不要立刻开始改逻辑,而是把整个流程再走一遍,确认每一步你都理解为什么。到这里,你已经完成了“解压即跑”,下面聊聊那些只有被坑过才知道的经验。5. 避坑指南:签名错误、限流与消息丢失的排查记录下面是几条来自真实事故现场的排查记录,每一条都会按照现象、原因、解决的顺序讲,方便你遇到同类问题时直接对照。5.1 签名计算总是报错:时间戳单位与换行符现象:配置完全正确、密钥复制无误,但发送请求后返回签名错误。原因:最常见的是本地时间与服务端时间偏差过大,或者把时间戳写成了毫秒值。企业微信服务端要求的是秒级时间戳 int(time.time()),不是 time.time() * 1000。另一个高频原因就是把拼接顺序搞反,正确算法是 timestamp + "\n" + secret,换了顺序结果完全不同。解决:先用一个临时脚本打印生成的 sign,再用企业微信提供的验证工具或官方文档里的示例算法对比,确认算法一致。接着检查本地系统时间是否与当前真实时间同步,偏差超过几分钟基本就会被拒。如果是内网机器的本地时间漂移,开启 NTP 同步或者手动校准一次,脚本立刻恢复正常。不要试图在代码里加一个“补偿秒数”,那是自欺欺人,也容易造成新的时间偏差。5.2 消息发多了被限流:错误码 45009 与共享配额现象:脚本第一次运行正常,第二次开始偶尔返回 errcode 45009,提示请求过于频繁。原因:单个企业微信机器人有发送频率限制,实际跑下来大概在每分钟 20 条左右,超过就限流。更隐蔽的是,同一个机器人的 webhook 被多个脚本共用时,所有脚本共享这个配额,而不是各自独立。解决:在设计阶段就把发送频率控制放在公共层,而不是在每个调用方后面乱加 sleep。如果只有低频通知,在 send_msg.py 入口加一个 time.sleep(0.5) 就够了;如果可能高频触发,建议在机器人类内部实现简单的令牌桶,限流在每分钟 10 条以内,留出余量。注意限流错误码需要和其他错误码一样被打印出来,方便判断到底是被限流还是其他故障。别把限流错误误判成网络问题,否则你会浪费一个下午在跟 DNS 和防火墙较劲。5.3 返回成功但群里没消息:旧 key 失效与机器人被移除现象:脚本运行输出 errcode 0、errmsg ok,但群里就是看不到新消息。原因:这是最迷惑人的一种情况,通常是因为群管理员重新生成过 webhook,或删除并重新添加了机器人,导致旧 key 已经失效,但服务端出于某些兼容原因仍返回成功。另外,如果机器人被移出群,同样可能发生副作用不明确的情况。解决:到企业微信的群机器人管理界面,对比当前 webhook 与 config.ini 里的 key 是否一致,不一致就更新。更稳妥的做法是写一个简单的“心跳测试”脚本,每天定时往群里发一条空消息或测试消息,一旦发现连续性失败就自动告警。不要指望服务端主动告诉你 key 失效,这个只能靠自己主动监测。5.4 Markdown 内容被转义后渲染异常现象:用 Markdown 消息发送告警详情,结果加粗符号失效、尖括号消失,HTML 标签直接显示成纯文本。原因:企业微信机器人对 Markdown 的支持并非完整 CommonMark,它对特殊字符有自己的转义机制,特别是 <、>、& 等符号,跟常规文档渲染不完全一致。解决:把动态内容里的特殊字符做转义后再拼进模板:def escape_markdown(text: str) -> str: return text.replace("&", "&").replace("<", "<").replace(">", ">")但要注意,只能对用户输入或第三方数据做转义,不能对整个模板做转义。很多人在封装函数的时候把所有 content 都传进去过一遍,结果连自己写的 # 标题、加粗符号也被转义了。核心思路是“动态数据转义,模板语法原样保留”,这条经验适用于所有消息类项目。另外顺带提两件容易被忽略的事:一是不要在公开仓库提交 config.ini,key 一旦泄露等于把你的群聊向全世界开放;二是所有网络请求都要设置超时,否则某个时刻企业微信服务端响应变慢,你的脚本就会一直挂着,表面上看是“不工作”,实际上是在等一个永远不来的响应。上面这些问题,资源里如果已经处理了就按它的方式用,没有的话按这几条自己补上即可。6. 二次开发:从单条发送到联动告警平台这套资源已经能解决“发送一条消息”的问题,但真正接入生产,你多半会碰到“发什么内容、什么时候发、失败怎么办”这几个扩展点。先说最常用的扩展:增加图片消息类型。企业微信机器人要求图片消息用 base64 字符串加 md5 值来内容寻址,你只需要在 WecomBot 类里加一个 send_image 方法,内部读取文件、计算 md5,再组装 payload 交给已有的 _post_payload,底层逻辑一行都不用改。再往前一步是“定时推送”。把发送方法接到 APScheduler 上,早上九点自动发报表模板:from apscheduler.schedulers.blocking import BlockingScheduler from wecom_bot import WecomBot bot = WecomBot("your_key", "your_secret") def job(): bot.send_text("每日指标更新:服务正常") scheduler = BlockingScheduler() scheduler.add_job(job, "cron", hour=9, minute=0) scheduler.start()如果你已经有监控报警系统,可以把机器人封装成一套通知渠道,在告警回调里调用 send_markdown,把异常堆栈、影响范围格式化后发到值班群。这一步的关键是让发送函数返回统一的结构,上层系统才能基于 errcode 判断是否需要升级告警,而不是依赖“是否抛异常”这种模糊边界。我自己的习惯是:每次接入新的 webhook 场景,都先花十分钟写一个最小复现脚本,只发送一条空文本消息验证密钥和签名,验证通过后再接业务模板。这样一来,出问题的时候永远不会把“密钥失效”和“业务代码写错”混在一起排查。从那以后我每次做消息推送类工具都会强制走一遍这个流程。希望帮到你。本文还有配套的精品资源,点击获取
简介:面向需要快速打通企业微信消息通道的 Python 开发者和运维人员,这份小型代码示例包以企业微信机器人为核心,用极轻量的方式解决文本、Markdown 与图文消息的自动推送问题,可覆盖内部通知、告警提醒、运营报表等常见场景。压缩包内仅有 1 个 Python 脚本,大小仅 2KB,却完整封装了三种发送逻辑:send_text_msg 处理普通文本,send_markdown_msg 按标题与正文构造 Markdown 负载,send_image_text_msg 完成图文卡片组装,调用时只需传入 webhook_url 和对应的消息参数即可。开发者可直接将脚本嵌入监控报警、定时任务或数据分析流程,也能基于它二次封装成更完善的消息服务;由于代码量小、结构清晰,非常适合初学者对照企业微信官方 API 文档熟悉请求格式,同时为有经验的工程师提供一份精炼的集成范例。目前该示例已有 7519 人学习,对于希望低成本验证机器人消息能力、快速落地自动通知的团队或个人,是一份实用性很强的参考资源。1. 这个资源能解决什么:一条 Python 脚本把企业微信机器人消息送到群里深夜收到线上告警时,最让人烦躁的不是问题本身,而是还要手动复制文案、打开企业微信、粘贴到群里。这个资源就是用 Python 实现企业微信机器人消息发送的完整工程包——它把从“添加自定义机器人”到“群里弹出消息”之间所有环节都封装好了,你只需要填上机器人 webhook 地址和加签密钥,就能用一条命令把文本或 Markdown 消息推送到目标群聊。它特别适合做告警通知、CI/CD 构建结果推送、定时报表下发的开发者,解压后填 key 就能跑,省掉重新踩一遍签名和限流的坑。2. 企业微信机器人消息发送:从 Webhook 到群消息落地的完整链路企业微信机器人看着像个黑匣子,其实它只是把 HTTP 协议上的能力包装成了一个可以配置的接口。把这条链路从发送端到接收端看透,后面读核心代码才会顺畅。2.1 机器人消息的触发原理与 Webhook 地址是怎么来的企业微信的机器人不是某种常驻进程,而是在群聊里由管理员手动添加的“自定义机器人”。添加完成后,企业微信会生成一条形如这样的 URL:https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx这里面 key 就是机器人的身份凭证。我们做的所有工作,本质上就是组装一条 HTTP POST 请求打到这个 URL 上。这条链路里没有长连接、没有消息队列,你的程序永远是客户端,企业微信服务端永远是接收端。所以它也决定了这个功能可以被任何语言调用,Python 只是其中一个可选实现。调用链路上有几个关键点值得先记住。第一,key 标识的是“哪个群里的哪个机器人”,而不是你个人。第二,如果创建机器人时开启了“加签”,那么请求 URL 必须附带 timestamp 和 sign 参数,sign 是由密钥与时间戳计算出的动态串。第三,POST 之后响应体是一个 JSON,里面有个 errcode 字段,0 表示成功,其他数字表示不同的失败原因。把这三点理解到位,后面二次开发基本不会走偏。典型响应体长这样:{ "errcode": 0, "errmsg": "ok" }注意 errmsg 只是给人类看的提示,真正的判断依据永远是 errcode。这个习惯请从一开始就养好,否则你会被“HTTP 200 就代表成功”这个直觉坑好一阵子。2.2 解压后的工程结构:各文件职责与配置约定解压这套资源后,你大概率能看到下面这样的目录布局(不同版本可能略有差异,以 README.md 为准):wecom_bot_demo/ ├── send_msg.py # 命令行入口,支持传内容参数 ├── wecom_bot.py # 机器人核心模块,封装签名、构造、发送 ├── config.ini # 配置文件,存放 webhook key 与 secret ├── requirements.txt # 第三方依赖,一般只有 requests └── README.md # 使用说明与示例send_msg.py 的职责是让用户最快跑通。它读取配置,创建机器人对象,把命令行传入的字符串作为消息内容发出。它的代码量很小,因为你真正要学的是 wecom_bot.py。wecom_bot.py 里包含签名函数、消息构造方法、底层 POST 方法、错误码判断与重试,是所有改动最频繁的地方。config.ini 的内容一般长这样:[bot] webhook_key = your_key_here secret = your_secret_here配置这块我想提醒一个最容易产生后悔药的地方:不要把这个文件提交到公开仓库。webhook key 一旦泄露,任何人都可以往你的群里发任意内容。我见过有人把 key 写在 send_msg.py 里发给同事,结果同事随手贴到技术论坛求助,过了几分钟他们的工作群就开始刷屏。让配置与代码分离不是讲究,是底线。2.3 三种消息类型与它们的 payload 结构企业微信机器人最常用的是文本、Markdown、图片三种消息类型,它们的发送参数对企业微信服务端来说都是裸的 JSON 结构。文本消息最小,只需要一个 content 字段:{ "msgtype": "text", "text": { "content": "这是一条来自机器人的消息" } }Markdown 消息用 markdown 字段承载,内容里的换行和颜色标签需要自己控制:{ "msgtype": "markdown", "markdown": { "content": "# 标题\n **加粗** 或者 <font color=\"info\">绿色</font>" } }图片消息则要先读取图片文件,计算 base64 与 md5 后提交:{ "msgtype": "image", "image": { "base64": "图片base64字符串", "md5": "图片内容md5" } }三种类型在字段设计上一个比一个重,所以核心模块里通常会把“构造各种 payload”做成分散的公开方法,而把“最终发送”收敛成一个私有方法 _send_payload。这样一来,以后你想发图文卡片,只需要新写一个构造方法,发送链路完全不用动。把三种消息放在一起对比,你会发现它们最主要的差别只在 content 字段的位置不同,HTTP 请求本身毫无差别。这也是为什么这个场景非常适合用 Python 做封装——语言本身的 dict 操作足够灵活,构造 payload 的过程几乎不需要额外依赖。2.4 频率限制与管理入口如果使用量上来,你会遇到频控问题。企业微信机器人的发送频率有配额限制,虽然官方文档给出的数字不一定很明确,但以我实际跑下来的经验,单个机器人每分钟大约能发送 20 条消息左右,超过就会收到一条错误码。这个限制对所有调用方共享,所以如果你有多个脚本同时往同一个群发消息,额度是共用的。这个细节会在后面的避坑章节展开。另外,机器人管理入口在群聊的“群机器人”设置里,这里可以关闭机器人、重新生成 webhook key,但也会导致旧 key 失效。设计系统时应该把机器人管理纳入日常权限管理,而不是等出了事故再去找。2.5 用 Python 而不是 curl:脚本化带来的复用价值也许你会问,既然本质是 HTTP POST,那直接用 curl 不就行了?确实,一条 curl 命令能完成发送,但项目里需要的是“可复用的能力”:同一个机器人要发文本、发 Markdown,还要在出错时重试;告警脚本要把日志内容动态填进模板;运维平台要接多个机器人。这些场景如果用 curl,就得每次手写一大串参数,维护成本直线上升。用 Python 封装后,调用方只需关心业务内容,网络细节全部收敛在类里,这就是这份资源比一条 curl 命令更值得下载的原因。3. 核心模块拆解:一个请求从时间戳到群聊消息的完整执行第2章画了一张链路图,现在把链路拆开逐行看代码。如果目标只是发一条消息,可以跳到第4章;如果要做二次开发,这部分才是真正值钱的地方。3.1 签名生成:动态密钥不是防伪造那么简单企业微信机器人开启“加签”后,服务端要求每个 POST 地址带上 timestamp 和 sign 参数。sign 的计算方法是把当前时间戳、换行符、密钥拼接成字符串,做 SHA256 的十六进制摘要。代码实现是下面这个样子:import hashlib import time def generate_sign(secret: str, timestamp: int) -> str: string_to_sign = f"{timestamp}\n{secret}" hash_obj = hashlib.sha256(string_to_sign.encode("utf-8")) return hash_obj.hexdigest()调用方式是在每次发送前取出当前时间戳,拼进 URL 里:timestamp = int(time.time()) sign = generate_sign(secret, timestamp) url = f"{base_url}?key={webhook_key}×tamp={timestamp}&sign={sign}"这里有两个最容易写错的地方。第一,timestamp 必须是秒级整数,如果写成 float 或者毫秒级,服务端验签时直接拒绝。第二,拼接顺序是 timestamp + "\n" + secret,换行符不能丢,顺序也不能调换。市面上相当一部分教学代码在这里就写反了,表面上看密钥没错,实际上一条都发不出去。这份资源如果实现正确,你在本地测试时能明显感觉到“一次通过”。那么为什么需要签名呢?在企业微信服务端看来,密钥是一把只有你和服务端共享的钥匙,而时间戳用于防止重放攻击。把时间戳和密钥一起做摘要,服务端可以用同样的算法验证请求是否来自密钥持有者。这也能解释为什么密钥永远不应该出现在 URL 明文里,而只能出现在 Hash 的计算过程中。3.2 消息发送入口与底层 POST 封装文本与 Markdown 的构造逻辑非常相似,唯一区别是字段名。常见的工程做法是用一个类把所有逻辑聚在一起,对外暴露语义清晰的方法,而不是让调用方自己拼 dict。下面是一段符合这类资源风格的代码:import requests class WecomBot: def __init__(self, webhook_key: str, secret: str = ""): self.base_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send" self.webhook_key = webhook_key self.secret = secret def send_text(self, content: str) -> dict: payload = { "msgtype": "text", "text": {"content": content} } return self._post_payload(payload) def send_markdown(self, content: str) -> dict: payload = { "msgtype": "markdown", "markdown": {"content": content} } return self._post_payload(payload) def _post_payload(self, payload: dict) -> dict: url = self._build_url() try: resp = requests.post(url, json=payload, timeout=5) data = resp.json() except Exception as exc: return {"errcode": -1, "errmsg": str(exc)} return data def _build_url(self) -> str: timestamp = int(time.time()) url = f"{self.base_url}?key={self.webhook_key}" if self.secret: sign = generate_sign(self.secret, timestamp) url += f"×tamp={timestamp}&sign={sign}" return url这个类设计有几个关键点。第一,构造函数的两个参数把调用与配置解耦,调用方不需要知道签名怎么算、URL 怎么拼,只需要传入业务内容。第二,send_text 和 send_markdown 只负责把业务数据翻译成企业微信定义的 payload,发送逻辑收进 _post_payload,这样每增加一种消息类型,你不用复制一遍网络代码。第三,异常被捕获后返回一个统一结构的 dict,上层就能从返回值里判断成功与否,而不会因为一个网络超时把整个调度流程打断。至于为什么需要从 JSON 响应里再取一次 errcode,是因为 HTTP 本身的 200 只代表“请求到达服务端”,不代表业务成功。举个例子,签名错误时服务端照样返回 HTTP 200,但 body 里的 errcode 会是某个业务错误码。如果只检查 requests.post 的返回状态,你根本不知道消息有没有真正发出去。这是我每次写类似脚本时都要提醒自己的习惯。3.3 超时配置与失败重试:消息必达的最后一道保障发送消息这类通知任务,网络抖动是常态,所以给 requests 加上超时和重试才算完整。超时参数建议分开设置连接和读取时间:requests.post(url, json=payload, timeout=(3, 5))timeout=(3, 5) 的含义是连接阶段最多等 3 秒,读取响应最多等 5 秒,这样可以避免服务端一直不返回数据时脚本被挂死。普通的单值 timeout=5 对两个阶段强加同一个时间,不够精细。重试可以参考下面的模式:import time def _post_with_retry(self, url: str, payload: dict, max_retry: int = 3) -> dict: for attempt in range(max_retry): try: resp = requests.post(url, json=payload, timeout=(3, 5)) data = resp.json() if data.get("errcode") == 0: return data if data.get("errcode") == 45009: # 限流 time.sleep(2 ** attempt) except requests.exceptions.RequestException: time.sleep(2 ** attempt) return {"errcode": -1, "errmsg": "retry exhausted"}重试间隔 1、2、4 秒,指数退避防止在限流时立刻再次冲击服务端。如果连续三次都失败,返回一个自定义错误结构,交给上层决定是记录日志还是换渠道告警。到这里,一次请求的生命周期才算完整:签名、构造、发送、校验、重试。整个过程没有黑匣子,任何一步出错都可以通过返回值定位到具体原因。4. 本地跑通:解压后从零到发出一条消息现在到了动手环节。假设你已经把 .rar 解压,进入目录。下面按顺序操作,每一步我都说明它在做什么,这样出了问题你知道去哪排查。4.1 环境准备:确认 Python 版本与安装依赖先确认本地 Python 解释器版本:python --version建议至少使用 3.8。这个资源的代码如果用了 f-string、类型注解,那么 3.8 以下是跑不动的。如果你机器上同时有 Python 2 和 Python 3,请用 python3 命令,否则可能出现语法不兼容的报错。然后安装依赖:pip install -r requirements.txtrequirements.txt 里通常只会写 requests。如果里面带版本号,如 requests==2.31.0,建议直接安装锁定版本,没必要为了追新把行为改了。如果你从不使用虚拟环境,那么全局安装也没关系,这个小项目不涉及复杂的依赖冲突;但如果想保持环境干净,可以这样:python -m venv .venv source .venv/bin/activate # Windows 用户用 .venv\Scripts\activate pip install -r requirements.txt虚拟环境的主要价值是隔离。等你后面同时维护好几个 Python 项目,就知道这么做能省下多少互相踩依赖的问题。这里顺手一提,不是强制要求。4.2 配置文件:webhook key 与加签密钥怎么填拿到企微机器人的 webhook 地址后,把 key 填到 config.ini 中。如果机器人开启了加签,把加签密钥写入 secret 字段。这里推荐一种更灵活的读取方式,让程序既能读配置文件也能读环境变量:import configparser import os config = configparser.ConfigParser() config.read("config.ini", encoding="utf-8") webhook_key = os.getenv("WECOM_KEY") or config.get("bot", "webhook_key") secret = os.getenv("WECOM_SECRET") or config.get("bot", "secret", fallback="")这样本地开发用 config.ini,部署到服务器时改用环境变量,代码不用动。很多临时脚本为了图省事会把密钥硬编码在文件里,但只要你换一次环境,就知道这种省事其实设了一个以后要还的债。验证配置是否读对,可以直接跑一行 Python:python -c "import configparser; c=configparser.ConfigParser(); c.read('config.ini'); print(c.get('bot','webhook_key'))"能打印出那一串 key 就说明文件格式没问题。如果报 NoSectionError,说明 config.ini 里缺少 [bot] 节点,手动补上即可。4.3 执行发送脚本与验证结果在目录里执行:python send_msg.py "这是一条测试消息"如果一切正常,终端会输出类似这样的 JSON:{"errcode": 0, "errmsg": "ok"}然后去群里看,你应该能看到“这是一条测试消息”。如果返回的不是 0,而是某个错误码,优先去查该错误码的含义;一般 45009 是限流,40048 是签名错误,92000 是 key 无效。这些错误码在 README 里一般都有表格。如果返回 0 但群里没有消息,问题大概率不在代码,而在机器人本身:可能 key 是旧的,可能机器人被移出群聊,也可能发到了另一个群。这时候不要盲目改代码,先回到群设置里核对一下机器人信息,往往比在代码里翻半天更有效。还有个常见误区是以为 errcode 非 0 一定是代码问题,实际上很多业务错误码直接指向配置或权限,排查的第一步永远是看错误码本身。测试通过后,建议先停一停,不要立刻开始改逻辑,而是把整个流程再走一遍,确认每一步你都理解为什么。到这里,你已经完成了“解压即跑”,下面聊聊那些只有被坑过才知道的经验。5. 避坑指南:签名错误、限流与消息丢失的排查记录下面是几条来自真实事故现场的排查记录,每一条都会按照现象、原因、解决的顺序讲,方便你遇到同类问题时直接对照。5.1 签名计算总是报错:时间戳单位与换行符现象:配置完全正确、密钥复制无误,但发送请求后返回签名错误。原因:最常见的是本地时间与服务端时间偏差过大,或者把时间戳写成了毫秒值。企业微信服务端要求的是秒级时间戳 int(time.time()),不是 time.time() * 1000。另一个高频原因就是把拼接顺序搞反,正确算法是 timestamp + "\n" + secret,换了顺序结果完全不同。解决:先用一个临时脚本打印生成的 sign,再用企业微信提供的验证工具或官方文档里的示例算法对比,确认算法一致。接着检查本地系统时间是否与当前真实时间同步,偏差超过几分钟基本就会被拒。如果是内网机器的本地时间漂移,开启 NTP 同步或者手动校准一次,脚本立刻恢复正常。不要试图在代码里加一个“补偿秒数”,那是自欺欺人,也容易造成新的时间偏差。5.2 消息发多了被限流:错误码 45009 与共享配额现象:脚本第一次运行正常,第二次开始偶尔返回 errcode 45009,提示请求过于频繁。原因:单个企业微信机器人有发送频率限制,实际跑下来大概在每分钟 20 条左右,超过就限流。更隐蔽的是,同一个机器人的 webhook 被多个脚本共用时,所有脚本共享这个配额,而不是各自独立。解决:在设计阶段就把发送频率控制放在公共层,而不是在每个调用方后面乱加 sleep。如果只有低频通知,在 send_msg.py 入口加一个 time.sleep(0.5) 就够了;如果可能高频触发,建议在机器人类内部实现简单的令牌桶,限流在每分钟 10 条以内,留出余量。注意限流错误码需要和其他错误码一样被打印出来,方便判断到底是被限流还是其他故障。别把限流错误误判成网络问题,否则你会浪费一个下午在跟 DNS 和防火墙较劲。5.3 返回成功但群里没消息:旧 key 失效与机器人被移除现象:脚本运行输出 errcode 0、errmsg ok,但群里就是看不到新消息。原因:这是最迷惑人的一种情况,通常是因为群管理员重新生成过 webhook,或删除并重新添加了机器人,导致旧 key 已经失效,但服务端出于某些兼容原因仍返回成功。另外,如果机器人被移出群,同样可能发生副作用不明确的情况。解决:到企业微信的群机器人管理界面,对比当前 webhook 与 config.ini 里的 key 是否一致,不一致就更新。更稳妥的做法是写一个简单的“心跳测试”脚本,每天定时往群里发一条空消息或测试消息,一旦发现连续性失败就自动告警。不要指望服务端主动告诉你 key 失效,这个只能靠自己主动监测。5.4 Markdown 内容被转义后渲染异常现象:用 Markdown 消息发送告警详情,结果加粗符号失效、尖括号消失,HTML 标签直接显示成纯文本。原因:企业微信机器人对 Markdown 的支持并非完整 CommonMark,它对特殊字符有自己的转义机制,特别是 <、>、& 等符号,跟常规文档渲染不完全一致。解决:把动态内容里的特殊字符做转义后再拼进模板:def escape_markdown(text: str) -> str: return text.replace("&", "&").replace("<", "<").replace(">", ">")但要注意,只能对用户输入或第三方数据做转义,不能对整个模板做转义。很多人在封装函数的时候把所有 content 都传进去过一遍,结果连自己写的 # 标题、加粗符号也被转义了。核心思路是“动态数据转义,模板语法原样保留”,这条经验适用于所有消息类项目。另外顺带提两件容易被忽略的事:一是不要在公开仓库提交 config.ini,key 一旦泄露等于把你的群聊向全世界开放;二是所有网络请求都要设置超时,否则某个时刻企业微信服务端响应变慢,你的脚本就会一直挂着,表面上看是“不工作”,实际上是在等一个永远不来的响应。上面这些问题,资源里如果已经处理了就按它的方式用,没有的话按这几条自己补上即可。6. 二次开发:从单条发送到联动告警平台这套资源已经能解决“发送一条消息”的问题,但真正接入生产,你多半会碰到“发什么内容、什么时候发、失败怎么办”这几个扩展点。先说最常用的扩展:增加图片消息类型。企业微信机器人要求图片消息用 base64 字符串加 md5 值来内容寻址,你只需要在 WecomBot 类里加一个 send_image 方法,内部读取文件、计算 md5,再组装 payload 交给已有的 _post_payload,底层逻辑一行都不用改。再往前一步是“定时推送”。把发送方法接到 APScheduler 上,早上九点自动发报表模板:from apscheduler.schedulers.blocking import BlockingScheduler from wecom_bot import WecomBot bot = WecomBot("your_key", "your_secret") def job(): bot.send_text("每日指标更新:服务正常") scheduler = BlockingScheduler() scheduler.add_job(job, "cron", hour=9, minute=0) scheduler.start()如果你已经有监控报警系统,可以把机器人封装成一套通知渠道,在告警回调里调用 send_markdown,把异常堆栈、影响范围格式化后发到值班群。这一步的关键是让发送函数返回统一的结构,上层系统才能基于 errcode 判断是否需要升级告警,而不是依赖“是否抛异常”这种模糊边界。我自己的习惯是:每次接入新的 webhook 场景,都先花十分钟写一个最小复现脚本,只发送一条空文本消息验证密钥和签名,验证通过后再接业务模板。这样一来,出问题的时候永远不会把“密钥失效”和“业务代码写错”混在一起排查。从那以后我每次做消息推送类工具都会强制走一遍这个流程。希望帮到你。本文还有配套的精品资源,点击获取