简介基于 Web API 的微信机器人项目面向需要给个人号或社群账号扩展自动化能力的开发者覆盖自动回复、消息转发、防撤回、留言统计等常用功能并支持按实际使用场景灵活调整触发规则。压缩包体积仅 71KB属于开源代码类工程资源结构简洁、无冗余素材适合具备一定编程基础、希望快速搭建微信辅助工具的中初级开发者学习与二次改造。目前已有 1107 人学习下载。通过这份源码可以梳理微信网页接口的登录、消息收发与事件处理封装方式看清自动回复与消息转发的完整调用路径也能理解消息类型判断与规则匹配的设计思路防撤回、留言统计等模块相对独立抽出后可直接迁移到其他会话类自动化项目中省去从零调试接口的时间。整个工程规模小巧是一份能快速上手的轻量级实战样例。1. 高度自定义的微信机器人到底能做什么自动回复、消息转发、防撤回、留言统计一把抓如果你手里有几个客服群、工作群或自己的私域流量池手动回复到半夜真的是体力活。高度自定义的微信机器人这个标题背后其实就是一套把自动回复、消息转发、防撤回、留言统计统一收编的框架代码。拿到手之后不是改个配置文件那么简单而是把消息回调当成业务入口所有功能都挂在同一条事件链上。这套东西最实用的场景是白天客服不在机器人先兜住“在吗”“多少钱”“怎么退款”这类高频问题群里有人发广告或敏感词自动转发到监察群留证同事撤回了重要通知防撤回模块把原文送到备份群晚上再做一次留言统计谁今天最活跃、哪个群聊咨询最多直接看报表。适合谁有一定 Python 基础、想给公众号或企业微信号做自动化运营的开发者不想从微信协议底层再造轮子。新手按步骤能跑通老手可以直接改 handler 加业务。下面我把路线选择、部署方式和踩过的坑揉在一起讲。2. 先定技术路线为什么自动回复和防撤回往往不是同一个方案在打开 zip 包之前先把底层机制搞清楚。很多人拿到代码第一反应是“装上就能用”结果登录半小时就开始掉线这就是没做选型导致的。微信机器人不是一套方案通吃所有功能不同能力对微信端的侵入深度不一样。2.1 三条路线Web 协议、桌面 Hook、企业微信 webhook 各自能干什么先给自己泼一盆冷水网页微信协议在 2023 年后大面积失效老一批基于 itchat/wxpy 的项目十个里有八个都登不上。标题里的 zip 如果还停留在 web 协议那大概率只能当研究源码用。我一般会把方案拆成三条路第一是桌面 Hook 路线也叫做 UI 自动化路线。它操作 Windows 上已经登录的微信桌面客户端通过读取界面元素和消息列表拿数据点击控件来发送消息。好处是能拿到撤回事件、完整聊天记录、图片文件避免 web 协议限制坏处是强依赖 Windows 环境、微信版本和窗口焦点后台挂机时偶尔会卡界面。第二是web 协议路线优点是代码简单、适合快速验证缺点是登录风控严重、扫码掉线频率高。如果 zip 里只给了itchat依赖那这套东西只能作为开发参考不建议直接在主力账号上跑。第三是企业微信 webhook 机器人。这个路线最稳因为它不需要登录个人号只是在群里加一个机器人通过 HTTP POST 推送消息。但它的能力范围很窄只能往外发接不到用户发来的消息更做不到防撤回。如果你的需求只是把 Zabbix 告警推到企业微信群用官方群机器人就够了根本不用碰个人号。关于热搜里那个“企业微信机器人个人号”的说法我要提醒一句企业微信和普通个人微信是两套体系机器人 webhook 并不能直接“接管个人号”。个人号想要自动回复还是得走 Hook 或消息回调中间层这也正是“高度自定义”这个标题里最有含金量的部分。我用一张表总结这三条路的取舍路线能做到接不到/做不到稳定性桌面 Hook自动回复、转发、防撤回、统计需要 Windows 常驻中高依赖微信版本Web 协议自动回复、转发防撤回困难、登录困难低频繁失效企业微信 webhook群内推送通知接不到用户消息高官方支持2.2 统一消息事件模型让不同功能的代码长在同一棵树上hook 方案拿到的消息格式五花八门有的是 JSON有的是 XML还有的是直接从窗口读出来的纯文本。如果每个功能各自解析一遍后期一定会翻车。所以我会在入口处强制转换成一个统一的事件对象再分发给下游 handler。下面这段代码就是事件模型的典型写法也是 zip 项目里最常出现的骨架class WeChatEvent: def __init__(self): self.type None # text / image / voice / recall /file self.msg_id # 全局唯一消息 ID self.content # 文本内容或文件路径 self.from_user # 发送者昵称 self.room_id # 群 ID私聊为空 self.create_time 0 # 消息产生时间戳逻辑说明这是把底层微信 SDK 回调的不同字段统一成一种格式后续自动回复、转发、防撤回、留言统计都只依赖这个类不直接依赖底层库。这样做的好处是哪天从桌面 Hook 切到另一种框架只需要改底层适配器上面所有 handler 一行不用动。参数说明type字段里我单独加了recall因为撤回事件在很多 hook 框架里没有独立字段必须通过监听“撤回通知消息”识别再映射到这个枚举里。room_id是区分群聊和私聊的关键转发逻辑依赖它做定向分发。如果你发现收不到撤回事件先排查这一层映射是不是被微信新版本改了消息格式。2.3 防撤回的缓存设计消息摘要何时写入、何时清理防撤回看起来像是“黑匣子”其实核心逻辑就两步先把每条消息偷偷存一份然后再收到“撤回通知”时把存下的内容重新发到指定群。这里最关键的不是怎么重发而是往哪存、存多久。我一般会在内存里用 LRU 缓存而不是直接落数据库。因为撤回只会发生在最近几十秒内聊天内容实时写入 SQLite 反而容易造成 IO 瓶颈。缓存结构大致是这样from collections import OrderedDict class MessageCache: def __init__(self, max_items5000): self.cache OrderedDict() self.max_items max_items def push(self, msg_id, event): if msg_id in self.cache: self.cache.pop(msg_id) self.cache[msg_id] event while len(self.cache) self.max_items: self.cache.popitem(lastFalse) # 淘汰最老消息 def pop(self, msg_id): return self.cache.pop(msg_id, None)逻辑说明OrderedDict在这里相当于一个自带淘汰策略的消息暂存区超过max_items就最老的先丢避免内存无限增长。防撤回只需要保留最近一段时间微信撤回通知本身携带原始消息 ID拿着 ID 到这个缓存里查查到了就重发查不到就说明已经超时或消息被淘汰。参数说明max_items我习惯设 5000按每天几百条消息来算足够存半天以上。如果你所在的群特别活跃需要调大到 10000 甚至更高反之如果机器人的内存只有 512MB建议调低到 1000防止被刷屏拖垮。加缓存时间上限会比只按条数淘汰更精准def is_expired(self, msg_id, expire_seconds300): event self.cache.get(msg_id) if not event: return True return time.time() - event.create_time expire_seconds这里expire_seconds300表示只防 5 分钟内的撤回因为微信的撤回窗口就是 2 分钟5 分钟绰绰有余。设太长没有意义只会让缓存越来越臃肿。3. 解压 zip 跑通最小实例自动回复与消息转发的第一版拿到 zip 之后不要急着双击跑。我习惯先把它当成一个陌生项目来做环境隔离跑通最小闭环后再往里面加防撤回和统计功能。最小闭环就是“收到消息 → 自动回复 → 按关键词转发”。3.1 解压、建虚拟环境、装依赖先把 zip 解压到一个纯英文路径下比如D:\wxbot或~/wxbot避免中文路径在 Windows 上造成的编码问题。然后在项目根目录建虚拟环境cd wxbot python -m venv venv # Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate pip install -r requirements.txt逻辑说明requirements.txt里一般会包含底层 hook 库、PyYAML 配置解析库、requests 网络请求库。如果你的项目里没有这个文件那就先手动装三个最核心的PyYAML、requests、jsonpath。底层 hook 库每家项目不一样直接装项目声明的那一个。参数说明Python 版本尽量用 3.8 到 3.10太新版本可能与底层 hook 库的二进制扩展不兼容。虚拟环境这一层不能省因为微信 hook 经常需要特定版本的pywin32放在全局环境里很容易和别的项目打架。3.2 编辑 config.yaml自动回复规则与转发规则的参数设计“高度自定义”落地时看的就是配置文件。避免把逻辑写死在代码里优先把自动回复规则、转发规则、监控群列表抽到 YAMLauto_reply: enable: true rules: - keywords: [在吗, 你好] reply: 你好我是自动助手稍后人工来回复你 - keywords: [价格, 多少钱] reply: 基础套餐 199具体资料我发你文档 default_reply: 已记录客服会尽快联系你 forward: watch_rooms: - crm_1群 - crm_2群 to_room: 客服备份群 keywords: [退款, 投诉, 差评] append_sender: true逻辑说明auto_reply.rules从上到下匹配命中第一条就不再往后走所以规则顺序就是优先级顺序越精确的关键词越要放前面。default_reply是兜底文案避免用户发一句无关话机器人没反应。forward里append_sender: true会在转发内容前面加上原发送者昵称方便追责。参数说明keywords默认做子串匹配所以“投诉”能命中“我要投诉你们”。“退款”这种词会误伤聊到“退款流程”的普通对话实际使用时要谨慎。我建议加一个use_regex: true扩展项用正则做更精准的匹配比如商品编号[A-Z]{3}\d{4}。watch_rooms不是填群昵称而是填 hook 层返回的群 ID具体值先跑一次日志输出再回填。3.3 写 handler 并启动让机器人处理消息、按关键词转发核心 handler 放在handlers.py里如下def handle_text(event): rules config[auto_reply][rules] for rule in rules: if any(keyword in event.content for keyword in rule[keywords]): bot.send(event.room_id or event.from_user, rule[reply]) return bot.send(event.room_id or event.from_user, config[auto_reply][default_reply]) def handle_forward(event): if event.room_id not in config[forward][watch_rooms]: return content event.content if any(x in content for x in config[forward][keywords]): prefix f[{event.from_user}] if config[forward][append_sender] else bot.send(config[forward][to_room], prefix content)逻辑说明handle_text先查自动回复命中后直接用原会话发送send 的接收方是房间 ID 或用户 ID避免回错人。handle_forward只处理被监控的群并且只挑包含关键词的消息同时在内容前拼上发送者信息。两个函数入口是独立的方便后续拆分进程或做异步队列。主程序入口长这样from wxbot_driver import create_bot from handlers import handle_forward, handle_text import yaml config yaml.safe_load(open(config.yaml, encodingutf-8)) bot create_bot(config.get(driver, {})) bot.register(text, handle_text) bot.register(text, handle_forward) bot.run(blockTrue) # 扫码登录后开始监听逻辑说明create_bot是根据配置文件里的驱动类型创建底层连接对象有可能是桌面 Hook 驱动也可能是 web 协议驱动。register把两个 handler 挂到 text 事件上消息到达时会按注册顺序先执行转发、再执行回复。run(blockTrue)会阻塞主线程保持进程一直挂机。登录时如果框架支持可以用手机号配对码登录部分 hook 框架登录时会输出一个配对码类似 Hermes 那种配对机制配对成功后手机会退到后台不需要一直亮屏。这里注意配对码登录也有失效时间掉线了就得重新配对后面避坑章节会细说。跑到这里你已经拥有了可以自动回复和按关键词转发的机器人。下一步才是这个 zip 文件比普通模板值钱的地方防撤回和留言统计。4. 把默认机器人改成你的私人订制防撤回、留言统计与自定义扩展标题里“高度自定义”才是重点。默认机器人只解决“有”的问题要做到真正能放进生产环境还得把防撤回、留言统计和外部接口接缝补实。4.1 防撤回模块消息缓存、撤回捕获与重发防撤回模块和 2.3 的消息缓存配合在收到普通消息时写入缓存在收到撤回事件时取出重发。这里给出一个可在现有框架上直接移植的类class RecallPlugin: def __init__(self, cache, forward_room): self.cache cache self.forward_room forward_room def on_normal_message(self, event): self.cache.push(event.msg_id, event) def on_recall_notice(self, recall_event): original self.cache.pop(recall_event.recall_msg_id) if not original: return bot.send(self.forward_room, f[防撤回] {original.from_user} 撤回了{original.content})逻辑说明on_normal_message在每一条消息进入时都调用一次把消息 ID 和事件对象按绑定关系存进缓存。on_recall_notice拿到撤回通知后根据recall_msg_id找到原消息如果找到就发到指定备份群。找不到时直接放弃不要尝试给原会话重发因为原会话的撤回提示会对用户可见。参数说明forward_room建议设置成一个只有管理员在的备份群不要转发回原群否则相当于“帮对方恢复撤回”容易激化矛盾。另外图片、视频、文件这类消息event.content里存的是本地文件路径重发时需要先判断路径是否存在再调用send_file而不是send_text。 一个升级做法是把文件复制到专门目录防止原文件被微信自动清理import shutil, os def backup_file(event, backup_dirbackup): if event.type not in (image, video, file): return None ext os.path.splitext(event.content)[1] dst os.path.join(backup_dir, f{event.msg_id}{ext}) shutil.copy2(event.content, dst) return dst逻辑说明backup_file在消息进入缓存前调用把原始文件复制到本地备份目录同时更新event.content为备份后的路径。这样即使微信把临时文件删了撤回时照样能把文件重新发出来。复制失败要记录日志不能影响主流程。需要提醒一句防撤回功能只适合在你自己是群主或管理员、且参与人知情的内部群里使用不要拿去偷窥用户聊天记录。合规边界一定要拎清楚。4.2 留言统计模块SQLite 落库、去重和按天聚合留言统计最怕的是把同一条消息重复计算。微信在断线重连后会对历史消息做一次同步如果你的代码把同步消息也当新消息处理统计数字就会翻车。所以必须用msg_id去重。先建表CREATE TABLE IF NOT EXISTS messages ( msg_id TEXT PRIMARY KEY, room_id TEXT, from_user TEXT, content TEXT, msg_type TEXT, ts INTEGER ); CREATE INDEX IF NOT EXISTS idx_messages_room_ts ON messages(room_id, ts);落库代码如下import sqlite3 def insert_message(event): db.execute( INSERT OR IGNORE INTO messages(msg_id, room_id, from_user, content, msg_type, ts) VALUES (?,?,?,?,?,?), (event.msg_id, event.room_id, event.from_user, event.content, event.type, event.create_time) ) db.commit()逻辑说明INSERT OR IGNORE是消息去重的关键当同一个msg_id第二次进来时数据库会直接忽略不会产生计数翻倍的问题。索引idx_messages_room_ts是为了加快按群和时间段统计的查询数据量过万时必须加。按天聚合的查询是这样的SELECT date(ts, localtime, start of day) as day, room_id, from_user, count(*) as cnt FROM messages WHERE msg_type text GROUP BY day, room_id, from_user ORDER BY day DESC, cnt DESC;参数说明date(ts, localtime, start of day)把时间戳规整到当天零点解决跨时区统计错位。group by三个字段能让你看出“某人在某天的某个群里说了多少条”是最常用的运营报表维度。如果想统计每天总数去掉from_user重新分组即可。统计结果不建议直接推给用户而是每晚固定时间把 top10 排行榜发到管理群。实现很简单先查 SQL再拼成文本用之前handle_forward里的发送逻辑调用一次即可。4.3 自定义命令与第三方 API 接入点“高度自定义”最爽的一刻是你给机器人加一个斜杠命令比如查快递、查天气、拉取内部订单。做法是新增一个命令路由把带/开头的消息分配到命令处理函数def handle_command(event): if not event.content.startswith(/): return None parts event.content.strip().split( , 1) cmd parts[0].lower() args parts[1] if len(parts) 1 else if cmd /stock: return query_stock(args) elif cmd /order: return query_order(args) else: return 未知命令输入 /help 查看支持列表逻辑说明handle_command不直接发送消息而是返回字符串结果由外层统一决定怎么发送。这样做的好处是返回结果可以被测试直接引用不用依赖微信环境。split( , 1)把命令和参数拆开比如/stock 600519cmd是/stockargs是600519。命令函数内部就是普通 Pythondef query_stock(code): endpoint fhttps://api.example.com/stock/{code} resp requests.get(endpoint, timeout5) data resp.json() return f{data[name]} 当前价 {data[price]}涨幅 {data[percent]}%参数说明第三方接口调用必须设timeout不设的话一旦上游接口卡死机器人会被拖到崩溃。我自己习惯分两层超时requests层的 5 秒是网络超时外层再包一个函数级别的 10 秒保险。接口返回异常时要返回兜底文案而不是抛出异常把整个消息循环打挂。命令路由挂到事件分发里def handle_all(event): result handle_command(event) if result: bot.send(event.room_id or event.from_user, result)到这里你已经把 zip 里的代码从一个“能跑的玩具”改成了“有业务逻辑的服务”。下面的坑是我在实际部署中反复踩过的每条都花了不少时间。5. 微信机器人避坑指南登录失效、消息漏发和防撤回失灵这一章直接进血泪经验。好多问题不是功能写错而是卡在微信端和底层 hook 的边界上。5.1 登录一扫码就掉线半小时后又要重扫现象微信扫码登录成功心跳正常但 10 到 30 分钟后进程还在微信那边却已经退出登录或者扫码登录提示“版本过低、请使用手机扫码”。原因高频的设备注册被微信风控识别也可能是桌面 Hook 驱动的内嵌浏览器标识不完整导致服务器判定为非官方客户端。另外如果你频繁切换网络 IP也会加速掉线。解决先把扫码登录的手机号换成小号不要用主号做实验。登录后固定在一个网络环境下挂机不要休眠 WiFi不要频繁插拔手机。如果框架支持配对码登录优先用配对码而不是扫码配对码会话的有效期更长。再不行就在配置里打开“心跳保活”每次心跳间隔控制在 30 到 60 秒模拟真人在线状态。5.2 群消息转发偶尔漏一条跟微信并发回调用例对不上现象10 条转发里偶尔少 1 条不是丢了而是两个回调同时进入handle_forward出现了资源竞争发送接口被后一个回调覆盖。原因很多 hook 框架的回调是线程池模式不同消息分配到不同线程你在处理函数里直接调用发送接口两个线程同时写同一个 socket 连接数据包互相干扰发送失败但没有日志。解决给消息处理加一个线程锁同时把发送操作放到单线程队列里。改动很小from queue import Queue from threading import Lock send_queue Queue() send_lock Lock() def safe_send(target, content): with send_lock: bot.send(target, content)再往上一层可以把所有handle_text用ThreadPoolExecutor(max_workers1)串行跑。消息量没到每分钟几百条时单线程完全够用还从根上避免了并发覆盖。日志里看到send failed时要把失败消息重新入队等 1 秒后再发。5.3 防撤回只挡得住文字图片和视频召不回来现象文字撤回能恢复图片、视频撤回后没有反应或在日志里看到文件路径失效报错。原因文字消息在缓存里存的是字符串图片视频存的是微信临时目录里的文件。微信撤回通知触发时临时文件已经被回收或改名路径失效。还有一种情况是撤回事件里带的msg_id与普通消息里的msg_id不一致导致按 ID 取不到缓存。解决参照 4.1 的backup_file方法在收到媒体消息的第一时间复制文件到备份目录并把备份路径替换进event.content。同时在撤回事件中打印一下recall_event.__dict__看原始消息 ID 是不是藏在recall_msg_id之外的其他字段里不同框架字段名差异很大。熟悉这层结构之后防撤回才算真正落地。5.4 留言统计越算越多同一条消息被重复记录现象数据库里msg_id并没有重复但统计条数明显超过实际消息数而且每天会多出一批固定数量的记录。原因机器人在断线重连、电脑重启后会对最近的历史消息做增量回放。回放时消息内容相同但不同 hook 框架往往会生成新的msg_id导致INSERT OR IGNORE失效。如果你还用时间窗口去重也会把凌晨重放的消息算到前一天。解决除了msg_id再增加一个业务去重键(room_id, from_user, content, ts)回放消息通常会在几秒内连续重放时间窗口内去重可以兜底。更稳定的做法是记录每条消息的create_time每次落库前先按时间戳倒序查最近 5 秒内有没有内容相同的消息有就跳过。注意时间窗口会和真实密集发言冲突所以(room_id, from_user, content, ts)是首选。5.5 自动回复在群聊里变成了“全员复读机”现象设置好了自动回复结果群里三个人同时问不同问题机器人把同一个默认回复发给了所有人或者有人 机器人它把回复发到了整个群而不是私聊。原因自动回复逻辑没有区分私聊和群聊默认回复发送给了event.room_id也就是群聊本身。在群里机器人只需要回复 它的那条消息而不是无差别回复所有内容。解决加一个force_private配置项群聊中的关键词命中后优先给发送者发送私聊消息而不是回复到群里。实现时要先判断event.room_id是否存在如果存在就用bot.send(event.from_user, rule.reply)而不是bot.send(event.room_id, ...)。同时过滤掉所有非 机器人的群聊消息只在成员清单里找到机器人昵称并确认 时才触发自动回复。6. 上线前这样验证双设备冒烟测试、消息链路自检和回归清单部署完成不代表能上线。我习惯在正式接客前做一轮消息链路自检用手机、电脑同时在线把自动回复、转发、防撤回、留言统计四个功能跑一个“假想用户路径”。这一步非常便宜提前发现八成问题。我会先写一段自检代码往文件传输助手发一条测试消息再由机器人自动回一条确认链路通着def smoke_test_driver(): bot.send(filehelper, ROBOT_SMOKE_TEST_START) time.sleep(3) test_event WeChatEvent( typetext, msg_idtest-001, content在吗, from_userself_test, room_id ) result handle_text(test_event) assert result 你好我是自动助手稍后人工来回复你逻辑说明这条验证把业务层和微信层解耦了不需要真的等微信消息回来再判断而是直接构造事件对象去检查我们的回复逻辑。微信层的连通性由filehelper那条消息确认业务层的正确性由构造事件确认。两层分开验证出问题时排查范围小很多。然后做一次完整的模拟用户行为用手机小号往群里发“我要退款”看是否转发到备份群接着在群里撤回这条消息看防撤回模块是否把原消息发到备份群最后查询留言统计表确认这条消息只被记录一次。回归清单我固定为三块自动回复命中规则、命中默认回复、未命中三种情况都要人工验证。转发关键词命中、非关键词不转发、转发条数与群内消息数一致。插件自定义命令/help返回正常第三方接口超时时能兜底返回文案。上线后也不要彻底松手。微信 hook 方案最怕通讯方式悄悄变化我现在的习惯是每天早中晚各看一次心跳日志确认“在线”状态没有被风控踢掉。如果某一天延迟大于 2 分钟我会先查网络再查微信客户端有没有弹出“操作频繁”的提醒最后再看 hook 库版本是否更新。做微信机器人就是这样代码只占三分剩下七分在运维和边界控制。你可以跑通自动回复也可以把防撤回和留言统计做得很深但能不能稳定跑一个月取决于前面避坑指南里的每一处细节。这些是我用一个又一个翻车教训换来的希望帮到你。本文还有配套的精品资源点击获取