简介面向希望基于 Web API 搭建微信自动化工具的开发者与爱好者这份开源项目压缩包提供一套可高度自定义的微信机器人实现思路重点覆盖自动回复、消息转发、防撤回与留言统计等常见需求适合用于个人助手、社群管理与消息留存等场景也可作为二次开发底座。资源包体积约71KB文件总数与主要文件类型暂未提供解压后需结合项目说明查看目录结构与依赖下载页显示已有1109人学习说明该方向具备一定关注度。其价值在于把多个高频功能收拢到同一套机器人框架中自动回复可用于关键词应答消息转发便于多群同步防撤回与留言统计则利于信息归档和运营分析借助 Web API 思路读者还能围绕权限、触发规则和业务逻辑做扩展降低从零搭建的门槛。1. 一套可自定义的微信机器人源码包先看边界再谈自动回复有人把“自动回复”理解成关键词匹配真跑起来才发现登录态、消息去重、转发重试、撤回事件顺序任何一个环节没兜住都会让机器人变成噪音制造机。这个压缩包给的是高度自定义的微信机器人源码资源核心能力包括自动回复、消息转发、撤回前本地留档、留言统计属于可二次开发的开源项目类型。适合做社群运营工具、内部通知桥接、客服分流原型的开发者不适合拿来做批量骚扰或采集无关人员信息。先把它当“消息总线 插件框架”看后面每一步才好落地。2. 解压后先跑最小闭环配置、登录态与消息入口压缩包最容易踩的坑是解压后直接双击启动结果配置文件没复制、数据目录没权限、登录态又混在默认配置里。我的习惯是先把它拆成“配置层、入口层、插件层”三段只跑通一条测试消息再开自动回复。这个资源包本身功能不少但最小闭环只需要三件事进程能起来、消息能归一化、动作能记录但不真正发送。2.1 目录分层与启动链路这类源码包通常不会把业务全塞进一个文件而是按app/、plugins/、config/、data/、tests/分层。目录名可能因包而异判断方法很简单找带main、bootstrap、gateway字样的入口文件再看谁负责读配置、谁负责收消息、谁负责发动作。先画清这条链路后面加规则和插件才不会乱。路径常见内容先确认什么app/启动入口、消息网关、路由、动作分发入口参数是否支持--config和--dry-runplugins/自动回复、转发、统计等扩展插件注册是显式列表还是目录扫描config/示例配置、规则文件、日志配置有没有config.example.yaml可复制data/本地缓存、留档库、发送队列当前用户是否有写权限tests/fixtures/回放样本、模拟消息有没有可直接重放的 JSONLscripts/初始化、迁移、压测脚本是否会改动真实数据目录先看入口文件里的参数解析不要急着改业务代码。很多“机器人不回消息”的问题根因不是规则写错而是配置路径读的是默认文件改的却是另一份。把config.local.yaml独立出来后面调试、回放、上线都能复用同一套代码。2.2 用 dry-run 把配置和登录态分离第一次启动不要直接连真实会话。先建独立目录、独立虚拟环境把依赖装干净再用--dry-run跑一遍。这个参数的作用是消息照常进入管线规则照常命中动作只写日志和本地文件不真正发出去。对自动回复和消息转发类项目来说这是最省后悔药的一步。# 1) 解到独立工作目录避开中文路径和带空格路径 mkdir -p ~/work/bot-resource unzip -q ./bot-resource.zip -d ~/work/bot-resource cd ~/work/bot-resource # 2) 建虚拟环境依赖隔离避免污染系统 Python python3 -m venv .venv source .venv/bin/activate python -m pip install -U pip pip install -r requirements.txt # 3) 复制本地配置先跑 dry-run不真正发消息 cp config.example.yaml config.local.yaml python -m app.main --config config.local.yaml --dry-run逻辑说明第一步解压到工作目录是为了让data/、logs/这类相对路径落在可控位置第二步用虚拟环境隔离依赖避免不同项目互相打架第三步的--config指定本地配置--dry-run只做管线演练。参数上重点看三个app.data_dir是否指向当前目录、app.log_level是否至少INFO、gateway.mode是否明确是本地缓存还是外部回调。首次启动若进入登录/绑定流程只在自有测试账号上完成不要用来路不明的会话缓存文件。Windows 下激活命令换成.venv\Scripts\activate路径里尽量不要带空格。启动后先看日志有没有“gateway started”“plugin loaded”“dry-run action”三类记录缺哪类就查哪段。没有消息进来时不要先怀疑规则先确认网关是否真的连上、登录态是否还有效。2.3 消息入口与插件注册一条文本怎么走到回复最小闭环的核心不是“能回复”而是“知道一条消息经过了哪些手”。常见做法是网关收消息后统一成内部结构做去重和自身消息过滤再交给路由匹配规则最后把动作交给分发器。插件只挂在钩子上不直接碰底层连接。下面这段启动代码是这类资源包常见的组织方式模块名可以按你手里的包替换。# app/bootstrap.py from app.config import load_config from app.gateway import Gateway from app.router import RuleRouter from app.dispatcher import Dispatcher from plugins.room_stats import RoomStatsPlugin def build_app(config_path: str): cfg load_config(config_path) gateway Gateway(cfg[gateway]) router RuleRouter.from_file(cfg[reply][rules_file]) dispatcher Dispatcher(cfg[forward], dry_runcfg[app][dry_run]) stats RoomStatsPlugin(cfg[stats], storegateway.store) # 顺序很关键先统计留档再走回复路由 gateway.on_message(stats.on_message) gateway.on_message(router.handle) gateway.on_recall(stats.on_recall) return gateway, dispatcher, stats这段代码里Gateway负责连接和消息归一化RuleRouter只做匹配Dispatcher负责发送和重试RoomStatsPlugin挂消息与撤回两个事件。dry_run必须一路透传到分发器否则路由层以为在演练发送层却真发出去了。on_message的注册顺序按业务定统计插件要先看到原始消息回复路由后看到也没问题但不要让多个插件各自直接发送否则同一条消息很容易触发多次回复。3. 自动回复与消息转发匹配优先级、幂等键和转发队列自动回复写起来快难的是“只回一次、回得对、不把机器人自己的消息再喂回规则”。消息转发也一样单条测试都正常一进群就出现重复转发、目标房间延迟、失败重试堆满队列。这个资源包把自动回复和转发放在同一套消息管线上所以规则优先级和幂等键要一起设计不能各写各的。3.1 回复规则先定优先级再谈关键词规则文件建议显式写priority不要让匹配顺序依赖 YAML 里的书写顺序。常见优先级是精确匹配 正则匹配 包含关键词 兜底回复。每条规则再带cooldown_s同一个用户在冷却时间内只回一次全局再加max_replies_per_minute防止某个群突然刷屏把机器人打成复读机。# config/rules.local.yaml rules: - id: exact_price priority: 100 match: {type: exact, value: 价格} action: {type: reply, template: 当前价目表见群公告人工客服稍后回复。} cooldown_s: 10 - id: regex_phone priority: 80 match: {type: regex, pattern: ^1[3-9]\\d{9}$} action: {type: tag, value: lead} - id: keyword_help priority: 60 match: {type: contains, any: [报错, 不能用, 怎么设置]} action: {type: forward, target: support-room} - id: fallback priority: 0 match: {type: always} action: {type: reply, template: 收到稍后处理。} cooldown_s: 30priority越大越先匹配命中后是否继续往下走要看路由实现里的stop_on_match。我的建议是默认命中即停只有打标签这类无副作用的规则才允许多条叠加。cooldown_s按“房间 用户 规则 id”做键不要只按用户否则同一个用户在两个群里会被互相影响。pattern里的反斜杠在 YAML 双引号中要转义写成\\d否则加载时容易报正则错误。还有一个常被忽略的点机器人自己发出的消息必须过滤。内部消息结构里通常有from_self或sender_type字段路由入口先判断命中就直接返回。没有这个判断兜底回复会回复自己的回复轻则循环两次重则把账号打进风控。3.2 消息转发用幂等键挡住重复队列限速转发的第一原则不是“快”而是“同一条消息同一个目标只发一次”。消息重连、网关重放、插件重复注册都会让同一条消息进入队列两次。解决办法是构造幂等键优先用msg_id没有就退化为房间、发送者、时间戳和文本摘要的组合。键里再拼上目标房间保证同一消息转发到不同目标时互不影响。import hashlib import time from collections import deque def forward_key(msg): # msg_id 是首选缺失时用房间、发送者、时间和文本摘要兜底 base |.join([ str(msg.get(room_id, )), str(msg.get(msg_id) or ), str(msg.get(sender_id, )), str(msg.get(ts, )), str(msg.get(text, ))[:80], ]) return hashlib.sha256(base.encode(utf-8)).hexdigest() class ForwardQueue: def __init__(self, sender, max_per_minute20): self.sender sender self.max_per_minute max_per_minute self.q deque() self.seen {} self.sent_at deque() def push(self, msg, target): key forward_key(msg) | target now time.time() # 10 分钟内同一目标只转发一次 if key in self.seen and now - self.seen[key] 600: return False self.seen[key] now self.q.append((key, msg, target)) return True def drain(self): while self.q: now time.time() while self.sent_at and now - self.sent_at[0] 60: self.sent_at.popleft() if len(self.sent_at) self.max_per_minute: time.sleep(0.5) continue key, msg, target self.q.popleft() try: self.sender.send(target, msg) self.sent_at.append(time.time()) except Exception: # 失败重入队生产环境要加最大重试次数和死信队列 self.q.append((key, msg, target)) time.sleep(1)max_per_minute是目标维度的限速不是全局。一个目标房间慢不应该拖垮其他房间反过来全局队列要有上限超过上限直接落盘或丢弃低优先级消息不能无限涨。seen字典要定期清理否则跑几天就变成内存黑洞。失败重试必须加最大次数三次失败后写入死信文件人工看一眼比无限重试靠谱。3.3 撤回留档与留言统计事件顺序和统计口径“防撤回”本质上是撤回事件到达前本地已经有一份消息快照。实现上不要在撤回事件里再去远端拉原文那时候通常已经拉不到了。正确顺序是消息到达时先upsert到本地留档表撤回事件到达时按msg_id更新recalled_at和撤回人不删除原文。留言统计则从留档表按房间、按天聚合统计口径要固定时区和日切点。-- data/schema.sql create table if not exists message_archive ( msg_id text primary key, room_id text not null, sender_id text not null, msg_type text not null, content text, created_at integer not null, -- 消息时间毫秒 recalled_at integer, -- 撤回时间未撤回为 null recalled_by text ); create index if not exists idx_archive_room_day on message_archive(room_id, created_at); create table if not exists daily_stats ( room_id text not null, stat_date text not null, -- 按固定时区切出的日期 messages integer not null default 0, recalls integer not null default 0, active_users integer not null default 0, primary key (room_id, stat_date) );msg_id做主键是了幂等重复消息不会插两遍。created_at存消息时间不要用入库时间否则回放样本会把统计打到今天。daily_stats是预聚合结果方便看板查询每天跑一次从message_archive重算能发现实时聚合的偏差。统计时排除机器人自己发的消息排除非文本消息要单独记口径不要把图片、系统提示混进“留言数”。4. 自定义插件开发生命周期、配置注入和回放调试这个包说“高度自定义”真正要看的不是配置项多少而是插件接口是否够薄。插件一旦直接操作网关连接、直接发网络请求、直接改全局状态后面加第二个功能就会互相踩。我的做法是只暴露四个钩子初始化、收到消息、收到撤回、刷盘关闭插件通过注入的store读写数据通过返回动作让分发器去发送。4.1 插件接口别做重四个钩子够用一个合格的插件生命周期大致是__init__读配置、建本地状态on_message做轻量判断和记录on_recall更新撤回标记flush把内存聚合落盘close做最后清理。钩子里不要做长耗时网络调用尤其不要在on_message里同步请求外部接口否则一个慢请求就能把消息管线堵住。配置注入建议传一个只读字典或配置对象不要让插件自己去读config.local.yaml。这样回放时可以用测试配置覆盖不会误连真实目标。数据访问统一走store哪怕当前是 SQLite后面换 PostgreSQL 也不用改插件逻辑。插件返回值只表达“我想做什么”真正发送由分发器统一做限速、重试、日志都在一处。4.2 写一个留言统计插件按房间、按天、带撤回标记下面这个插件做三件事统计每个房间每天的消息数、活跃用户数、撤回数命中“留言/报名/咨询”关键词时写一条线索记录撤回时给原消息打标记。代码不依赖具体网关只依赖store的接口方便回放测试。# plugins/room_stats.py from collections import defaultdict from datetime import datetime, timedelta import re class RoomStatsPlugin: def __init__(self, config, store): self.config config self.store store self.keyword re.compile(config.get(keyword_pattern, r留言|报名|咨询)) self.bucket defaultdict(lambda: { messages: 0, recalls: 0, active_users: set() }) def on_message(self, msg): if msg.get(from_self): return if msg.get(type) ! text: return room msg.get(room_id) if not room: return day self._day_key(msg.get(ts)) key (room, day) self.bucket[key][messages] 1 self.bucket[key][active_users].add(msg.get(sender_id)) if self.keyword.search(msg.get(text, )): self.store.append_lead(room, day, msg) def on_recall(self, evt): room evt.get(room_id) if room: day self._day_key(evt.get(ts)) self.bucket[(room, day)][recalls] 1 self.store.mark_recalled(evt.get(msg_id), evt.get(operator_id)) def flush(self): for (room, day), value in self.bucket.items(): self.store.upsert_daily( room_idroom, stat_dateday, messagesvalue[messages], recallsvalue[recalls], active_userslen(value[active_users]), ) self.bucket.clear() def _day_key(self, ts_ms): dt datetime.fromtimestamp(ts_ms / 1000) cutoff int(self.config.get(daily_cutoff_hour, 4)) if dt.hour cutoff: dt - timedelta(days1) return dt.strftime(%Y-%m-%d)keyword_pattern是正则默认只匹配三类词按业务改成更窄的模式别把“咨询”这种高频词放得太宽。daily_cutoff_hour设成 4 点是为了让凌晨消息归到前一天和常见运营口径一致。active_users用集合去重刷盘时只取长度不把用户 id 落进日统计表。flush必须可重复执行upsert_daily要按主键覆盖不能累加否则重启一次统计就翻倍。4.3 用回放样本验证插件不靠线上群试错插件写完先在本地回放不要直接扔进真实群。准备一份 JSONL每行一条消息包含type、room_id、sender_id、msg_id、ts、text这些字段再准备一条撤回事件验证recalled_at是否更新、日统计是否只加一次。回放命令通常带--speed控制速度带--assert做结果断言。# tests/fixtures/messages.jsonl 里放正常消息、重复消息、撤回事件 python -m app.replay \ --config config.local.yaml \ --sample tests/fixtures/messages.jsonl \ --speed 20 \ --assert-replies 2 \ --assert-forwards 1 \ --assert-daily-messages 3--speed 20表示按 20 倍速回放避免测试等太久--assert-replies断言自动回复次数--assert-forwards断言转发次数--assert-daily-messages断言日统计消息数。回放前把data/指到临时目录别把测试数据写进正式库。回放三遍第一遍正常消息第二遍重复消息第三遍撤回和乱序事件。三遍都过再考虑连授权测试群。5. 避坑与排查登录态、重复回复、转发堆积、撤回缺失和统计偏差这类项目在本地跑通只是开始真正耗时间的是上线后的“半正常”状态进程活着但没消息、回复发了两遍、转发延迟越来越高、撤回留档偶尔缺一条、统计数和手工数对不上。下面五条是我见过最容易复发的坑每条按“现象、原因、解决”拆开照着日志查基本能定位。5.1 登录态失效进程还活着消息不进来现象日志有定时心跳但半天没有一条消息进入管线重启后又正常一阵。原因通常是登录态过期、会话缓存被替换或者网关连接断了但重连逻辑只重连了空壳没有重新绑定消息回调。解决加一个“最后消息时间”健康检查超过阈值就告警启动日志打印登录态来源和过期时间重连后必须重新注册回调不能只重建连接对象。不要复用来路不明的缓存文件测试账号和正式账号的登录态分开存。5.2 同一条消息触发多次回复现象用户发一句“价格”群里出现两条甚至五条相同回复。原因常见三种多个插件都注册了on_message并各自发送网关重连后重放了最近消息没有做消息去重规则命中和兜底各回一次。解决发送动作统一收口到分发器入口用msg_id做去重TTL 至少覆盖重连窗口路由设置stop_on_match命中高优先级规则后不再走兜底。日志里把msg_id、规则 id、发送目标打全重复时一眼能看出是谁发的。5.3 转发队列堆积与重试风暴现象目标房间的消息延迟从几秒涨到几分钟日志里同一key反复出现。原因通常是没有限速、失败立即重试、同步发送阻塞队列或者目标房间不可用时没有熔断。解决队列设上限按目标做令牌桶限速失败重试用指数退避三次失败进死信文件连续失败达到阈值就暂停该目标等人工恢复。转发线程和消息接收线程分开接收端只入队不等待发送结果。5.4 撤回留档缺失或顺序错乱现象撤回事件到了但本地查不到原文或者原文有撤回标记没打上。原因是消息入库是异步的撤回事件先到或者msg_id在归一化时被改写两边对不上。解决消息到达时先做轻量写前日志再异步落库msg_id在入口层统一生成并透传不允许插件二次改写撤回更新用upsert查不到就写一条“孤立撤回事件”供排查。回放样本里故意把撤回事件放在消息事件前面能提前暴露顺序问题。5.5 留言统计对不上现象看板显示 100 条留言手工数出来 87 条。原因可能是时区不一致、日切点不同、重复消息被算两次、机器人自己的消息混进来或者实时聚合和离线重算用了两套口径。解决固定UTC8和日切小时msg_id唯一约束挡住重复统计 SQL 明确排除from_self和非文本类型每天从留档表重算一次和实时结果对账。对不上时先看“消息总数”和“去重后消息数”的差值再查时区不要一上来就改业务规则。6. 上线前的回放验证三张表把自动回复、转发、统计压一遍我现在给这类机器人做上线前检查只看三张表回复表、转发表、统计表。回复表验证“该回的回一次不该回的不回”转发表验证“同一条消息同一目标只发一次失败有死信”统计表验证“重复消息不翻倍撤回标记不丢日切口径一致”。表格不用复杂能覆盖边界就行。验证项输入期望精确回复文本“价格”只回一条冷却期内不再回自身消息from_selftrue的文本不触发回复和统计重复转发同一msg_id连推两次目标房间只收到一次撤回留档先消息后撤回原文还在recalled_at有值日切统计凌晨 3 点和 5 点各一条3 点归前一天5 点归当天回放命令跑三遍分别用正常样本、重复样本、乱序撤回样本断言不过就不连真实群。我吃过一次亏本地只测了正常消息上线后网关重连重放自动回复连发四条群里直接炸锅。从那以后我每次改规则或插件都强制走一遍“正常、重复、撤回”三组回放再去看死信文件和日统计对账。希望帮到你。本文还有配套的精品资源点击获取