简介这是一份基于 Web API 的轻量微信机器人开源项目面向需要自动化运营社群或对微信协议开发感兴趣的开发者。项目将自动回复、消息转发、防撤回、留言统计等常见需求模块化整体 zip 包仅 71KB小巧易分发适合直接部署与二次改造。已有 1107 人学习下载具备一定实践参考价值。借助该项目可了解微信机器人事件监听与消息处理的基本链路掌握规则触发、消息转发、撤回拦截等功能的落地写法留言统计模块也能帮助梳理数据采集与汇总思路为后续扩展群管理、智能问答等能力提供基础。对想快速搭建个人微信助手或深入研究 Web 接口调用的开发者来说是一份紧凑而实用的开源参考。1. 自建微信机器人从自动回复到防撤回这份资源能让你少走三个月弯路做过消息自动化的人基本都经历过这样的阶段先搜到一个号称高度自定义的机器人项目下载下来发现要么是编译好的黑匣子要么依赖一个早已失效的网页接口改两行配置就报错最后只能躺在收藏夹吃灰。这份资源不太一样它不是成品工具而是一套以PYPluginLoader为核心、支持热插拔消息处理模块的机器人工程骨架覆盖自动回复、群消息转发、撤回拦截、留言统计四条主线每个功能都能从config.yaml或plugins/目录里找到对应位置去改。适合两类人一是想给社群或工作群做值班机器人的运营者二是想理解消息中间件怎么写、规则引擎怎么挂的开发者。我会顺着「这东西怎么跑起来 → 每个模块的改法 → 部署后你一定会踩的坑 → 安全的边界」这条线用一组可复现的最小配置把它拆完。2. 运行架构与消息流水线先搞清消息从哪来、到哪去2.1 不是网页接口也不是hook注入它是事件驱动中间件很多人对微信机器人的第一反应是找协议库去模拟手机端登录但这类方案最大的问题是资源里写的依赖版本容易随协议升级而失配今天能跑明天就掉线。这份资源采用的是另一条路线以事件驱动中间件的方式把消息源、插件路由、动作执行三层拆开。消息源负责监听一个目录下的消息实体文件例如inbox/*.json插件路由按rules.yaml里的匹配规则把消息分发给对应模块动作执行只负责回调结果。这种设计的直接好处是你不需要理解底层协议细节所有消息格式统一为 JSON 结构。资源里自带了一个simulator.py用于在无微信环境时投递测试消息我第一次跑通就是靠它而不是拿自己的账号去做真实收发。换句话说这份资源给你的是一个可控的消息处理闭环而不是一个直接连着真实账号的冒险器。2.2 目录结构速览六个目录分别装什么下载并解压后第一件事不是看 README而是先把目录结构读明白。资源的核心布局如下wechat-automation/ ├── main.py # 入口负责加载配置与启动 IPC 服务 ├── config.yaml # 全局配置监听路径、日志级别、白名单 ├── rules.yaml # 规则引擎关键词匹配与消息路由表 ├── inbox/ # 消息落地目录每一条消息一个 JSON 文件 ├── archive/ # 处理后消息的归档区 ├── plugins/ # 插件目录每个功能模块一个子目录 │ ├── auto_reply/ # 自动回复模块 │ ├── forwarder/ # 消息转发模块 │ ├── anti_recall/ # 防撤回记录模块 │ └── msg_counter/ # 留言统计模块 └── simulator.py # 本地模拟投递工具用于离线测试代码逻辑说明main.py是唯一需要常驻的进程它做的事是轮询inbox/目录读取文件、走规则引擎、调用插件、最后写archive/。这样的好处是任何插件崩溃都不会拖垮主进程顶多就是那一条消息进了archive的error/子目录。参数说明config.yaml里的poll_interval默认是 1.0 秒意思是最多 1 秒延迟。想提高响应速度可以调到0.3但 CPU 占用会同步上升低配小主机建议保守一点。这个架构给了我一个启发不管你对真实账号对接抱多大期望先把这套本地事件循环跑通后续就算要接任何通道也只是多写一个消息源适配器的问题。2.3 初始化环境到首次运行四个命令环境要求只有 Python 3.9 以上第三方依赖只pyyaml一个这在同类资源里算轻量。安装与启动的完整过程如下python -m venv botenv source botenv/bin/activate # Windows 下改为 botenv\Scripts\activate pip install pyyaml python main.py --init # 生成默认配置目录 python simulator.py --msg 你好 # 投递一条测试消息逻辑说明--init会在当前目录生成可用的config.yaml与空目录结构省去手工建目录的环节。simulator.py --msg则是向inbox/写入一条 JSON 消息等同于模拟一个用户发来文本你好。参数说明--init还支持--force参数用于重置已被改动过的配置文件。我一般建议首次跑通前不要碰源码先把模拟消息走通看到archive/里有对应输出文件再开始改规则。这一步能过滤掉大量环境问题因为后续修改配置时如果出错你能快速判断是环境问题还是规则写法问题而不是两头找原因。3. 四个核心模块的落地改造自动回复、消息转发、防撤回、留言统计3.1 自动回复从固定话术到带状态的关键词引擎自动回复模块的默认实现是纯关键词精确匹配配置都在rules.yaml的auto_reply段auto_reply: - name: greeting keywords: [你好, hello, hi] reply: 你好我是值班助手。留言请直接发送我会统计并定时汇报。 - name: help keywords: [帮助, help, 怎么用] reply: 发送任意消息即视为留言。回复关键词「统计」可查看今日留言数。状态回复是指同一会话内根据上一次关键词决定下一次回复。在plugins/auto_reply/handler.py中核心逻辑长这样class ReplyHandler: def __init__(self, rules: dict): self.rules rules self.session {} def handle(self, msg: dict) - str: content msg.get(content, ) uid msg.get(from_uid, unknown) # 命中状态等待词返回下一次预期的关键词清单 if uid in self.session: expected self.session.pop(uid) reply f请回复以下选项{expected} return reply # 普通关键词匹配遍历规则表 for rule in self.rules: for k in rule.get(keywords, []): if k in content: # 命中后设定下一次回复的期待分支 if rule.get(expect_next): self.session[uid] / .join(rule[expect_next]) return rule[reply] return self.rules[-1][reply]逻辑说明handle()每次被调用时先检查会话表self.session如果该用户处于上次回复的等待分支中就直接返回候选词否则走关键词遍历。这里用from_uid区分不同用户会话数据只放在内存里重启即清除。这样设计能满足值班机器人最常见的分支对话场景比如先问“需要查天气还是查快递”下一轮再按用户回答分发。参数说明rules[-1][reply]是兜底回复也就是无命中时返回的内容。规则表越靠前优先级越高因此要把高频问候词放在前面。如果要把会话改为持久化存储只要把self.session的 get/set 操作替换为shelve或sqlite3即可接口不变。有个细节需要注意匹配用的in是子串匹配消息里包含“你好”就会触发想精确全等可以改成content k。3.2 消息转发正则路由与群聊白名单配合使用消息转发模块解决的是多群消息汇聚或跨群通知问题。默认规则是从指定群聊将含特定关键词的消息转发到目标群。以rules.yaml的forwarder段为例forwarder: - name: project_sync from: group_project to: group_report match: 紧急|故障|报错 # 正则表达式 - name: daily_standup from: group_all to: group_archive match: 日报这些group_xxx在初始化时会映射到真实标识。plugins/forwarder/handler.py中的路由逻辑如下import re class Forwarder: def __init__(self, rules: list): self.rules rules self.transport None def set_transport(self, transport): 注入消息发送回调默认只写入 outbound/ 目录 self.transport transport def handle(self, msg: dict) - bool: group msg.get(group_name, ) content msg.get(content, ) for rule in self.rules: if group rule[from] and re.search(rule[match], content): payload { type: forward, to: rule[to], content: content, } if self.transport: self.transport.send(payload) else: print(f[模拟转发] {rule[to]}: {content}) return True return False逻辑说明这是典型的管道-过滤器模式。transport是一个可注入的发送器资源默认实现是打印日志方便验证路由是否命中。真实对接时只需把print替换为实际的外部调用模块本身不需要改。这个设计让我比较放心——测试和真实运行之间只差一层 transport 实现。参数说明re.search用的是正则搜索而非字符串匹配因此规则里的match写成紧急|故障就能覆盖多个词。注意特殊字符转义比如想匹配版本号v1.0需要写成v1\.0。如果同一消息同时命中两条转发规则当前实现会都转发想只走第一条可以在命中后break出去。3.3 防撤回设计采集与展示分离不做逆向这是最容易让人误解的模块。资源里的防撤回不是拦截消息撤回的网络包而是记录已经发生的事件以便在消息撤回前把内容存下来当作留痕。做法是插件定时扫描inbox/中的事件消息当检测到recall类型时从archive/中找回原始内容追加一条撤回记录import os import json import glob class RecallRecorder: def __init__(self, archive_dir: str): self.archive_dir archive_dir def record_recall(self, event: dict) - str: origin_id event.get(origin_msg_id, ) # 到已归档目录里找回原始消息文件 target os.path.join(self.archive_dir, f{origin_id}.json) if not os.path.exists(target): return 未找到原始消息可能消息过期或归档被清理 with open(target, r, encodingutf-8) as fp: origin json.load(fp) record { recalled_uid: event.get(from_uid, ), content: origin.get(content, ), timestamp: event.get(ts, 0), } out_path os.path.join(self.archive_dir, recall_records.json) with open(out_path, a, encodingutf-8) as fp: fp.write(json.dumps(record, ensure_asciiFalse) \n) return f已记录撤回消息{origin.get(content, )[:20]}逻辑说明整个模块建立在一个前提之上——消息先落地入archive/随后才会收到撤回事件。因此防撤回的本质是「比撤回快一步归档」。record_recall()用origin_msg_id找到归档文件把原始内容写入追加日志。这样即使消息被撤回留痕记录依然存在。参数说明append模式写recall_records.json是逐行 JSON方便后续用jq或 Python 做统计。如果怕文件过大可以改成每个小时一个文件或者只记录文本长度超过 10 的消息。这里有一个容易忽略的点如果源消息在归档前就发生了撤回这条记录就会显示「未找到原始消息」。因此要让该模块可靠main.py的轮询间隔必须小于消息撤回的窗口期建议poll_interval不超过 0.5 秒同时归档逻辑要优先处理消息落地而不是集中到某一时刻批量写。3.4 留言统计用 SQLite 聚合替代内存计数留言统计模块的目标是回答「今天谁留了言、留了多少条」。默认实现将每条消息解析出的from_uid、group_name、ts写入 SQLite 数据库这样重启不清零查询也更灵活。建表与写入的核心函数如下import sqlite3 import time DB_PATH data/msg_stats.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS msg_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, uid TEXT NOT NULL, group_name TEXT NOT NULL, ts INTEGER NOT NULL, content TEXT ) ) conn.commit() conn.close() def record_message(msg: dict): conn sqlite3.connect(DB_PATH) conn.execute( INSERT INTO msg_log (uid, group_name, ts, content) VALUES (?, ?, ?, ?), (msg.get(from_uid), msg.get(group_name), int(time.time()), msg.get(content, )[:500]), ) conn.commit() conn.close() def query_today_count(group_name: str) - int: day_start int(time.time()) - 86400 conn sqlite3.connect(DB_PATH) cur conn.execute( SELECT COUNT(*) FROM msg_log WHERE group_name? AND ts?, (group_name, day_start), ) row cur.fetchone() conn.close() return row[0] if row else 0逻辑说明init_db()只负责建表幂等可重复调用。record_message()在每一条有效消息进入主流程时被调用把关键字段落库。query_today_count()按群组统计 24 小时内的留言条数。这里我特意在 content 字段上做了 500 字截断避免因长消息导致库体快速膨胀。参数说明表结构里ts存的是 Unix 时间戳秒统计窗口是86400秒。想按自然日统计可以用datetime计算当天零点的 timestamp但这会涉及时区问题。部署环境若跨时区建议直接用 UTC 零点做切分避免定时统计出现 8 小时偏差。如果你想扩展成 Top 用户排行把COUNT(*)换成GROUP BY uid ORDER BY COUNT(*) DESC LIMIT 5即可表结构完全不用改。4. 调试与模拟把 message flow 可视化先于真实环境暴露问题4.1 消息全链路追踪从 sink 到 archive 的每个环节打点机器人这类消息系统最棘手的问题是故障难以复现消息为什么没有转发为什么没触发回复在真实聊天环境里往往只能靠用户反馈去猜。这份资源提供了一套比较实用的办法——在config.yaml里开启debug_flow: true主进程就会在执行链路的每个环节写一行结构化日志debug_flow: true log_level: DEBUG开启后main.py的调试输出大致如下2025-01-10 10:00:01 [DEBUG] sink: inbox/20250110_100001.json 2025-01-10 10:00:01 [DEBUG] route: matched - auto_reply[greeting] 2025-01-10 10:00:01 [DEBUG] handler: auto_reply - reply generated 2025-01-10 10:00:01 [DEBUG] archive: moved to archive/20250110_100001.json这段日志的价值在于它能让你看到消息从进入到归档的完整生命周期。如果消息没有日志输出那就是消息源没接进来如果只有sink没有route那问题在规则匹配如果有route没有handler那问题在插件内部抛异常了。实际排查时我会把上面四行日志当作定位用的分水岭先确定断点在哪一段再进对应模块查代码。日志级别的设定可以这样理解DEBUG输出每一条消息的完整流转适合单条消息排错但在高并发场景下会产生大量 IO不建议长时间开启INFO级别则只记录插件主动打出的关键动作比如「转发成功」「记录撤回一条」对日常监控更友好。4.2 模拟器进阶批量消息与撤回事件注入simulator.py除了支持--msg单条模拟还提供了批量模式。这个模式在测试防撤回模块时很有用先批量注入多条消息再注入撤回事件看记录是否完整。用法示例python simulator.py --batch tests/case_fixture.json python simulator.py --recall --origin_msg_id 20250110_100003逻辑说明--batch从 JSON 数组文件读取多条消息并依次写入 inbox--recall指定撤回事件--origin_msg_id对应之前注入的某条消息编号。这个组合能把「先收消息、再被撤回」的完整链路在本地轻松走一遍。参数说明批量投递时每条消息的ts建议按顺序递增否则防撤回记录的时间顺序会乱统计模块算出来的时序也会不对。如果要模拟多条规则同时命中的极端场景可以在批量文件里放同一条内容然后观察转发模块是否会出现重复投递。5. 避坑与常见问题接手这份资源最容易翻车的五个点5.1 撤回事件消息与原始消息不在同一个文件轮次现象防撤回模块报了「未找到原始消息」但archive/里确实有该文件。原因主进程轮询消息时撤回事件消息先于原始消息被读取并进入record_recall()处理此时原始消息尚未归档 。解决在main.py的轮询循环里对消息按时间戳做一次排序确保消息先落库、再处理撤回事件。5.2 规则命中顺序不可控导致回复优先级错乱现象消息同时包含「你好」和「帮助」系统回复的是帮助内容而非问候语。原因auto_reply规则表顺序固定为从上到下资源默认把「帮助」规则放在了greeting前面。解决在rules.yaml中把优先级高的规则放在列表前面或给规则加priority字段在handle()里按该字段倒序排序后遍历。5.3 模拟器能跑通但真实消息一直不进系统现象simulator.py投递的消息能正常处理但真实环境的发送方报告消息没有反应。原因多数情况是 inbox 监听的是本地路径而真实消息投递方是一个独立进程或远端服务它没有权限或没有正确配置写入到同一目录。解决先检查监听目录是否有新的.json文件落下如果没有说明问题出在消息源适配器与资源核心逻辑无关如果有文件落下但没反应再排查config.yaml文件路径是否配置为绝对路径。5.4 防撤回模块长期运行后记录文件无限膨胀现象recall_records.json在运行一周后已到几百 MB打开都困难。原因代码里用的是追加写入模式没有做任何轮转。解决在record_recall()中按天切换文件名例如recall_records_20250110.jsonl同时加一个按周清理的钩子或运维定时任务去处理旧文件。注意只清理已归档的记录不要动archive/下的原消息文件。5.5 修改路由规则后不重启就不生效线上排查时容易误判现象改了rules.yaml的相关匹配表达式但测试消息的行为没变化。原因资源默认是启动时一次性加载规则运行期间不会热更新。解决在main.py的轮询循环中比对配置文件修改时间或在调试阶段直接重启进程。考虑到设计简洁性我建议重启进程而不是在代码里引入文件监听否则在高频写入机器人的场景下会造成额外的 INotify 开销。提示遇到任何一条问题时先确认自己的版本文件是否完整重点看plugins/目录下是否缺少__init__.py再决定是改代码还是改配置。很多报错都源于解压时丢了空目录文件。6. 工程化收尾把短生命周期脚本变成长期稳定运行的伴生服务6.1 用脚本化自检代替「启动后看日志」初次接触这个项目时我有一个习惯每跑通一个功能就在终端用完整命令串验证一次。等所有功能都验证完我把这些命令整理成一个smoke_test.sh每次改完规则或插件后先执行一遍合格后再真正部署# 1. 环境检查 python --version | grep 3.9 echo python ok pip show pyyaml /dev/null 21 echo pyyaml ok # 2. 配置语法检查 python -c import yaml; yaml.safe_load(open(config.yaml)); print(config ok) # 3. 链路检查自动回复 转发 留言计数 python simulator.py --msg 你好 | grep 你好 python simulator.py --msg 紧急 系统故障 | grep 模拟转发 python -c from plugins.msg_counter import handler; print(handler.query_today_count(group_all)) # 4. 模拟撤回 python simulator.py --batch tests/recall_fixture.json python simulator.py --recall --origin_msg_id tests_recall_msg_1 grep 测试消息 data/recall_records.json这套自检脚本的妙处在于它同时验证了「代码能跑」和「规则符合预期」两件事。特别是第四步如果不模拟撤回防撤回模块在真实环境出问题时你根本没法手动构造一个撤回事件来复现。6.2 消息洪峰下的限流策略之前做某跨平台系统的接入时遇到过半夜被大量重复留言打爆的情况。这个项目本身没有内置限流所以我自己在main.py的 route 环节加了一个简单的滑动窗口计数器核心逻辑是按用户限流from collections import defaultdict import time class Throttle: def __init__(self, window: int 5, max_count: int 10): self.hits defaultdict(list) self.window window self.max_count max_count def allow(self, uid: str) - bool: now time.time() q [t for t in self.hits[uid] if now - t self.window] self.hits[uid] q if len(q) self.max_count: return False self.hits[uid].append(now) return True逻辑说明allow()在同一用户 5 秒内超过 10 条消息时返回False主进程收到False会跳过该条消息不转发、不回复、不落统计库。window和max_count决定了峰值容忍度外部攻击行为比较明显时可以把window调短到 2 秒同时加大max_count到 20以放行正常活动但压制异常洪泛。参数说明这里的defaultdict(list)会随着用户增长累积旧时间戳虽然allow()内部会清除超窗记录但活跃用户非常多时建议加一个全局定期清理任务把超过window的用户 key 直接删掉避免内存持续攀升。6.3 日志与备份的最终去向项目运行时间长了archive/目录会积累大量消息文件。我的习惯是每周末做一次完整打包并压缩然后单独备份到指定目录data/msg_stats.db则用sqlite3 .backup命令做热备。以下是我实际在用的轮转脚本# 按天归档昨天之前的消息避免主目录无限增长 mkdir -p backups/$(date %Y%m%d) find archive/ -name *.json -mtime 1 -exec mv {} backups/$(date %Y%m%d)/ \; tar -czf backups/archive_$(date %Y%m%d).tar.gz backups/$(date %Y%m%d)/ rm -rf backups/$(date %Y%m%d)逻辑说明find -mtime 1只移动超过一天的文件避免误动当天正在处理的消息tar做完归档压缩后直接清掉未压缩的原始文件节省磁盘。对于留言统计使用的 SQLite 数据库直接复制文件可能造成数据不一致所以最好用 sqlite3 的 backup 命令而不是cp。6.4 关于边界与习惯坦白说刚接触这类工程时我也尝试过一个月内把每个模块都改一遍结果就是越改越乱最后不得不回滚重来。后来养成的习惯是每次只改一个变量、跑一次smoke_test.sh、看一段日志确认无副作用后再动下一个点。特别是在加限流或者清理策略时先观察结果再决定要不要保留。从那以后我每次部署前都要跑一遍自检脚本然后看一眼archive/error/目录有没有新增文件再决定是否推进。这个习惯帮我避免了很多次把「测试环境一切正常」误判为「生产环境也可以放心」的翻车。希望这篇文章里关于流程、规则字段和日志定位的经验能让你接手这套资源时少走几步冤枉路。如果它帮你节省了哪怕半天排查依赖或定位路由问题的时间也就值了。本文还有配套的精品资源点击获取