不用天天盯着手机！个人微信二次开发实现微信群关键词自动监测

之前聊过 Trie + 上下文窗口的关键词监测方案，但那个方案对小团队单账号来说偏重——Trie 要内存、上下文判断要规则、多线程要 CPU。这篇给一个轻量方案：不用 Trie，简单前缀匹配 + LRU 消息去重 + 单线程单轮询，资源开销比 Trie 方案低 80%。

一、简单前缀匹配：一个字一个字扫

不用建树，直接遍历敏感词列表，对每条群消息做 in 判断——关键词 in 消息内容。对 95% 的场景够用（明确的敏感词：广告、赌博、政治）。上下文判断这一步在轻量方案里省略，靠运营在后台做误报清理迭代词库。

词库维护在本地 JSON 文件里，支持热加载（每隔 30 秒读一次文件，有变更就重新加载）。

二、LRU 消息去重：避免回调重推的重复告警

平台回调会因为网络原因重复推送同一条消息——如果不做去重，同一条敏感消息会触发好几次告警。用 LRU（最近最少使用）缓存存最近 1000 个 msg_id，进来先查缓存、命中就跳过，没命中才处理并存缓存。

LRU 缓存用 OrderedDict 实现，1000 条内存不到 1MB，单线程无锁，开销极低。

三、单线程单轮询：每秒 1 次够用

小团队单账号群消息量一般在每秒 0-5 条，单线程每秒轮询 1 次足够覆盖，延迟可接受（最多 1 秒）。轮询拿到新消息列表后逐条处理，处理完进入下一轮。

单线程无锁、异步写告警日志，整个进程 CPU 占用稳定在 5% 以内，跑在低配 VPS 或树莓派上都没问题。

轻量方案对照

组件	做什么	资源开销	适用场景
前缀匹配	词库 in 消息	O(W×N)	词库 < 50 个词
LRU 去重	msg_id 查缓存	内存 < 1MB	避免重复告警
单线程轮询	1 秒 1 次	CPU < 5%	单账号小群量

轻量关键词监测实现

import time, threading, json, os from collections import OrderedDict class LightweightMonitor: LRU_SIZE = 1000 # 去重缓存容量 POLL_INTERVAL = 1 # 轮询间隔 1 秒 HOT_RELOAD_SEC = 30 # 词库热加载间隔 def __init__(self, api, store, word_file): self.api = api self.store = store self.word_file = word_file self.words = self._load_words() self.lru = OrderedDict() self.last_msg_id = store.last_msg_id() threading.Thread(target=self.run, daemon=True).start() threading.Thread(target=self.hot_reload, daemon=True).start() def _load_words(self): if not os.path.exists(self.word_file): return [] with open(self.word_file, encoding="utf-8") as f: data = json.load(f) return data.get("sensitive", []) def hot_reload(self): while True: time.sleep(self.HOT_RELOAD_SEC) new_words = self._load_words() if new_words != self.words: self.words = new_words self.store.log("words_reloaded", len(self.words)) def run(self): while True: events = self.api.get("/group/message/poll", params={ "since": self.last_msg_id}).json()["items"] for ev in events: self.last_msg_id = ev["msg_id"] # 1. LRU 去重 if ev["msg_id"] in self.lru: continue self._lru_add(ev["msg_id"]) # 2. 前缀匹配 hit = self._scan(ev["content"]) if hit: self._alert(ev, hit) self.api.ack(self.last_msg_id) time.sleep(self.POLL_INTERVAL) def _lru_add(self, msg_id): if msg_id in self.lru: self.lru.move_to_end(msg_id) self.lru[msg_id] = True if len(self.lru) > self.LRU_SIZE: self.lru.popitem(last=False) def _scan(self, content): for word in self.words: if word in content: return word return None def _alert(self, ev, hit): # 异步告警：写日志 + 推给管理员 threading.Thread(target=self._do_alert, args=(ev, hit), daemon=True).start() def _do_alert(self, ev, hit): self.store.log_scan(ev, hit) owner = self.store.group_owner(ev["group_id"]) self.api.post("/send", { "receiver": owner, "content": (f"群 {ev['group_id']} 命中敏感词「{hit}」" f"发送人 {ev['from']}" f"内容 {ev['content'][:100]}")[:500]}).json()

落地建议

群关键词监测不一定要上 Trie + 上下文窗口那套重型方案——小团队单账号用简单前缀匹配 + LRU 去重 + 单线程轮询就够了，整个进程 CPU 占用 5% 以内、内存不到 1MB。LRU 去重避免回调重推的重复告警，词库热加载支持运营随时更新不用重部署。重词库（≥50 个）或多账号大群量再升级到 Trie 方案。