简介这是一套面向电报TG群组运营与营销场景的关键词监听机器人源码基于普通账号实现适合需要监控多个群或频道消息、及时捕捉潜在客户关键词的运营者与开发者。其核心思路是让监听号潜伏在群内当有人发送预设关键词时自动通知再由其他账号私聊触达且普通号不易被群成员察觉兼顾隐蔽性与实用性。资源包共52个文件约100KB以36个php脚本为主体配合html页面、json配置、txt说明文档及docker-compose.yml、sh启动脚本等部署文件结构上分为配置、监听逻辑、运行时与容器编排几部分便于二次修改与快速上线。目前已有518人学习下载热度稳定。读者可获得完整可运行的关键词监控源码、配置示例与部署脚本并参考说明文档理解监听流程与关键词配置方式适合作为群组营销工具或消息监控项目的实践起点。1. 关键词监听机器人普号也能跑的电报群消息监控方案做社群运营或者竞品情报的同行大概率都遇到过这种场景手里攥着几十个电报群想第一时间抓到某个关键词出现比如品牌名、竞品型号、行业黑话靠人肉盯屏根本不现实眼睛盯瞎了还漏消息。这份关键词监听机器人源码就是冲着这个痛点来的——它用普通账号普号登录监听指定群组里的消息流命中关键词就触发记录或转发同时保留人工实时监听的入口方便运营同学随时接管判断。它适合三类人一是做闲鱼关键词监控、二手交易情报的玩家二是需要盯竞品动态的市场人员三是想拿一套能改的 Python 源码自己二次开发的工程师。整套逻辑不依赖机器人账号普号即可跑降低了账号门槛但也意味着并发和风控上要更小心后面会细说。2. 环境搭建与普号登录把监听机器人跑起来的第一步2.1 技术栈选型与依赖安装这套源码走的是 Telegram 客户端协议路线常见做法是基于 Telethon 或 Pyrogram 这类 MTProto 库而不是 Bot API。原因很直接Bot API 拿不到群里的全量消息机器人必须被拉进群且权限受限很多私密群根本进不去而普号登录后只要账号在群里就能像正常用户一样读取消息流这才是关键词监听能落地的前提。我一般会先建一个干净的虚拟环境避免和系统里的其他库打架python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install telethon python-dotenvTelethon 负责和 Telegram 服务端通信python-dotenv 用来把 api_id、api_hash、手机号这些敏感信息从代码里剥离到 .env 文件避免硬编码泄露。装完之后建议 pip list 确认一下版本Telethon 不同大版本在事件注册的写法上有差异1.x 和 2.x 的events.NewMessage参数不完全一样踩过这个坑的都知道报错信息有多迷惑。2.2 申请 api_id 与配置凭证普号登录绕不开 api_id 和 api_hash这两个值要去 Telegram 官方的应用管理页面自己申请填个应用名和平台就行审核是即时的。拿到之后写进 .env# .env API_ID1234567 API_HASHabcdef1234567890abcdef1234567890 PHONE8613800000000 SESSION_NAMEmonitor_session这里 PHONE 要带国际区号SESSION_NAME 是会话文件的命名登录成功后会在本地生成一个 .session 文件下次启动就不用重新收验证码了。注意这个 session 文件等同于账号凭证别随手丢进 Git 仓库我见过有人把 session 传到公开仓库结果账号被踢下线的血泪案例。2.3 首次登录与验证码处理首次登录需要接收 Telegram 发来的验证码如果账号开了两步验证还得输密码。下面是一段最小可运行的登录骨架import os from dotenv import load_dotenv from telethon import TelegramClient load_dotenv() api_id int(os.getenv(API_ID)) api_hash os.getenv(API_HASH) phone os.getenv(PHONE) session os.getenv(SESSION_NAME, monitor_session) client TelegramClient(session, api_id, api_hash) async def main(): await client.start(phonephone) me await client.get_me() print(f登录成功: {me.username or me.id}) with client: client.loop.run_until_complete(main())client.start()会自动判断 session 是否有效无效就触发验证码流程终端里会提示你输入。get_me()用来验证登录态能打印出用户名或 ID 就说明通了。参数上 phone 必须和申请 api_id 时用的账号一致否则会报验证码收不到或者账号不匹配。跑通这一步后面监听逻辑才有地基。3. 关键词匹配与消息监听核心逻辑怎么写才不漏不炸3.1 事件注册与消息过滤监听的核心是给客户端注册一个 NewMessage 事件然后在回调里做关键词判断。Telethon 支持在事件层做初步过滤把不需要的群和消息类型挡在外面减少回调压力from telethon import events # 要监听的群可以是用户名、邀请链接解析后的 ID 或数字 ID TARGET_CHATS [-1001234567890, some_group_username] # 关键词列表命中任意一个就触发 KEYWORDS [求购, 出闲置, 型号A, 型号B] client.on(events.NewMessage(chatsTARGET_CHATS)) async def handler(event): text event.message.message or if not text: return for kw in KEYWORDS: if kw in text: sender await event.get_sender() print(f[命中] 群{event.chat_id} 关键词{kw} 发送者{sender.id}) print(f内容: {text[:120]}) breakchats参数限定只监听目标群避免全量消息涌入。event.message.message拿的是纯文本图片、语音这类没有文本的消息直接跳过。关键词用简单的in判断适合大多数场景如果要支持正则或者大小写不敏感把判断换成re.search并加re.IGNORECASE即可。这里用 break 是为了同一条消息命中多个关键词时只记一次避免刷屏。3.2 关键词匹配策略与参数调优简单in匹配在中文场景下有个坑分词边界模糊比如关键词「型号A」会命中「型号AB」这种误报。常见做法是给关键词加边界约束或者用正则的\b对中文效果有限配合前后字符判断。我一般会准备两套关键词一套精确词走正则一套模糊词走包含匹配分开配置import re EXACT_PATTERNS [re.compile(r型号A(?!\w)), re.compile(r求购\s*\d)] FUZZY_KEYWORDS [出闲置, 低价出] def match_keywords(text): hits [] for p in EXACT_PATTERNS: if p.search(text): hits.append(p.pattern) for kw in FUZZY_KEYWORDS: if kw in text: hits.append(kw) return hits精确模式用负向断言(?!\w)挡住后缀扩展模糊词保留包含匹配的召回率。参数上关键词列表建议外置到 JSON 或数据库改词不用重启进程监听群列表同理运营同学随时能加群减群。命中之后除了打印通常还要落库或转发到指定频道这部分按业务接就行。3.3 人工实时监听的接入方式源码里提到的「人工实时监听」不是让机器人替人判断而是提供一个可交互的通道命中关键词后把消息推到指定会话人工在那边看上下文、决定要不要跟进。实现上可以复用同一个客户端把命中消息转发到一个私有频道或收藏夹ALERT_TARGET me # 转发给自己也可以填频道 ID async def alert(event, text, hits): await client.send_message( ALERT_TARGET, f命中 {hits}\n群: {event.chat_id}\n内容: {text[:200]} )ALERT_TARGET填me就是发到自己收藏夹适合个人用团队场景填一个内部频道 ID多人同时看。这样人工监听和自动匹配是解耦的机器人只管捞人只管判职责清晰。注意转发频率要控制关键词太宽泛时容易把自己刷爆后面避坑章节会讲限流。4. 避坑与常见问题排查普号监听的五个真实翻车点4.1 现象登录后收不到验证码原因通常是手机号格式不对或者账号本身被限制登录。Telegram 对频繁换设备登录的账号会触发风控验证码可能延迟甚至不下发。解决方式是确认 PHONE 带国际区号先在一台常用设备上正常登录一次稳定几天再跑脚本。如果一直收不到换用已登录设备的 session 文件迁移比反复请求验证码安全。4.2 现象监听一段时间后掉线普号长时间挂着容易被服务端断开尤其是网络抖动后。原因是 MTProto 连接没有自动重连或者 session 被其他设备挤掉。解决方式是在启动时加自动重连参数并捕获断开事件重新连接client TelegramClient(session, api_id, api_hash, connection_retries5, retry_delay3, auto_reconnectTrue)connection_retries控制重试次数retry_delay是重试间隔秒数。另外别在多个地方同时用同一个 session 文件会互相踢。4.3 现象关键词命中率低或误报多原因是匹配策略太单一。中文没有天然词边界纯包含匹配要么漏要么炸。解决办法是分层配置高频精确词用正则加边界长尾词用包含匹配再定期看命中日志调整词表。我一般会先跑一天只记录不告警统计命中分布再决定哪些词该收紧。4.4 现象消息量太大把程序拖死监听几十个活跃群时回调里做同步 IO比如写数据库、发 HTTP 请求会阻塞事件循环。原因是 Telethon 是异步框架回调里任何同步阻塞都会拖慢整个消息处理。解决方式是把耗时操作丢到异步任务或队列里import asyncio async def handler(event): text event.message.message or hits match_keywords(text) if hits: asyncio.create_task(alert(event, text, hits))用create_task把告警异步化回调本身快速返回消息流就不会积压。4.5 现象账号被限制或封禁原因是行为太像机器人高频读取、短时间大量加群、频繁转发。普号的风控比机器人账号更敏感。解决方式是控制加群节奏监听群数量别一次拉满转发加个最小间隔。真被限制了先停脚本用官方客户端正常用几天再看。这块没有后悔药只能靠平时克制。5. 进阶技巧把监听机器人做成可维护的情报管道跑通基础监听只是起点真正拉开差距的是可维护性。我习惯把关键词、监听群、告警目标全部抽到配置文件代码只读配置不写死import json def load_config(pathconfig.json): with open(path, r, encodingutf-8) as f: return json.load(f) # config.json 结构示例 # { # chats: [-1001234567890], # exact: [型号A(?!\\w)], # fuzzy: [出闲置], # alert_target: me, # min_interval: 2 # }min_interval是同一关键词两次告警的最小间隔秒数用来防刷屏。加载配置后正则模式用re.compile预编译避免每次匹配都重新编译。这套结构改词、加群、换告警目标都不用动代码运营同学自己就能维护。验证监听是否真的在工作别只看日志我一般会做一次自测用小号在目标群发一条含关键词的消息看告警是否在几秒内到达。如果没到按顺序查——session 是否有效、群 ID 是否正确、关键词是否拼错、告警目标是否可达。这个自测流程我每次改完配置都强制走一遍比事后翻日志快得多。还有一个容易被忽略的点消息去重。Telegram 在断线重连后可能重推部分消息导致同一关键词重复告警。常见做法是用(chat_id, message_id)做唯一键落库前查一下或者用内存里的 LRU 缓存挡最近几百条。这个细节不做重连一次就能把你刷到怀疑人生。从那以后我每次部署监听机器人都先把配置外置、自测流程、去重逻辑这三样配齐再上线省下来的排查时间远比前期多花的十分钟值。希望帮到你。本文还有配套的精品资源点击获取