简介：这份资源是一套面向个人学习者的 TG 电报群关键词监听机器人源码，适合对即时通讯自动化监控、PHP 后端开发或社群运营感兴趣的入门与进阶用户。资源使用普号隐身监控思路，可对群组与频道消息进行关键词触发，并在命中后通过伪装账号实现人工响应，主账号隐蔽性较强。压缩包共 57 个文件，约 94KB，以 PHP 源码为主，辅以配置文件、JSON 关键词规则、Docker 部署配置、shell 脚本及教程文档，覆盖事件监听、日志记录、API 扩展、会话迁移等模块，整体结构清晰，便于按需学习和二次开发。已有 129 人浏览学习。内含关键源码与说明文档，能帮助读者快速理解关键词监听、事件分发、多账号伪装等实现路径，并参考 docker-compose 与环境配置快速搭建调试环境，适合用于学习隐私防护、消息监控及机器人开发相关知识。
1. TG电报群关键词监听：从“看不过来的群”到“秒级人工响应”
做社群运营和舆情监控的人，迟早会面对同一个尴尬局面：手里握着十几个甚至几十个 TG 电报群，既要盯竞品动态，又要抓用户吐槽，还要第一时间响应潜在客户。人工一个个群里翻，一天下来什么都干不了。这个标题里的“TG电报群关键词监听机器人源码”，本质就是一套用普号（普通用户账号，而非 Bot 账号）直接混在群里监听关键词、命中后把消息和上下文推给人工，再由人工一键回复的完整工程方案。它能解决三个具体问题：群多导致漏看、关键词靠眼搜、以及发现商机后响应太慢。适合已经有明确监听场景、想把“人肉盯群”换成半自动流水线，又不想把号交给第三方 SaaS 的从业者。我先说结论：这套做法比 Bot API 监听更灵活，但坑也多在账号安全上，后面会逐个拆开讲。
2. 先立住架构：为什么普号监听必须绕开 Bot API，以及源码的模块边界
2.1 Bot API 与 MTProto 两条路线：监听群消息到底差在哪
TG 生态里做机器人，大多数人的第一反应是走 Bot API，用 @BotFather 创建机器人拿 token，然后靠getUpdates或setWebhook收消息。但 Bot API 有一个天然的硬限制：机器人只能收到被/start或/invite加入的群里的消息，而且收不到普通用户之间的悄悄话、收不到频道里某些类型的历史内容，更做不到“像普通用户一样潜伏在群里”这件事。只要群里有人把机器人权限一改，或者群主设置了“仅管理员可添加机器人”，Bot 立刻变成瞎子。
普号监听走的是另一条路：用 MTProto 协议直连 Telegram 服务器，以你自己的普通账号身份登录，加入指定群聊后，通过事件系统实时收取这个账号能看见的所有消息。Telethon 是这条路最常用的 Python 实现，整套监听源码的核心就建立在这个库之上。用普号有几个 Bot 给不了的性价比：不需要申请机器人、不需要群主批准（你本来就在群里）、消息可见范围跟你的账号完全一致，群友的昵称、头像、回复链上下文都能拿到。代价是账号有风控风险，后面第 4 章专门讲怎么把风险压到最低。
2.2 源码模块拆解：监听器、规则引擎、响应台三个部分各管什么
这套源码不是单个脚本，而是三个模块的协作。监听器负责跟 Telegram 服务器保持长连接，把所有新消息按“群 ID + 频道 ID + 消息 ID + 发送者 ID + 文本”的结构接住；规则引擎负责判断这条消息值不值得关注，最基础的是关键词匹配，进阶一点就是正则、排除词、按群分组、按时间段生效；响应台是一个给人工操作员用的管理界面，常见实现是 FastAPI 后端加一个极简 Web 页面，监听器命中关键词后把消息写进任务池，操作员在页面上看到并决定“回复”或“忽略”，回复时以普号身份原路发回群里。
把这三个模块分开，而不是写在一个脚本里，原因很现实：监听长连接对稳定性要求高，规则调整对响应速度要求高，人工操作界面又完全是另一套交互逻辑。混在一起时，改一个关键词规则不小心触碰到连接对象，整个监听就断了。分开之后，监听器挂了重启它不影响规则库，规则引擎有 bug 只影响命中率，响应台宕机消息也还在队列里躺着，不会丢。
2.3 最小目录结构与启动顺序：拿到源码后先跑起来
一份能直接落地的源码，拿到手先看目录结构，心里要有数。常见做法是下面这样，我为这个标题补全一套最小可运行结构：
tg_keyword_monitor/ ├── config.yaml # 账号、API ID、监听群列表、规则文件路径 ├── requirements.txt # telethon / fastapi / uvicorn / pyyaml ├── monitor/ │ ├── listener.py # 监听器：Telethon 事件循环，产出原始消息 │ ├── rules.py # 规则引擎：关键词与正则的匹配、命中判定 │ └── queue.py # 任务队列：命中消息排队等待人工处理 ├── responder/ │ ├── server.py # 人工响应台：FastAPI 接口 │ └── sender.py # 回复发送器：节流发送，复用普号会话 └── run_monitor.sh # 一键启动监听器启动顺序要遵守“先规则、再响应、最后监听”的原则。原因是监听器一启动，消息就会像水一样灌进来，如果规则引擎没趁热加载、响应台还没起来接消息，命中消息会积压在队列里，重启后反而要补处理一堆旧任务。推荐先python -m responder.server单独启响应台，确认页面能打开，再python monitor/listener.py启动监听。首次登录 Telethon 需要在终端交互输入手机号和验证码，验证码经常延迟到，所以监听器脚本里要加session文件持久化，第二次启动就不用重新授权，这点在 5.1 节还会展开讲。
3. 关键词监听的核心实现：从消息流抓取到规则命中，再到事件入队
3.1 消息流抓取：用 Telethon 的 events.NewMessage 把群聊变成事件流
监听器是整个系统的水龙头，它要做的只有一件事：把指定的群聊消息变成一个接一个的 Python 事件。Telethon 的写法很标准：
from telethon import TelegramClient, events client = TelegramClient( session_file, # 普号登录凭证，持久化保存 api_id, # 从 my.telegram.org 申请的应用 ID api_hash, # 与 api_id 配对的应用密钥 ) TARGET_GROUPS = [-1001234567890, -1009876543210] # 要监听的群 ID @client.on(events.NewMessage(chats=TARGET_GROUPS)) async def handler(event): msg = event.message print(f"[{event.chat_id}] {msg.sender_id}: {msg.text}") async def main(): await client.start() print("监听中，等待消息...") await client.run_until_disconnected() if __name__ == "__main__": client.loop.run_until_complete(main())这段代码的关键点有两个。第一，chats=TARGET_GROUPS传的是群的实体 ID，负号开头表示超级群，这个值在监听时最好直接从event.chat_id中打印出来回填，而不是靠肉眼猜。第二，await client.run_until_disconnected()会让进程一直挂着长连接，Telethon 内部会自动处理消息差量同步，也就是 Telegram 服务器会把进程离线期间漏掉的消息补发过来，这是监听器不漏消息的底层保障。
3.2 关键词规则引擎：多关键词、非/与逻辑和正则匹配怎么组织
监听器拿到消息后，规则引擎负责回答“这条要不要推给人工”。规则别搞太复杂，一个 JSON 文件足够，常见做法是下面这种结构：
{ "groups": { "-1001234567890": { "keyword_rules": [ { "id": "rule_bug", "keywords": ["闪退", "崩溃", "bug"], "priority": 1 }, { "id": "rule_price", "keywords": ["价格", "报价", "多少钱"], "exclude": ["免费", "白嫖"], "priority": 2 } ], "regex_rules": [ { "id": "rule_email", "pattern": "[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}" } ] } } }匹配逻辑用一段简短代码实现：先做关键词层级的快速过滤，再做正则层级的精确捕获。文本先统一转小写，避免“BUG”和“bug”被当成两个词；命中后记录规则 ID、命中关键词和上下文，方便人工判断。优先级是给响应台做排序用的，竞品突然提你产品名，和普通用户随口说一句“价格”，紧急程度肯定不一样。规则文件建议放在能热加载的位置，响应台提供接口触发规则重新读取，这样不用重启监听进程。
3.3 命中事件写入任务队列：监听和人工响应之间的一层缓冲
直接让监听器调用响应台发送消息，是最省事但最脆弱的写法。监听线程一卡，响应全卡；响应台一重启，消息全丢。所以中间要加一个任务队列，把“发现”和“处理”切开：
import asyncio from dataclasses import dataclass, field from datetime import datetime @dataclass class HitTask: task_id: str group_id: int msg_id: int sender_id: int text: str matched_rule: str created_at: datetime = field(default_factory=datetime.utcnow) status: str = "pending" # pending / replied / ignored class TaskQueue: def __init__(self): self._tasks = asyncio.Queue() async def push(self, task: HitTask): await self._tasks.put(task) async def pop(self) -> HitTask: return await self._tasks.get() def add_to_db(self, task: HitTask): """同步写入 SQLite，防止进程退出丢任务""" # INSERT INTO tasks(...) VALUES(...) pass队列用asyncio.Queue就够了，但记得在push的时候同步写一条 SQLite 记录。这样即使响应台挂了，重启后从数据库捞未处理的任务补办，不至于客户在群里问完价格等半小时没人理。队列消费者是响应台的 WebSocket 推送逻辑，操作员看到的每一条“待回复”，都是从pop()拿出去的。
4. 实时人工响应系统：命中消息送到操作员面前，再以普号身份回复
4.1 人工审核台：监听命中后，操作员在页面上看到什么、操作什么
关键词命中只是“发现问题”，真正产生价值的动作是“人工响应”。响应台最朴素的形态是一个仅限内网访问的 Web 页面，用 FastAPI 提供两个接口：一个 WebSocket 接口把TaskQueue.pop()出来的新任务实时推送到浏览器，一个 POST 接口接收操作员的回复指令。整个链路用一段代码就能看清：
from fastapi import FastAPI, WebSocket from pydantic import BaseModel app = FastAPI() class ReplyRequest(BaseModel): task_id: str response_text: str @app.websocket("/ws/tasks") async def task_stream(ws: WebSocket): await ws.accept() while True: task = await queue.pop() await ws.send_json({ "task_id": task.task_id, "group_id": task.group_id, "sender_id": task.sender_id, "text": task.text, "matched_rule": task.matched_rule, }) @app.post("/api/reply") async def send_reply(req: ReplyRequest): task = load_task(req.task_id) if task.status == "replied": return {"ok": False, "reason": "already_replied"} await sender.send_to_group( group_id=task.group_id, text=req.response_text, reply_to=task.msg_id, ) task.status = "replied" return {"ok": True}这段接口设计有几个容易忽略的点。WebSocket 循环里queue.pop()是阻塞的，没有新任务时连接就这么挂着，浏览器端不需要做轮询，体验上消息是“秒弹”的，这就是实时性的来源。回复接口里必须校验task.status，避免操作员双击按钮导致同一句话发两次，群友会觉得这个号被入侵了。reply_to=task.msg_id可以让回复在群里以“引用那条消息”的形态出现，真人操作的感觉更强，比单独顶一条新消息好得多。
4.2 回复发送节流与失败重试：普号连发和秒回是风控重灾区
这段是血泪经验。人工操作员在页面上连续处理三五条消息，点击速度极快，如果不做发送节流，普号会在几秒内连发多条消息，直接把自己送进风控名单。sender 模块必须要做三件事：最小间隔、失败退避、次数限制。常见做法是信号量加最小间隔：
import asyncio class ThrottledSender: def __init__(self, min_interval=3.0): self._lock = asyncio.Lock() self._last_send = 0 self._min_interval = min_interval async def send(self, client, group_id, text, reply_to=None): async with self._lock: now = asyncio.get_event_loop().time() wait = self._min_interval - (now - self._last_send) if wait > 0: await asyncio.sleep(wait) try: await client.send_message(group_id, text, reply_to=reply_to) self._last_send = asyncio.get_event_loop().time() except FloodWaitError as e: await asyncio.sleep(e.seconds + 60) # 额外多等 60 秒min_interval=3.0是最保守的起步值，意思是同一账号两次发送之间至少隔 3 秒。如果你的号是养了很久的老号，可以降到 1.5 秒；新号建议拉到 5 秒以上。FloodWaitError是 Telegram 服务器的限流机制，它报的seconds是服务器要求你等待的时间，如果短时间频繁触发，说明间隔还得拉大。这里宁可让消息晚到 5 秒，也不要让账号进小黑屋。
4.3 隐身监控的“三层克制”：账号侧、行为侧、消息侧的参数怎么定
标题里的“隐身监控”，不是真的穿隐形衣，而是让这个普号在群里的行为特征尽可能贴近一个普通真人，不引起群主和服务器两边的注意。我一般按三个层面来定参数。
账号侧：普号必须用真实手机号注册的普通账号，不要用刚注册一天的新号干监听，新号进群就挂脚本，特征太明显。会话文件固定保存，不要让每次重启都触发一次验证码登录，频繁登录本身就是风险信号。行为侧：监听进程除了收取消息，还要产生一点正常的“呼吸感”。每天随机给群里发一两条无关痛痒的回复，或者偶尔在几个群里冒个泡，避免账号变成一个“只收不发”的哑巴。消息侧：发送频率、回复长度、是否带引用，都有参考值。下面这张表是我常用的参数范围：
| 参数 | 新号建议值 | 老号建议值 | 风险说明 |
|---|---|---|---|
| 连续回复条数上限 | 单次最多 3 条 | 单次最多 8 条 | 超过容易触发限流 |
| 发送最小间隔 | 5 秒 | 3 秒 | 秒回是危险信号 |
| 单条回复字数 | 20 字以上 | 无所谓 | 太短显得是脚本 |
回复带reply_to | 建议不带 | 建议带 | 新号引用高频消息易被注意 |
| 每日主动发言次数 | 3 次以内 | 10 次以内 | 维持“活人感” |
这套参数表不是拍脑袋，对应的是 Telegram 风控里对自动化行为的常见判定维度：消息频率、回复模式、账号活跃曲线。最后说合规边界：这套源码只应用于你本人所在的群、你有管理权限或获得授权的群。拿普号去监控别人私有群、抓取用户隐私数据、做骚扰式营销，都是明确不能碰的红线，账号一旦被封没有任何后悔药。
5. 常见问题排查：普号掉登录、漏消息、回复被吞的五个坑
5.1 坑一：登录状态明明有效，重启后却提示未授权
现象：监听器跑了一周，重启进程后 Telethon 报错，要求重新输入手机号和验证码。原因：session 文件没落在固定路径，或者临时目录里的文件被系统清理掉。很多源码把 session 文件写在当前目录，你用systemd跑和用nohup跑的工作目录不一样，重启后找不到旧 session，自然要重新登录。解决：把 session 文件路径设置成绝对路径，例如/opt/tg_monitor/session/account.session，并给文件权限设成600，再配合chattr +i防止意外删除。顺带检查一下是不是有多个进程共用了同一个 session 文件，Telethon 多进程同时读同一个 session 会导致文件损坏，表现也是“未授权”。
5.2 坑二：关键词命中后重复推送，同一条消息被人工处理了三次
现象：群里一条消息同时包含“价格”和“稳定”，操作员收到了两条几乎一样的待办任务。原因：规则文件里两个规则都命中了这条消息，而监听器没有做“同一消息的去重”。解决：在 TaskQueue.push 之前，用(group_id, msg_id, rule_id)做一次唯一性检查，同一条消息同一个规则只允许入队一次。更稳妥的做法是给 SQLite 的 tasks 表加唯一索引，数据库层面兜底，防止并发情况下代码判断失效。
5.3 坑三：人工回复发送成功但群里看不到，或者延迟几分钟才出现
现象：响应台显示发送成功，但群里就是看不到；要么就是客服那边过了很久才冒出来。原因：多半是 FloodWait 已经在后台潜伏发生，Telethon 的send_message在等待期内不会报错，而是把消息挂起排队，看起来是成功了，实际卡在发送队列里。解决：看sender.py的日志里有没有FloodWaitError记录，有就把min_interval调大。另一个低频原因是回复的reply_to指向的消息已经被删除或撤回，Telegram 会拒发引用消息，这种报错是MessageIdInvalidError，处理方式是在发送接口里捕获异常并退化为不带引用的发送。
5.4 坑四：正则规则导致 CPU 打满，消息一来整个进程卡死
现象：群内消息量一涨，监听器进程 CPU 飙到 100%，消息处理延迟明显变长。原因：规则里写了包含大量回溯的正则，比如(.*)*这类贪婪嵌套表达式，Telethon 的 handler 又是单线程事件循环，一个消息卡住，后面所有消息排队等。解决：在规则读取时用re.compile预编译所有正则，不要每次匹配都现编译；正则本身做复杂度限制，任何带.*嵌套的模式都要重写；给每条消息的匹配设超时，Python 的regex库支持超时参数，标准re不支持，所以生产环境我倾向于换regex模块跑低质量规则。最保险的方式还是先做关键词粗筛，只有包含相关关键词的消息才跑正则，不相关的则直接跳过。
5.5 坑五：操作员回复的内容带着时区错乱标记，群友以为你是机器人
现象：回复文本里包含时间信息时，比如“我们 9 点前给你答复”，操作员在北京时间 9 点发的，群里却显示的是 4 点。原因：响应台 Web 页面默认读本地时区，而发送出去的文本是字符串拼装，没有做统一的时区归一化。解决：后端统一用 UTC 存储时间，前端页面显示时再转换到操作员的本地时区；如果在回复文本里嵌时间，模板里固定写“北京时间 UTC+8”或“群里显示时间”的说明，不给歧义留空间。
6. 进阶：把监听从“能用”做到“稳”：规则灰度、消息去重与上线验收清单
6.1 规则灰度：新关键词别直接上全量，先在单个群试跑
直接往生产的规则文件里加一个大词，比如把“稳定”加进监听范围，第二天操作台的待办量可能直接翻十倍。我踩过一次坑，以为加一个词没事，结果群里有个人发了一长串广告，里面带“稳定”的文本被拆成三四条命中，整个操作台被刷屏。现在我的做法是给规则文件加一个enable_gray_groups字段，新规则先只在粉丝量少的内部试运行群里生效，观察两天的命中量和误报率，再决定要不要推全量。这样的灰度机制也能让操作员在响应台直接对某条命中点“误报”按钮，把误报样本自动存进一个 exclusion 列表，下次同类型的消息直接过滤。
6.2 群内去重与跨群同类话题聚合：从“处理消息”升级到“处理事件”
同一句话被转发到五个群，每个群各命中一次，操作员就得处理五遍，这是典型的低效。推荐在规则引擎之上再做一层聚合：以命中文本的相似度为键，30 分钟内视为同一事件，操作台只展示一条“该话术在 3 个群被提及”，操作员可以一次性回复所有群或选择忽略。相似度匹配不用引入向量库，简单的文本归一化加哈希即可：去掉所有空白字符、标点、emoji，再取前 30 个字符做分组键。这套逻辑能让操作员每天处理的消息量直接砍半，是提升系统体验性价比最高的一步。
6.3 上线验收清单：从部署到稳定运行的七项检查
最后给你一张我在每套监听系统上线前都会过的验收清单，照着打勾就行：
| 检查项 | 验证方法 | 通过标准 |
|---|---|---|
| 登录持久化 | 重启监听器进程 | 无需重新验证码 |
| 群消息接收 | 目标群里手动发一条测试消息 | 监听日志秒级出现 |
| 关键词命中 | 发一条带关键词的消息 | 操作台出现待办任务 |
| 去重生效 | 同一条消息命中两个规则 | 只产生一条待办 |
| 人工回复 | 操作台点击回复 | 群里 5 秒内出现且无报错 |
| 发送节流 | 连续回复三条消息 | 时间间隔不小于设定值 |
| 掉线重连 | 手动断开服务器网络 30 秒 | 监听进程自动重连并补拉离线消息 |
做这套系统的过程中，我最大的感受是：关键词监听本身不难，难的是让一个普通账号在持续自动化的前提下不出事。所以我的习惯是所有激进策略永远先在“可牺牲的测试号”上跑一周，确认行为特征没有异常再迁移到正式号上。希望这个方案能帮你把盯群的精力省下来，放到真正需要人判断的回复上，也希望你在用自己的普号跑这套源码前，先把合规边界想清楚——技术是工具，别让工具反过来烧了你的账号。最后说一句：所有参数都是起点不是终点，你的群活跃度、账号年龄、发送频率都得慢慢调，才能找到最适合自己那台机器的最优曲线，希望帮到你。
本文还有配套的精品资源，点击获取
