简介一套基于 PHP 的关键词监听机器人源码用于在 Telegram 电报群中监控群聊消息当内容命中预设关键词时自动触发通知支持人工实时监听。面向对电报机器人开发、PHP 常驻进程、群消息事件处理感兴趣的开发者适合个人学习与技术研究。全部文件共 57 个压缩包仅 85KB其中 36 个 PHP 文件构成核心逻辑涵盖服务入口、配置加载、事件观察者、控制器路由、异常处理、日志记录与数据库迁移等模块另有 JSON/YAML 配置文件、Shell 部署脚本和 DOCX 教程文档辅助理解目录清晰易于定位。已有 132 人学习。通过这份源码读者可以了解监听机器人的完整目录结构与分层设计掌握关键词匹配、事件分发、日志记录等功能的实现思路附带的配置示例与部署脚本还能帮助快速搭建可运行的同类工具或在此基础上扩展更多群聊自动化能力。代码中还包含多种客户端库适配与多账号监听方案适合深入拆解其消息接收和身份隐藏机制。1. 关键词监听机器人解决什么问题盯群这件事机器先做、人工把关一个针对TG电报群消息的关键词监听机器人要解决的问题非常具体你在几个群里但做不到7×24小时盯着屏幕群里又随时可能出现不能错过的话——竞品报价、客户吐槽、活动链接、行情喊单。手动切群看一定会漏群里消息多的时候甚至会漏掉最该第一时间知道的那一条。这套方案把“盯群”拆成两段监听机器人先按关键词把新消息全部筛一遍命中后推到你和同事面前再由人工实时确认、处置。机器负责不遗漏人负责做判断。它适合社群运营、品牌舆情、行情监控、渠道投放这几类从业者也适合想自己维护源码的开发者。监听范围有个前提只处理你账号本身就能看到的群消息不碰任何不可见的私密对话。下面按我实际落地时踩过的路径展开从选型一直讲到排查。2. 为什么TG群关键词监听首选Telethon用户协议Bot API的三条硬边界2.1 监听机器人的两种形态官方Bot与Userbot在做TG群关键词监听之前要先分清两个东西官方Bot和Userbot。官方Bot是你在BotFather里创建的机器人账号走Telegram公开的HTTP接口Bot API。它的定位是“在群里提供服务”自动回复、发命令、定时推送、踢人禁言都合适。Userbot则是拿你自己的普通用户账号通过MTProto协议客户端库Python里最常见的是TelethonPHP圈有MadelineProto把账号当成一个可编程客户端。标题里的“监听机器人”在我处理过的绝大多数方案里用的都是后者而不是Bot API。判断标准很简单如果需求是“给群提供功能”用Bot如果需求是“把我视角内看得见的每一条群消息按关键词捞出来”就得Userbot。关键词监控的核心诉求是“不漏”而Bot这种官方形态天生就不适合“不漏”这件事。2.2 Bot API的三条硬边界第一条边界是隐私模式privacy mode。Bot在群里默认只收到命令、被提到它的消息和直接回复它的消息群成员之间聊得再火热Bot API回调里一个字都没有。要让Bot能“听到”群里的日常消息必须在BotFather里/setprivacy选择Disable或者把Bot设为群管理员。实际运营中大部分群主不会为你的监听需求把Bot设成管理员这一步经常卡住。第二条边界是群类型的限制。私有群、私有频道的Bot权限更严格Bot被邀请进群后Bot API能拿到的消息上下文和媒体信息并不完整补拉历史消息也很别扭。而做关键词监听经常要面对的正是这类不公开的群。第三条边界是推送机制。Bot API没有真正的服务端长连接依赖webhook或长轮询getUpdates拉取更新。监听群多了以后请求频率稍微上去就吃429限流Bot API更新补拉还有时间窗口限制重启服务后丢消息是常态。Telethon这类MTProto客户端走的是长连接服务端有新消息直接推过来机制完全不同实时性也完全不是一个量级。2.3 MTProto的核心优势账号视角全可见、长连接实时推送Telethon用你的用户账号登录后Telegram服务端会把账号视角下所有可见消息作为update实时推送。这个update里带着聊天ID、消息ID、发送者ID、时间戳、文本、媒体描述caption和附件信息基本能还原你本人在TG客户端里看到的一切。这个“可见”非常关键不需要把账号设为群管理员不需要关闭隐私模式不需要被群主授权因为你登录的就是一个普通群成员的账号群消息对你本来就是可见的。监听进程只需要订阅事件收到消息后做关键词过滤命中就转发。GitHub上能搜到不少免费python源码里面监听类脚本一大半是Bot API写法或半成品Telethon写法大多只做了“收消息判断关键词”缺持久化、去重和人工处置闭环。类似“闲鱼关键词监控”那种需求也是同一个套路——事件加规则加推送区别只在于数据源不同但真正能稳定跑在群聊实时事件上的方案很少。后面三章就是在补这几块最小链路、命中后的人工实时监听、以及运维期的坑。2.4 选型对照与需要提前准备的事下表适合直接抄进方案文档里对比维度官方BotBot APIUserbotTelethon群内消息可见性受隐私模式和管理员权限限制账号视角下全部可见实时性轮询或webhook有轮询间隔长连接服务端推送秒级历史消息补拉受限且有时窗iter_messages支持分页回放消息上下文部分群拿不完整与客户端一致账号风控机器人形态风险低用户账号自动化有封号风险适用场景客服机器人、指令机器人群关键词监控、数据采集前置准备四件事第一到Telegram官方开发者后台my.telegram.org申请api_id和api_hash这是Userbot登录凭据BotFather的token这里用不上。第二准备一个专门的监听账号别用主力号——自动化行为一旦被判定异常账号会被限制备用号损失可控。第三提前把所有要监听的群都入好确认账号能看到群里的历史消息和新消息。第四评估风险Telegram服务条款不允许用户账号自动化Userbot有封号可能做生产方案之前先想清楚业务能不能承受账号失效以及不要把它用在任何不可见的私密对话上。注意如果需求只是“某个群出现特定指令就自动回复”请用官方Bot别拿Userbot硬扛风控差异非常大。3. 30分钟跑通最小监听链路Telethon登录、提取群消息与关键词命中3.1 依赖与登录先把会话跑起来环境准备就两行pip install telethon登录代码是最小形态from telethon import TelegramClient api_id 123456 # 开发者后台申请的API ID api_hash abcdef... # 开发者后台申请的API Hash client TelegramClient(monitor, api_id, api_hash) client.start() client.run_until_disconnected()首次运行会在终端提示输入手机号、验证码账号开了两步验证还要输入密码。登录成功后登录态缓存在当前目录的monitor.session文件里之后重启不会再要求验证码。这个session文件就是账号的“后悔药”我一般会在每次改动代码前把它备份一份服务器上跑定时任务时尤其重要session丢了等于账号要重新走一遍风控验证流程。参数说明TelegramClient第一个参数是session名字也可以是路径start()不传参数时走交互输入。服务器上无人值守跑建议改成client.start(phonelambda: 手机号)验证码通过环境变量或远端输入别把账号机密写死在源码里。run_until_disconnected()会一直跑事件循环这一步先确认session能稳定建立再往下面加逻辑。3.2 监听循环与关键词归一化匹配最小监听加关键词过滤import re from telethon import TelegramClient, events KEYWORDS [报价, 现货, hype] client TelegramClient(monitor, api_id, api_hash) client.on(events.NewMessage()) async def on_new_message(event): raw event.raw_text or folded raw.casefold() for kw in KEYWORDS: if kw.casefold() in folded: print(f命中 chat{event.chat_id} msg{event.message.id}: {raw}) break client.start() client.run_until_disconnected()逻辑说明events.NewMessage()不传chats时监听的是账号可见的全部新消息先全局收集再过滤适合第一版验证。event.raw_text会同时取消息正文和媒体描述caption这是最容易踩对又踩错的地方——很多网上的源码只取event.text文字写在图片描述里的消息会全部漏掉。casefold()是比lower()更彻底的归一化英文关键词建议统一先做这一层否则“BTC”和“btc”会被当成两个词。正则规则可以直接叠加RULES [ re.compile(r(BTC|bitcoin)\s*(报价|价格), re.I), re.compile(r空投|airdrop, re.I), ]这时循环里把关键词判断换成any(rule.search(folded) for rule in RULES)正则的好处是能容忍变体比如行情相关群经常把“BTC / USDT 报价”写成带空格或大小写混排的形式re.I加上\s*能覆盖大部分这类情况。关键词如果交给非技术同事维护我建议把规则放外部JSON文件程序启动时加载运营加词不用改代码重启。3.3 命中后最小动作原样转发到处置群DISPATCH_CHAT ops_group # 处置群的username或数字ID client.on(events.NewMessage()) async def on_new_message(event): raw event.raw_text or folded raw.casefold() if any(rule.search(folded) for rule in RULES): await event.message.forward_to(DISPATCH_CHAT)forward_to把原消息原样转发到处置群人工在TG里看到的就是群里那条原文发送者、时间、图片、文件全都在。这比“复制文本再发一遍”强在保留完整上下文——竞品投放、行情波动这类场景上下文本身就是情报。如果想在消息里附带链接方便一键跳转构造方式要注意私有群和公开群的区别def message_link(event): chat event.chat if getattr(chat, username, None): return fhttps://t.me/{chat.username}/{event.message.id} # 私有群/频道 id 形如 -100xxxxxxxxxx链接只保留后面的数字段 peer_id str(chat.id) if peer_id.startswith(-100): peer_id peer_id[4:] return fhttps://t.me/c/{peer_id}/{event.message.id}说明公开群用t.me/群用户名/消息ID就能跳转私有群没有公开用户名只能走t.me/c/群ID/消息ID格式而且ID要去掉-100前缀。如果消息来自单聊这个链接不适用所以生产实现里要先判断event.is_group或event.is_channel再接这个函数。3.4 “实时”的真相事件推送、断线重连与补拉对账“实时监听”是相对的。长连接稳定时服务端推送消息到进程到命中转发延迟一般在1到3秒。但有两类情况会打破“不漏”这个前提。第一类长连接断线的窗口。进程崩溃、服务器重启、网络切换重连期间Telegram不会把每一条错过的消息都补推给你Telethon提供了client.catch_up()处理离线积压更新但生产环境我更信任对账方案——定时主动拉最近的消息和本地已处理的记录做差集缺的补跑一遍关键词。这才是“不漏”的最后一道保险。from datetime import datetime, timedelta from telethon import TelegramClient async def reconcile(client, chat_id, aftertimedelta(minutes10)): since datetime.now() - after async for msg in client.iter_messages(chat_id, offset_datesince): # 与本地DB比对 msg.id缺失的重新走关键词逻辑 pass第二类账号风控导致的被动离线。账号被限制登录期间事件通道直接静默这个没有代码层面的解法只能靠对账和人工巡检发现。实时性受影响最大的一环不在网络而在于事件处理器里不要做重活——不要在命中回调里做网络请求、OCR或复杂计算这些会阻塞事件循环消息一多实时就变成延迟。命中的消息先转发转发之后再开协程去补元数据是更稳妥的写法。4. 关键词命中之后的人工实时监听推送通道、去重与审核闭环4.1 两种人工监听落地方式机器只能保证“找到”不能保证“判断得对”。关键词监控里一旦加了排除词、同义词和上下文规则误报率就不可能为零。人工实时监听的价值是机器筛出候选人做最终确认。落地方式分两种。轻量级直接在Telegram里建一个处置群把命中消息全部转发进去坐席在TG里看、在TG里标记。适合1到3人、日命中几百条的场景这也是最容易复现的方式。重量级把命中消息通过Webhook推到企业微信、钉钉、Slack或者自建一个浏览器页面多个坐席跟着实时消息流处理适合团队化和需要操作留痕的场景。无论哪种方式有两件事必须先做去重和持久化。否则处置群刷屏、重复消息把人工完全淹没是新手必翻的翻车点。4.2 SQLite落库与msg_id去重在Telegram体系里chat_id加msg_id能唯一定位一条消息。这个唯一键就是去重阀。import aiosqlite DB_FILE monitor.db async def init_db(): async with aiosqlite.connect(DB_FILE) as db: await db.execute( CREATE TABLE IF NOT EXISTS hits ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER NOT NULL, msg_id INTEGER NOT NULL, rule TEXT, hit_text TEXT, status TEXT DEFAULT open, created_at TEXT DEFAULT (datetime(now)), UNIQUE(chat_id, msg_id) ) ) await db.commit() async def save_hit(chat_id, msg_id, rule, hit_text): async with aiosqlite.connect(DB_FILE) as db: try: await db.execute( INSERT INTO hits(chat_id, msg_id, rule, hit_text) VALUES(?,?,?,?), (chat_id, msg_id, rule, hit_text), ) await db.commit() return True except aiosqlite.IntegrityError: return False主监听事件里转发之前先save_hit返回False说明这条消息已处理过直接return。这个约束把重复推送堵死在入口。为什么用aiosqliteTelethon的事件循环是异步的事件回调里如果用同步sqlite3每次写库都会短暂阻塞整个监听循环命中频率一高延迟拉满。aiosqlite把写库操作丢到线程池不阻塞事件处理。如果日命中只有几十条普通sqlite3也能凑合但团队复现建议一步到位。参数说明status字段留给人工标记默认open表示未处理人工处理完改成done后续报表、告警、统计都基于这个字段。hit_text存原始文本方便事后复盘误报——这个字段在调关键词规则时是救命数据。4.3 处置群内的人工标记/done指令处置群是人工坐席区。命中消息转发进来后人工看到觉得这条需要跟进就回复一条“/done”机器人自动把对应记录的状态改成done。client.on(events.NewMessage(chatsDISPATCH_CHAT)) async def on_ops_message(event): if not event.is_reply: return cmd (event.message.text or ).strip() if cmd.startswith(/done): reply await event.get_reply_message() await mark_done(reply.id) async def mark_done(msg_id): async with aiosqlite.connect(DB_FILE) as db: await db.execute(UPDATE hits SET statusdone WHERE msg_id?, (msg_id,)) await db.commit()逻辑说明这里监听的对象换成了处置群只处理回复消息。event.get_reply_message()拿到的是人工回复的那条原始命中消息。结合msg_id去更新状态人机通过消息的上下文完成一次闭环。操作上的两个细节第一处置群如果人多建议设成“仅管理员可发言”或约定只有坐席能发指令避免误触发第二/done只是语义上的约定除了Telegram客户端自带的slash命令提示没有任何框架层面的约束所以指令解析要容忍前后空格和大小写。4.4 进阶WebSocket把命中消息实时推给浏览器坐席一旦日命中上千条处置群会变成刷屏地狱常见做法是加一个Web面板。原理不复杂监听进程命中后把消息结构发到一个本地WebSocket服务浏览器页面保持连接新消息实时弹出来坐席在页面上点击“处理”。一个FastAPI的骨架from fastapi import FastAPI, WebSocket app FastAPI() clients set() app.websocket(/ws/hits) async def ws_hits(ws: WebSocket): await ws.accept() clients.add(ws) try: while True: await ws.receive_text() # 只做心跳保活 except Exception: clients.discard(ws) async def broadcast(payload: dict): for ws in list(clients): await ws.send_json(payload)监听进程命中后调用broadcast({...})所有在线的坐席页面同时收到这条候选消息。相比处置群Web面板可以按规则、按群、按状态筛选操作也全部留痕在SQLite里。代价是多维护一个服务。个人小规模使用处置群完全够用先把4.2和4.3跑通量上来了再引入Web面板不迟。5. 监听机器人常见问题与避坑漏消息、重复推送、账号风控的排查记录5.1 现象进程活着但新群消息收不到现象最迷惑进程日志正常、没有报错群里明明有新消息监听端就是没动静。原因通常是三个叠加监听账号在另一台设备上也登录着Telegram的会话策略会把更新分流到其他登录端导致这个进程收不到全部推送或者启动时事件注册晚于连接建立离线期间积压的update没被消费再或者长连接断开后没有重连成功进程看起来活着实际上已经“死”了。解决监听账号只在服务器这一个会话里保持登录其他设备全部退出启动流程里先await client.catch_up()再进入监听循环同时用对账脚本兜底这类问题如果发生在凌晨人工基本发现不了最后能抓住漏报的只有增量对账。注意查这类问题先开日志用logging.basicConfig(levellogging.INFO)看Telethon的连接状态和update接收心跳进程“假活”会立刻暴露。5.2 现象同一条关键词消息被推送了两次现象处置群里重复消息成对出现去重表里也能查到重复记录。原因八成是转发逻辑和链接构造逻辑同时启用——一条消息既被forward_to转发了又被自定义文本拼接链接发了一遍。还有一个隐蔽原因对账任务和事件任务同时命中同一条消息两个协程都过了去重判断竞争条件导致双写。解决所有进入处置群或推送通道的操作统一走save_hit这条入口以SQLite插入成功为前置条件插入失败直接放弃对账任务只处理“已落库但未推送”的消息而不是把差集全量重新转发一遍。去重逻辑本身并不难难的是给每一条对外发送路径都套上同一道闸。5.3 现象明文关键词命中不了“被改写”的群聊原文现象规则里写着“BTC”群里天天聊“大饼”“饼子”“bitcoin”一条都命不中监控某个活动名群友用缩写、表情、Unicode空格把词拆得稀碎。原因很直白关键词监控本质是字符串匹配群聊语言却是高度变体的人和人之间的交流充满别名、暗语和输入习惯差异。解决维护一张高频同义词表把别名映射到标准词匹配时先做一次归一化替换英文关键词统一casefold并压缩连续空白正则规则写成容忍中间空格和标点变体的模式比如r华\s尔\s街能命中多种写法。但要清醒这类变体补全是无底洞只补高频词别追求100%召回剩下的靠人工实时监听兜底。5.4 现象图片/文件的文字在描述里监听逻辑却把消息丢了现象群里发了一张行情截图描述文字里带着关键价格监听端没有任何反应。原因大概率是源码里写了一句if not event.message.text: return——图片类消息的正文在caption里而event.message.text对纯媒体消息往往是空字符串这一行continue把整条消息滤掉了。网上流传的免费python源码里这个写法非常常见。解决统一取event.raw_text它同时覆盖正文和媒体描述如果raw_text还是空再取event.file.name做文件名匹配再往后如果消息里的图片本身带关键信息接OCR是另一个话题代价较高通常先不加。代码上一行就能修正raw event.raw_text or if not raw: raw event.file.name or 5.5 现象userbot账号被要求二次验证或登录受限现象某天监听账号突然要求输入验证码或登录后立刻掉线再或者发送消息被禁。原因不是代码bug是用户账号自动化本身违反Telegram服务条款风控系统按登录端、行为频率、消息模式综合判断。这不是靠换代码能解决的属于账号使用层面的风险。解决不用主力号跑监听账号风险与业务解耦自动化行为密度尽量低——只读不主动发能转发就不发新文本账号在官方客户端定期登录活跃一下维持正常使用画像一旦出现受限先停服务观察不要用任何方式绕过绕过只会加重风控。业务规划上把监听任务分级高价值群的监听预留官方Bot通道作为紧急备份Bot形态在风控上比Userbot安全得多。6. 验证监听链路可靠性的三个小技巧测试消息、对账脚本与灰度上线6.1 造一条“关键词测试消息”实测端到端延迟监听进程跑起来之后先建一个测试群用小号发一条包含测试关键词的消息比如“测试报价888”然后计时从发出到处置群收到转发中间隔了几秒。连续测三次取中位数能得到这个方案在当前网络环境下真实的端到端延迟。我在测试文本里都会带上时间戳这样看处置群消息的时间差就能直接算延迟不用另外翻日志。这一步能同时验证四件事session登录态有效、事件通道通、关键词规则命中、转发落库走通。如果处置群里压根没出现消息从“没转发、转发了但格式不对、延迟特别高”三个现象去回溯能快速定位是事件没收到、规则没匹配还是转发环节写崩了。6.2 对账脚本用增量拉取兜底验漏事件通道正常不代表不漏。用前面3.4的reconcile思路写一个独立对账脚本每5分钟拉取最近10分钟的消息和SQLite里已落库的msg_id做差集差集里的消息重新走一遍关键词匹配。连续跑24小时观察差集数量是否保持在0。如果差集反复大于0说明事件通道存在断流优先查5.1里的session冲突、长连接重连和账号状态。对账脚本本身就是漏报监控的最后一道防线上线监听的第一周先别急着加群把对账差集的稳定性当成核心指标看。6.3 灰度上线先低价值群、再扩展血泪经验我第一次做的就是一次性把所有群接进来结果命中洪峰直接把处置群刷爆人工完全跟不上。后来改成先单群、单关键词跑三天确认日志、去重、人工标记三个环节都顺了再按周扩容。扩容顺序建议先加群、后加词每加一批观察一天日命中量超过处置群人工处理能力之前先上Web面板或加坐席不要让消息积压。这套方案做完我最大的改变是不再迷信“监听等于关键词匹配”真正值钱的是命中之后的处置闭环和对账兜底。自己手写一遍才知道免费的python源码能给你事件循环但给不了你踩完坑之后的运维意识。希望帮到你。本文还有配套的精品资源点击获取