简介:基于 Web API 的高度自定义微信机器人开源方案,适合需要将微信消息处理自动化的开发者、运维人员、社群运营者及 Python 爱好者。整体能力覆盖自动回复、消息转发、防撤回、留言统计等常用场景,接口设计较为灵活,便于结合自身业务需求做个性化定制与功能裁剪,适合从零搭建轻量级微信助手。代码体积仅 71KB,包体十分轻量、结构紧凑,无需复杂依赖即可快速阅读与部署试用。目前已有 1100 余人学习下载,热度持续,对初探微信接口开发与消息自动化的人群具有实际参考价值。通过这份源码,读者可以看清各功能模块的组织方式,理解自动回复规则配置、消息转发链路、防撤回探测以及留言统计的数据处理逻辑,同时积累消息监听、协议交互与异常处理方面的排错经验,可作为开发个人微信机器人与学习接口编程的实用起点。
1. 高度自定义微信机器人:先搞清这份资源能做什么,再决定要不要解压
我把这份微信机器人开源项目压缩包翻了一遍,先给结论:它不是一个开箱即用的成品,而是一个面向个人微信协议、高度自定义的半成品框架,核心能力覆盖自动回复、消息转发、防撤回拦截、留言统计四条链路,几乎全部靠配置文件驱动。适合谁?适合手上管着两三个微信号、想用半自动方式顶住重复咨询的运营,也适合要在群里做消息记录和活跃度统计的产品。不适合谁?想做全自动营销、批量拉群裂变的人,趁早放弃。个人微信机器人有三个绕不开的坎:登录态说掉就掉、消息通道随时可能被收紧、防撤回本质上是在协议边缘做补发。把这三件事想明白,再往后看配置和踩坑才有意义。
2. 让机器人跑起来:登录链路、Hermes 配对码与常驻进程
2.1 先看清消息通道:它走的是 Web 协议,不是注入 hook
拆微信机器人,技术路线基本分三类。第一类是基于浏览器 Web 微信的 WebSocket 协议,典型代表是 itchat,靠扫码登录,接收消息事件后做本地处理。第二类是基于手机客户端的 hook 注入,俗称 Pad 协议,最稳但 token 成本高。第三类是企业微信官方 webhook 机器人,最稳但只能单向推送群消息。这份压缩包从代码风格和依赖清单看,走的是第一种:网页协议加本地运行。扫码登录后,机器人用本地 WebSocket 通道接收所有消息事件。
为什么选这条路?因为它自定义程度最高。服务端没有限制你能接到哪些消息类型,文本、图片、语音、系统通知都能进回调;企业微信 webhook 只能推到群里,做不了双向交互。代价也很直接:所有消息事件都由本机进程处理,进程挂了,机器人的记忆就停留在重启前的最后一刻。
解压后目录一般是这个结构,我先给一份尽量小但对得上号的文件地图:
wechat-bot/ ├── main.py # 入口:注册消息监听并启动常驻进程 ├── auto_reply.py # 自动回复规则引擎 ├── forward.py # 消息转发器 ├── anti_recall.py # 防撤回拦截模块 ├── stats.py # 留言统计模块 ├── config.json # 全局配置文件 ├── requirements.txt # 依赖清单 └── runtime/ # 运行时目录:存 token、统计缓存main.py 里的核心启动段不长,我第一次跑几乎原样搬下来,只在日志输出上做了改动:
# main.py 核心启动段(截取自项目 main.py 的运行入口) import itchat from itchat.content import TEXT, PICTURE, RECORDING, NOTE @itchat.msg_register([TEXT, PICTURE, RECORDING, NOTE]) def unified_entry(msg): # 所有消息先落到本地日志,再按类型分发给后面的模块 print(f"[{msg.createTime}] {msg.user.nickName} -> {msg.type}") return route_message(msg) # route_message 在 auto_reply.py 里定义 itchat.auto_login(hotReload=True, enableCmdQR=2) itchat.run()这段代码在整个项目里扮演的是总线角色。@itchat.msg_register 把消息回调绑定到 unified_entry 函数上,TEXT、PICTURE、RECORDING 分别对应文本、图片、语音三类常规消息,NOTE 则专门承接系统通知,比如“某某撤回了一条消息”“某某邀请你加入群聊”。这些类型枚举不是随便选的:防撤回模块依赖 NOTE 触发,转发模块依赖 TEXT 和 RECORDING 接收,少了任何一个注册项,对应功能会在运行时静默失效,而且不会报错。itchat.auto_login 里的 hotReload=True 表示把登录 token 序列化到本地文件,进程重启后能复用登录态,不必反复扫码。enableCmdQR=2 是给无图形界面服务器用的,二维码会以字符画形式打印在终端里。
启动方式没什么悬念,两条命令就能跑起来:
# 安装依赖(注意 itchat 在 Python 3.9+ 有兼容问题,requirements 里要锁版本) pip install -r requirements.txt # 前台启动,先把日志看明白再谈后台守护 python main.py依赖安装阶段我通常会把 requirements.txt 里的 itchat 版本锁死。常见翻车点是拉到旧版 itchat 后,WebSocket 握手直接报错或连不上,第五节会专门讲排查。
2.2 第一次登录:扫码、Hermes 配对码与字符画二维码
这个点很容易被新手当成玄学:部分基于 Web 协议二次开发的机器人在扫码之后会额外输出一串短码,英文叫 pairing code,在项目日志里对应的是 Hermes pairing code。不要跳过它,也不要以为它必填。它的作用相当于一次性的设备绑定确认码:手机微信扫码的同时,机器人客户端把这串 code 回传给服务端,用来标记“这个设备确实是本人授权登录的”。配对码有有效期,常见的实现是 60 秒左右,过期以后必须重新扫码,不是输错重试几次就能绕过的。
我第一次在无图形界面服务器上跑这类项目时,卡了十几分钟:二维码字符画是打印出来了,但太模糊,手机扫了几次都没识别。后来翻代码才知道,enableCmdQR 不只是开关,不同取值会影响字符画的分辨率,配合终端字体调整后再扫,成功率明显提升。这一步解决的是“字符画二维码分辨率太低”的问题,扫码反映到终端会特别清晰。
配对码的坑还不止扫码这一次。有部分机器人项目把配对码的校验结果写进了 token 文件,也就是说,如果你在 A 终端扫码拿到了配对码,再去 B 终端用同一份 token 启动,服务端会判断设备不一致,直接让会话失效。这解释了为什么同一份压缩包在本地跑得好好的,迁移到云服务器就反复掉线。解决方法是“新机器必须从零扫码”,不要拷贝别的机器人的 token 硬启动。
登录完成以后,紧接着要解决进程守护问题。个人微信机器人的连接本质是一条常驻 WebSocket,网络波动或服务端重启都会让它断开。常见做法是挂 systemd 服务,或者用 nohup 加定时重启脚本做简易守护:
# 用 nohup 起后台进程,并配合 restart.sh 做进程守护 nohup python main.py >> runtime/bot.log 2>&1 & # restart.sh:每 60 秒检测一次主进程,不在就拉起来 while true; do if ! pgrep -f "python main.py" > /dev/null; then python main.py >> runtime/bot.log 2>&1 & fi sleep 60 done这段守护脚本的粒度控制在分钟级。原因是 Web 协议的心跳间隔一般在 30 到 90 秒,一分钟检测一次能基本保证掉线后及时拉起,又不会因为频繁重启造成 token 文件竞争。注意 sleep 60 不是拍脑袋定的:拉起太快会导致两个进程同时抢同一个 token 文件,微信端会判定异常登录并强制下线。
2.3 依赖锁版本与 token 隔离:两个最容易被忽略的配置
requirements.txt 在资源包里通常只有三行左右,但每行都有讲究。我见过无数次“装完依赖机器人连不上”的问题,最后都追溯到版本没锁:
| 依赖包 | 推荐约束 | 原因 |
|---|---|---|
| itchat | ==1.2.2 或项目自带版本 | 新版对 Web 协议握手做了调整,旧版更兼容 |
| requests | >=2.20 | 登录二维码下载依赖它,版本太旧会报 SSL 错 |
| schedule | 任意稳定版 | 定时任务和健康探测用,版本影响小 |
token 隔离的意思是:每个微信号配一套独立的 runtime 目录,不能多个微信号共用 runtime/itchat_token.pkl。两个进程同时写同一个 token 文件,轻则反复掉线,重则触发账号保护。这个坑我在多开测试时踩得很深,后面章节会再说。
3. 自动回复与消息转发:把规则写进 config.json,别改代码
3.1 自动回复:三种匹配模式与 delay 参数
这个项目里自动回复的入口在 auto_reply.py,但它真正读取的是 config.json 里的 auto_reply 数组。配置文件驱动的设计在运营场景里特别实用:改回复话术不需要碰代码,运营自己编辑 JSON 就能生效。最常用的规则片段长这样:
{ "auto_reply": [ { "name": "关键词-价格", "match_type": "keyword", "keyword": "价格", "reply": "标准报价 199 元/年,批量有折扣,具体私聊我。", "delay_seconds": 2, "enable": true }, { "name": "正则-微信号", "match_type": "regex", "pattern": "加我[微vV]信[::]?([a-zA-Z0-9_-]+)", "reply": "我的微信号是:$1,添加请备注来源。", "delay_seconds": 1, "enable": true }, { "name": "免打扰时段", "match_type": "timewindow", "start": "23:00", "end": "08:00", "reply": "人工在线时间为 9:00-22:00,留言稍后回复。", "delay_seconds": 0, "enable": true } ] }这些参数在代码里几乎是一一对应的。match_type 决定走哪条匹配分支:keyword 是子串命中,规则里含“价格”两个字就会触发;regex 走正则匹配,适合拦截微信号、手机号这类半结构化信息,$1 代指正则第一个捕获组的内容;timewindow 是时间窗口规则,注意它放在数组第三条,匹配时是顺序从上到下,前面的规则一旦命中,后面的规则不再执行,所以免打扰规则必须放在最后,否则它会挡掉白天所有自动回复。delay_seconds 代表回复前的模拟人工延迟,单位秒,设成 0 会让机器人秒回,风控特征会很明显,我一般最少给到 1~2 秒。
落到代码上,auto_reply.py 的匹配逻辑就是循环读数组,按 enable 字段过滤,再按顺序返回第一条命中结果。三个参数最容易改错:一是 keyword 用的是子串匹配,不是完全相等,规则里写“价格”会连同“价格表”“价格政策”一起命中;二是 regex 规则在 JSON 里写反斜杠要双写,不然解析直接异常;三是 timewindow 如果跨天(比如 23:00 到 08:00),代码默认不会自动跨天判断,需要自己把规则拆成两条。
完整匹配逻辑我可以还原成下面的简化伪代码,方便理解执行顺序:
# auto_reply.py 规则匹配核心(简化还原) import json, re, time with open("config.json", encoding="utf-8") as f: config = json.load(f) def auto_reply(msg_text): for rule in config["auto_reply"]: if not rule.get("enable", True): continue if rule["match_type"] == "keyword" and rule["keyword"] in msg_text: time.sleep(rule.get("delay_seconds", 0)) return rule["reply"] if rule["match_type"] == "regex" and re.search(rule["pattern"], msg_text): time.sleep(rule.get("delay_seconds", 0)) return rule["reply"] if rule["match_type"] == "timewindow": now = time.strftime("%H:%M") if rule["start"] <= now <= rule["end"]: time.sleep(rule.get("delay_seconds", 0)) return rule["reply"] return None执行顺序的问题就在于“命中即返回”。如果你把正则规则写在关键词规则前面,那么“价格”两个字会先被正则里的某些模式命中,关键词规则根本没机会执行。配置规则时,要把宽泛规则放后面,把精确规则放前面。我习惯的顺序是:精确关键词 → 正则 → 兜底话术 → 免打扰。
3.2 转发链路:目标群、黑名单与并发限速
转发模块解决的问题更具体:把核心群的消息同步到工作群,把特定联系人发来的消息推给助理。forward.py 里我一般会调这样一段:
# forward.py 转发策略参数化写法 import itchat # source_group 是监听源,target_users 是转发目标列表 forward_rule = { "source_group": "客户对接群", "target_users": ["助理-小王", "商务-外部备份群"], "message_types": ["TEXT", "PICTURE"], "ignore_text": ["签到", "打卡"], "skip_bot_self": True } def route_forward(msg): if msg.user.nickName != forward_rule["source_group"]: return if msg.type not in forward_rule["message_types"]: return if msg.text and any(k in msg.text for k in forward_rule["ignore_text"]): return for target in forward_rule["target_users"]: itchat.send(msg.text or msg.file_name, toUserName=target)这个片段把转发逻辑拆成了四层过滤器:先判断消息源是不是目标群,再判断消息类型,然后做关键词黑名单过滤,最后循环发送到目标。PICTURE 这类媒体消息转发时,代码里用的是 file_name 拼出的缓存路径;ignore_text 是给群里的签到、打卡这类高频噪音准备的;skip_bot_self 字段在原始代码里作用是跳过机器人自己发到群里的消息,防止死循环转发。
需要注意,target_users 里既能写个人联系人昵称,也能写群名,但重名问题会导致 itchat.search_chats 返回多个结果。实际使用强烈建议在配置里直接写备注名或群 ID,不要写昵称。另一个隐蔽的坑是媒体消息:图片、语音在 Web 协议下会先落成临时文件,拿到的是本地路径。如果你把机器人部署在 A 服务器,却在 B 机器上看日志,这个路径在你本地不存在。媒体转发必须依赖同一台机器上的文件缓存,不要把缓存路径二次加工,直接把完整路径交给 itchat.send 即可。
提示:转发规则的执行与自动回复规则相互独立,二者不会阻塞,但都会占用同一个 WebSocket 发送通道。消息量一大,同时刻多个 itchat.send 会被服务端限流,表现为发送超时。务实经验是转发的目标控制在 5 个以内,超过就分批发,每批间隔 1 到 2 秒。
3.3 热加载规则:改配置不重启进程
配置文件驱动最大的好处是可以热加载。很多运营同学一开始是改完 config.json 就重启进程,结果每改一次掉线一次。我这里更推荐的做法是给配置加文件监听,检测到 config.json 的修改时间变化后自动重新加载规则:
# config_watcher.py 监听配置文件变化并热加载规则 import os, json, time, threading last_load_time = 0 global_rules = {} def load_config(): global global_rules, last_load_time with open("config.json", encoding="utf-8") as f: global_rules = json.load(f) last_load_time = os.path.getmtime("config.json") print("config reloaded at", time.strftime("%H:%M:%S")) def watch_config_loop(): while True: mtime = os.path.getmtime("config.json") if mtime != last_load_time: load_config() time.sleep(2) # 启动时先加载一次,再挂后台线程监听 load_config() threading.Thread(target=watch_config_loop, daemon=True).start()这段监听逻辑的核心是拿文件的修改时间快照做对比,而不是每次都重新读盘。2 秒的轮询间隔对配置类文件足够,并不会占用额外资源。注意 watch 线程用 daemon=True 挂后台,主进程退出时不会因为线程还活着而卡住。
每次改规则后,我验证是否生效的方式是直接给机器人发一条测试消息。如果命中日志出现在 runtime/bot.log 里但回复内容还是旧的,那说明配置在内存里被缓存了,需要检查 load_config 是否被正常触发。
4. 防撤回与留言统计:把黑匣子拆开看,才知道边界在哪
4.1 防撤回的真实逻辑:它是“先缓存,再补发”
很多人误以为防撤回是直接调协议接口阻止对方删除。实际上,几乎所有基于 Web 协议的机器人做的都是同一件事:消息到达时先把原文和元信息存在本地,一旦收到“对方撤回了一条消息”的系统通知,就触发补发动作,把本地缓存原文再发一遍。anti_recall.py 的核心流程可以还原成这样:
# anti_recall.py 防撤回模块的核心缓存逻辑 import itchat, re, time from itchat.content import NOTE recall_cache = {} @itchat.msg_register(NOTE) def watch_recall(msg): # 系统通知内容形如 "xxx撤回了一条消息" content = msg.content if "撤回了一条消息" not in content: return m = re.search(r"(.+?)撤回了一条消息", content) if not m: return nickname = m.group(1) cached = recall_cache.get(nickname) if cached: itchat.send(f"[防撤回] 对方撤回了:{cached['text']}", toUserName=msg.user.userName) # 补发完立刻清除缓存,避免同一段话重复补发 del recall_cache[nickname] # 所有文本消息先写进缓存,供防撤回模块查询 def cache_text(msg): if msg.type == "TEXT": recall_cache[msg.user.nickName] = {"text": msg.text, "ts": time.time()}这段代码的运行前提是把 cache_text 挂到全局消息入口。实际操作里有一个关键点:缓存键用的是昵称而不是用户 ID。微信 Web 协议里 msg.user.userName 拿到的是不直观的微信号 ID,而系统通知文本里只出现昵称。用昵称做键会有同名误伤,两个人同名时后发消息会覆盖前一个人的缓存,导致撤回补发时内容对应错人。改法是用用户 ID 倒查最近一条消息,昵称只用来匹配系统通知。我在业务里通常维护一个 userId 到最新消息的映射,再用系统通知里的昵称做二次确认。
防撤回的适用范围要说清楚:只对机器人自己所在的会话有效。文本消息没问题,图片和语音会有限制,因为媒体 ID 过期时间一般在 3 天左右。如果撤回发生在缓存过期之后,补发就会失败。这个限制无解,是底层协议规定的,不是代码 bug。
4.2 媒体消息与 ID 过期:两个硬性边界
防撤回模块处理图片、语音时,本质是拿到消息的 mediaId 和类型。补发时 itchat 会先根据 mediaId 去缓存文件,如果文件已经被清理,补发出来的消息就是空的。有一些机器人项目会在收到媒体消息时立刻下载到本地,防撤回时再把本地文件作为附件发送,这是一种可用的补强方案,但会有两个代价:本地磁盘占用快速增长、下载动作本身会加长消息处理链路。
我在实盘部署时给防撤回模块定的底线是:只承诺文本消息的防撤回成功率,媒体消息保留“能补发但可能为空”的降级表现。这样运营反馈预期也清晰,不会拿图片丢失来问责。如果你需要批量保护图片,更现实的做法是单独对重要群开启全部媒体自动下载,而不是与防撤回耦合在一起。
4.3 留言统计:pickle 存储、进程锁与导出
留言统计模块在 stats.py 里,做的事情本质是把经过消息链路的记录去做落库和聚合。原项目存储方式用 pickle 写到 runtime/msg_stats.pkl,每条消息落一次用户名、时间、类型、文本四个字段。核心统计逻辑可以简化成这样:
# stats.py 留言统计与汇总 import pickle, time, os from collections import defaultdict class MsgStats: def __init__(self, path="runtime/msg_stats.pkl"): self.path = path self.data = defaultdict(list) if os.path.exists(path): with open(path, "rb") as f: self.data = pickle.load(f) def record(self, msg): self.data[msg.user.userName].append({ "time": time.strftime("%Y-%m-%d %H:%M:%S"), "nick": msg.user.nickName, "type": msg.type, "text": msg.text }) self._flush() def _flush(self): with open(self.path, "wb") as f: pickle.dump(dict(self.data), f) def summary_by_user(self): out = {} for uid, items in self.data.items(): out[uid] = { "nick": items[-1]["nick"], "count": len(items), "last_time": items[-1]["time"] } return out这个模块看着简单,问题藏在落盘策略里。原始代码可能每一条消息都直接 pickle.dump,消息量上去以后会拖慢整个消息链路,导致防撤回缓存和自动回复同时被卡。我的改进是每 10 条或每 30 秒写一次盘。另一个隐蔽坑是 pickle 多进程冲突:守护脚本拉起第二个进程时,两个进程同时写 msg_stats.pkl 会把文件写坏,需要给统计模块加进程锁:
# stats.py 加进程锁的写盘方法,避免多开抢占 import fcntl def _locked_flush(self): with open(self.path + ".lock", "w") as lock_f: fcntl.flock(lock_f, fcntl.LOCK_EX) with open(self.path, "wb") as f: pickle.dump(dict(self.data), f) fcntl.flock(lock_f, fcntl.LOCK_UN)加锁之后,多开场景下统计模块不会因为写盘竞争而丢数据。留言统计的业务出口一般是日报或周报,项目本身没有图表,但可以做一个极简文本汇总:
# summary.py 生成文本版留言汇总 for uid, info in stats.summary_by_user().items(): print(f"{info['nick']}: {info['count']} 条,最后发言 {info['last_time']}")把这段输出重定向到文件,配合定时任务每天凌晨跑一次,就是一份不需要前端的运营活跃日报。
统计维度的扩展也很方便。如果你想按时间段统计,比如只看每天 9 点到 22 点之间的留言,可以给 record 加一个时间过滤参数,把落在时段外的消息只计数不入明细,这样日报会更贴近运营人力覆盖时段。如果想把统计导出成表格,用 sqlite 替代 pickle 会更合适,但代价是需要维护一张表结构和索引,项目里没有现成代码,属于进阶改造范围。
5. 微信机器人避坑:登录掉线、消息收不到、防撤回翻车,三条血泪经验
5.1 半夜掉线:hotReload 救不了
现象:机器人白天工作正常,凌晨两三点以后大概率静默离线,第二天早上看日志只有一条 Connection closed,或者干脆没日志。
原因:微信服务端对 Web 协议长连接的心跳检测比较灵活,凌晨低流量时段更激进。hotReload 保存的是登录态 token,不是网络连接;网络断开会话失效后,重启进程虽然可以复用 token,但 token 本身也有有效期。另外一个常见诱因是服务器网络出口的 NAT 超时,空闲连接超过一定分钟数就会被切断。
解决:不要指望 hotReload 是后悔药,它只能省扫码。要长期在线,一是加主动心跳,用 schedule 库每 20 秒调用一次 itchat 的心跳接口;二是在守护脚本里加定时重启,比如每 6 小时 kill 一次主进程再拉起,主动换 token 刷新。实际测试里,6 小时一轮重启的稳定率远高于 24 小时不重启。
5.2 消息静默收不到:被 mute 还是被风控
现象:机器人能收到自己主动发的消息,但别人发来的消息一直不出现,日志里没有任何报错。
原因:一多半是联系人或群开了消息免打扰,免打扰群的系统推送不会进入 Web 消息流;另一半是微信对高频操作账号做了降级处理,表现是消息延迟推送或干脆不推。注意这里的“不推”是服务端不推,不是机器人没收到。
解决:第一步检查目标群是否开启免打扰,关掉再试。第二步看日志里 NOTE 类型消息是否还在;如果 NOTE 还在但 TEXT 不进来,基本就是风控。风控没有代码层解法,只能降频:把自动回复 delay_seconds 提高到 1.5 秒以上,限制每 10 秒最多发 3 条消息,暂停转发高活跃群。
5.3 防撤回补发空内容
现象:别人撤回了消息,机器人提示“对方撤回了:”,后面是空白,或者干脆没触发补发。
原因:缓存写入时机晚于撤回通知。Web 协议在弱网环境里消息顺序可能乱序,撤回通知先于原文到达,缓存里查不到内容。
解决:在撤回通知到达后加一个 0.5 到 1 秒的短等待,再查缓存,给原文消息一个到达窗口。如果还是拿不到,降级记录“撤回时间加来源用户”,避免发送空内容。还有一个容易翻车的点:机器人自己撤回的消息也会触发 NOTE 通知,要跳过机器人自己,否则会出现机器人自己撤回又被机器人自己补发的循环。
5.4 企业微信机器人是另一条路,别混用
现象:拿这套个人微信机器人的代码去跑企业微信群机器人,登录直接失败,或者消息发不出去。
原因:个人微信机器人走 Web 协议,企业微信机器人走官方 webhook API,两者会话体系完全不同。代码里 itchat.search_chats 这类接口在企业微信环境下根本不存在,强行套用必然报错。
解决:如果需求只是单向投递,比如 Zabbix 告警推送、服务器监控通知发到企业微信群,直接改用企业微信机器人 webhook,官方接口稳定且不需要登录态,更不需要扫码。企业微信群机器人个人号本质上就是一个 Webhook URL,配好之后 curl 就能发消息。个人微信机器人只保留在需要双向交互的场景里,两套逻辑拆开部署,不要混在一个进程里。
这个区分很重要。很多人看到“机器人”三个字就默认是一套代码通吃,其实个人微信机器人和企业微信群机器人是完全不同的技术生态,前者偏交互,后者偏通知。接到需求时先问一句“要不要接收用户回复”,再决定走哪条路,能省很多折腾时间。
6. 上线前跑一轮验证:规则命中、通道延迟、统计落盘,一步都别省
一个机器人配好不等于能上线。我部署这套东西后的习惯是,先花一个下午把三条链路逐一验证,全部跑通了再进生产。第一个要验的是自动回复的命中顺序。把 config.json 里的规则调成只留一条,发一条测试消息看是否命中;然后逐步加回规则,每加一条就重复一次,确认顺序没有遮蔽。常见问题是“价格”同时触发关键词和正则规则,正则规则虽然排在后面,但命中后依然会返回——原因在于代码是顺序匹配而不是优先级匹配,所有规则一律先到先得。你把规则数组的顺序当成优先级列表用,就没问题。
第二个要验的是消息传输延迟。微信服务端本身有快通道和慢通道之分,文本消息从对方发出到机器人日志打出来,正常应该在 1 到 2 秒内;如果超过 5 秒,说明通道被风控或网络节点不通。这个延迟无法在代码层完全解决,只能换网络节点和降低频率。我习惯把下面这段定时探测加进守护脚本,每 30 分钟给机器人自己发一条测试消息,记录延迟:
# probe.py 定时给机器人自己发心跳消息,验证消息链路 import itchat, time from datetime import datetime def health_probe(): start = time.time() # 给机器人自己发一条文本消息,通道正常的话它会收到并回执 itchat.send("health-check", toUserName=itchat.get_friends()[0]["UserName"]) elapsed = time.time() - start print(f"[{datetime.now()}] channel delay: {elapsed:.2f}s") return elapsed < 2.0这段健康探测的意义在于,把“机器人进程是否活着”和“消息通道是否通”分开。很多问题表现为机器人没有宕机,但消息就是发不出去,本质是通道已经断掉而进程没退出,用健康探测一秒就能定位。
第三个要验的是统计落盘一致性。跑完一轮测试数据后,对比 runtime/msg_stats.pkl 里的条数和实际收到消息数,偏差如果超过 1%,优先排查统计模块是不是被进程锁挡掉了写入。再检查多开场景:同一时间只允许一个主进程写统计文件,另一个进程的写入会被拒绝,这是设计如此,不是 bug。最后一个验证是连续跑满 72 小时,期间不做人工干预。72 小时能覆盖两个深夜时段,最容易暴露心跳掉线。我第一次跑的时候,在第 30 小时收到掉线告警,排查后确认是服务器 NAT 超时切断了空闲连接,之后在守护脚本里加入每 6 小时重启一次主进程,问题再没复现。
从那以后,我每次部署微信机器人都会强制走一遍这三件事:规则命中顺序、通道延迟探测、统计落盘校验,再缓两天才敢放心让它挂机运转。希望这份拆解对你有用,也祝你少踩我踩过的这几个坑。
本文还有配套的精品资源,点击获取