简介这是一套面向TelegramTG生态运营者的关键词监听机器人源码适用于社群营销、竞品监控与实时用户触达等场景尤其适合需要隐蔽部署、多群协同监听的中高级开发者或运营人员。资源包共52个文件以36个PHP核心脚本为主含server.php、work.php、bootstrap.php等主控与工作模块辅以3个配置类JSON与HTML说明文档、Docker容器化部署文件docker-compose.yml、start/stop/reload脚本及环境配置文件.env.docker、config.php整体仅100KB轻量易部署。已有511人学习下载资源结构清晰包含完整运行流程从监听关键词配置监听关键词.json、服务启停控制到人工介入触发私聊的逻辑闭环并附带《关键词监听教程.docx》《资源说明必看.txt》等实操指引。读者可直接复用该普号方案实现无痕监听——多个TG账号协同工作主号不露面有效规避群内识别与封禁风险。1. 项目缘起从被动搜索到主动感知的信息需求在信息爆炸的时代我们常常陷入一个矛盾一方面我们害怕错过任何重要的消息另一方面我们又无法忍受信息洪流带来的噪音。尤其是在一些即时通讯社群中比如TelegramTG上的技术交流群、行业动态群、项目社区等有价值的信息往往转瞬即逝被淹没在闲聊、表情包和无关链接里。手动爬楼翻看聊天记录效率低下且容易遗漏。这种“信息焦虑”催生了一个非常具体的需求能否有一个“哨兵”24小时不间断地帮我盯着群聊一旦出现我关心的关键词就立刻通知我这就是“关键词监听机器人”诞生的背景。它不是一个复杂的AI而是一个精准的“触发器”。其核心逻辑非常简单模拟一个真实的Telegram用户即“普号”普通账号加入目标群组持续监听所有新消息将消息文本与预设的关键词列表进行匹配。一旦匹配成功就通过预设的渠道如另一个TG私聊窗口、Webhook、邮件等将这条消息的上下文发送者、时间、内容、直达链接实时推送给监听者。我最初开发这类机器人是为了跟踪几个加密货币项目和开发者社区的动态。这些社群的讨论质量很高但消息刷屏速度极快人工盯盘几乎不可能。通过部署监听机器人我成功抓取到了多次重要的项目更新预告、漏洞披露和空投信息其价值远超投入。今天我就把自己在开发和运维这类“TG关键词监听机器人”过程中的核心思路、技术选型、避坑经验以及“人工实时监听”的增强方案毫无保留地分享出来。无论你是想追踪竞品动态、关注行业热点、收集用户反馈还是单纯不想错过群里的重要通知这套方案都能为你提供一个高可用的自动化解决方案。2. 核心架构解析一个监听机器人的五脏六腑一个稳定可用的TG关键词监听机器人远不止几行“if text in keywords”的代码。它需要一套健壮的架构来应对网络波动、TG API限制、消息格式多样性以及自身的稳定运行。下面我们来拆解它的核心组件。2.1 消息获取层与Telegram API的对话这是机器人的“眼睛”和“耳朵”。我们需要一个客户端与Telegram服务器建立连接并接收消息。主流有两种选择MTProto协议客户端这是Telegram的原生协议效率高、功能全。Telethon和Pyrogram是Python中最流行的两个MTProto库。它们需要你用一个真实的手机号注册一个Telegram应用api_id和api_hash这个账号就是你的“普号”。监听机器人就是用这个账号的身份在群里活动。Telethon更底层异步支持好功能强大但学习曲线稍陡。Pyrogram封装更友好同步/异步都支持文档清晰对新手更友好。选择建议对于监听这种I/O密集型任务异步是更好的选择。Pyrogram的异步模式asyncio写起来更简洁社区活跃是我目前的首选。Bot API通过BotFather创建的机器人账号。它无法以“普通成员”身份主动加入大多数需要审核的群组除非被设为管理员且无法读取机器人加入之前的消息历史。这对于“潜伏监听”公开或私密群组来说是一个致命缺陷。因此对于需要以普通成员身份监听任意群组的场景必须使用MTProto协议的“普号”方案。关键配置与初始化使用Pyrogram初始化一个客户端核心是妥善保管你的api_id,api_hash和会话文件。会话文件.session保存了登录状态避免每次运行都需重新登录和可能触发二次验证。from pyrogram import Client app Client( my_listener_session, # 会话文件名 api_idYOUR_API_ID, api_hashYOUR_API_HASH, phone_numberYOUR_PHONE_NUMBER # 国际格式如 8613800138000 ) async def main(): async with app: # 此时客户端已登录并准备就绪 # 可以在这里获取已加入的群组列表或等待消息 pass重要安全提示api_id和api_hash等同于你的账号密码绝对不要泄露或上传至公开仓库。会话文件也包含敏感信息需同等保护。建议使用环境变量或配置文件来管理这些凭证。2.2 消息处理与过滤层大脑中的模式匹配这是机器人的“大脑”。它需要处理原始消息并判断是否触发警报。这一步有几个关键考量消息类型处理Telegram消息不只有文本。还有图片、视频、文档、链接等。我们的关键词匹配主要针对文本。对于媒体消息我们需要提取其caption说明文字或文件名进行匹配。对于链接可能需要提取URL本身或预览标题。关键词策略精确匹配 vs 模糊匹配“空投”是精确匹配。但有时我们需要模糊匹配比如包含“漏洞”或“bug”的消息。这可以通过正则表达式实现例如r(漏洞|bug|安全問題)。关键词列表管理关键词不应硬编码在代码里。最好使用一个外部配置文件如JSON、YAML或数据库来存储支持动态增删改查无需重启机器人。排除干扰可以设置“排除词”列表例如当消息同时包含“广告”和“福利”时即使触发了“福利”关键词也不报警以减少垃圾信息干扰。上下文获取一条孤立的触发消息可能信息量不足。优秀的监听器应该能捕获这条消息的上下文例如消息链接生成一个可以直接点击跳转到该条消息的t.me链接需要chat_id和message_id。前后消息有时触发词出现在对话中需要结合前后几条消息才能理解。可以尝试捕获触发消息前后各2-3条消息内容一并上报。发送者信息记录发送者的ID、用户名如果有和显示名称有助于判断消息来源的可靠性。2.3 通知与上报层发出警报的嘴巴这是机器人的“嘴巴”。当匹配成功后需要以某种形式通知你。常见方式有Telegram私聊将触发消息转发到你的另一个TG账号或同一个账号的“Saved Messages”。这是最直接、最实时的方式。机器人可以调用send_message方法将格式化好的报警信息发给你。Webhook将触发消息的结构化数据JSON格式POST到一个你指定的服务器地址。这种方式非常灵活你可以用服务器对接钉钉、飞书、企业微信、Slack甚至存入数据库或触发其他自动化流程。邮件/SMS作为备用通知渠道但实时性较差。数据库存储所有触发消息都存入数据库如SQLite、MySQL、PostgreSQL便于后续统计分析、回溯和去重。一个简单的Telegram私聊上报示例async def send_alert(app, chat_id, message): 发送警报到指定私聊 alert_text f 关键词触发警报\n\n alert_text f**群组**{message.chat.title}\n alert_text f**发送者**{message.from_user.mention if message.from_user else N/A}\n alert_text f**时间**{message.date}\n alert_text f**内容**\n{message.text or message.caption or 无文本内容}\n alert_text f**直达链接**{message.link} # 假设你的个人用户ID是 123456789 await app.send_message(chat_id123456789, textalert_text)2.4 持久化与状态管理层机器人的记忆机器人需要“记住”自己的状态比如已处理消息ID避免网络重连或重启后重复处理旧消息。通常记录最新处理的消息ID只处理比它ID更大的消息。关键词列表动态加载最新的关键词配置。运行日志记录监听状态、错误信息便于排查问题。这些状态可以存储在内存中重启丢失但更可靠的做法是使用轻量级数据库如SQLite或者简单的文件如JSON来保存。3. 从零搭建手把手实现你的第一个监听机器人理论讲完我们进入实战。我将以Pyrogram异步为核心搭配SQLite和配置文件构建一个基础但功能完整的监听机器人。3.1 环境准备与依赖安装首先确保你的Python环境是3.7或更高版本。创建一个新的项目目录并建立虚拟环境。mkdir tg-keyword-listener cd tg-keyword-listener python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖pip install pyrogram[tgcrypto] aiosqlitepyrogram[tgcrypto]Pyrogram库及其加密依赖。aiosqlite异步版本的SQLite驱动避免数据库操作阻塞事件循环。3.2 获取Telegram API凭证登录 my.telegram.org 。选择 “API development tools”。填写应用信息任意如MyListener提交后你会获得api_id和api_hash。请妥善保存。3.3 项目结构与核心代码项目目录结构如下tg-keyword-listener/ ├── config.yaml # 配置文件 ├── keywords.txt # 关键词列表每行一个 ├── listener.py # 主程序 ├── database.py # 数据库操作 └── utils.py # 工具函数1. 配置文件 (config.yaml)api_id: YOUR_API_ID api_hash: YOUR_API_HASH phone_number: 8613800138000 # 你的手机号 alert_chat_id: 123456789 # 接收警报的TG用户ID session_name: listener_session # 可选需要监听的群组ID列表负数如果为空则监听所有已加入的群 target_chats: [-1001234567890, -1009876543210]如何获取chat_id你可以让机器人先运行起来在目标群组发一条消息然后在代码中打印出message.chat.id这个id就是chat_id。普通群组的id是负数。2. 关键词文件 (keywords.txt)每行一个关键词或正则表达式。空投 Airdrop (?i)bug # (?i)表示不区分大小写 漏洞 重要通知 紧急3. 数据库模块 (database.py)负责记录最新处理的消息ID实现断点续传。import aiosqlite import asyncio DB_PATH listener.db async def init_db(): 初始化数据库 async with aiosqlite.connect(DB_PATH) as db: await db.execute( CREATE TABLE IF NOT EXISTS last_message ( chat_id INTEGER PRIMARY KEY, message_id INTEGER NOT NULL ) ) await db.commit() async def update_last_msg_id(chat_id: int, message_id: int): 更新某个聊天最后处理的消息ID async with aiosqlite.connect(DB_PATH) as db: await db.execute( INSERT OR REPLACE INTO last_message (chat_id, message_id) VALUES (?, ?) , (chat_id, message_id)) await db.commit() async def get_last_msg_id(chat_id: int) - int: 获取某个聊天最后处理的消息ID如果没有记录则返回0 async with aiosqlite.connect(DB_PATH) as db: async with db.execute(SELECT message_id FROM last_message WHERE chat_id ?, (chat_id,)) as cursor: row await cursor.fetchone() return row[0] if row else 04. 主监听程序 (listener.py)这是机器人的核心逻辑。import yaml import re import asyncio from pyrogram import Client, filters from pyrogram.types import Message from database import init_db, update_last_msg_id, get_last_msg_id # 加载配置 with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 加载关键词 with open(keywords.txt, r, encodingutf-8) as f: keyword_lines [line.strip() for line in f if line.strip() and not line.startswith(#)] # 编译正则表达式普通关键词转为精确匹配的正则 keyword_patterns [] for line in keyword_lines: if line.startswith((?) or ( in line: # 简单判断是否为复杂正则 keyword_patterns.append(re.compile(line)) else: # 将普通关键词转换为单词边界匹配避免匹配到单词的一部分 keyword_patterns.append(re.compile(rf\b{re.escape(line)}\b, re.IGNORECASE)) app Client( config[session_name], api_idconfig[api_id], api_hashconfig[api_hash], phone_numberconfig[phone_number] ) def contains_keyword(text: str) - bool: 检查文本是否包含任何关键词 if not text: return False for pattern in keyword_patterns: if pattern.search(text): return True return False async def send_alert(message: Message): 格式化并发送警报 chat_title message.chat.title or 私聊/频道 sender message.from_user sender_info f{sender.first_name} {sender.last_name or } ({sender.username}) if sender else 未知 # 获取消息文本或说明 content message.text or message.caption or [非文本消息] # 截断过长的内容 preview (content[:200] ...) if len(content) 200 else content alert_msg ( f** 关键词触发警报**\n\n f**来源群组**{chat_title}\n f**发送者**{sender_info}\n f**时间**{message.date}\n f**消息预览**\n{preview}\n\n f[点击查看原消息]({message.link}) ) try: await app.send_message(chat_idconfig[alert_chat_id], textalert_msg, disable_web_page_previewTrue) print(f[警报已发送] {chat_title} - {message.date}) except Exception as e: print(f发送警报失败: {e}) app.on_message(filters.group ~filters.edited) # 监听群组新消息忽略编辑消息 async def handle_new_message(client: Client, message: Message): 处理新消息 chat_id message.chat.id # 如果配置了特定群组则检查是否在目标列表中 target_chats config.get(target_chats) if target_chats and chat_id not in target_chats: return # 获取该聊天最后处理的消息ID实现断点续传 last_msg_id await get_last_msg_id(chat_id) if message.id last_msg_id: return # 已处理过的旧消息跳过 # 检查消息内容是否触发关键词 text_to_check message.text or message.caption or if contains_keyword(text_to_check): await send_alert(message) # 更新最后处理的消息ID await update_last_msg_id(chat_id, message.id) async def main(): await init_db() # 初始化数据库 print(数据库初始化完成。) print(正在启动监听机器人...) async with app: # 获取所有已加入的对话可以打印出来确认 # async for dialog in app.get_dialogs(): # print(f{dialog.chat.title or dialog.chat.first_name} - ID: {dialog.chat.id}) print(监听已启动。按 CtrlC 停止。) await asyncio.Event().wait() # 永久运行等待事件 if __name__ __main__: try: asyncio.run(main()) except KeyboardInterrupt: print(\n监听已停止。)3.4 运行与测试将config.yaml中的api_id,api_hash,phone_number,alert_chat_id替换为你的真实信息。在keywords.txt中添加你想监听的关键词。在终端运行python listener.py。首次运行会要求你输入手机号收到的验证码和可能的两步验证密码。登录成功后会话文件listener_session.session会被保存下次运行无需再次验证。让这个账号加入你想要监听的Telegram群组。当群组中出现包含关键词的消息时你的alert_chat_id对应的账号就会收到警报。4. 进阶实现“人工实时监听”与增强功能基础的监听机器人已经能工作但“人工实时监听”意味着更高的交互性和可控性。我们可以将它升级为一个带有简单管理后台的机器人允许你动态管理关键词、查看触发记录、甚至临时暂停监听。4.1 构建一个简单的管理命令接口我们可以利用Pyrogram同样可以接收私聊消息的特性让监听机器人本身也响应你的命令。例如当你向监听账号私聊发送/keywords时它回复当前的关键词列表发送/add 漏洞时它动态添加关键词。实现思路在app.on_message处理函数中增加对私聊消息的过滤和命令解析。设计一套简单的命令如/keywords列出所有关键词。/add 关键词添加一个关键词。/remove 关键词移除一个关键词。/pause暂停监听。/resume恢复监听。/stats查看今日触发统计。将关键词的增删改查与文件keywords.txt或数据库实时同步。代码片段示例集成到主程序中from pyrogram import filters import shutil from datetime import datetime # 全局监听状态 LISTENING_ENABLED True app.on_message(filters.private filters.incoming) async def handle_private_command(client: Client, message: Message): 处理私聊命令 global LISTENING_ENABLED, keyword_lines, keyword_patterns if not message.text: return text message.text.strip() user_id message.from_user.id # 只允许特定的管理员用户你自己操作 if user_id ! config[alert_chat_id]: await message.reply_text(权限不足。) return if text /keywords: kw_list \n.join([f{i1}. {kw} for i, kw in enumerate(keyword_lines)]) reply f**当前关键词列表共{len(keyword_lines)}个:**\n{kw_list if kw_list else 暂无关键词} await message.reply_text(reply) elif text.startswith(/add ): new_kw text[5:].strip() if not new_kw: await message.reply_text(请输入要添加的关键词。) return if new_kw in keyword_lines: await message.reply_text(f关键词 {new_kw} 已存在。) else: keyword_lines.append(new_kw) # 更新正则列表 keyword_patterns.append(re.compile(rf\b{re.escape(new_kw)}\b, re.IGNORECASE)) # 写回文件 with open(keywords.txt, a, encodingutf-8) as f: f.write(f\n{new_kw}) await message.reply_text(f已添加关键词: {new_kw}) elif text.startswith(/remove ): kw_to_remove text[8:].strip() if kw_to_remove in keyword_lines: keyword_lines.remove(kw_to_remove) # 重建正则列表简单处理实际应更精细 keyword_patterns [] for line in keyword_lines: if line.startswith((?) or ( in line: keyword_patterns.append(re.compile(line)) else: keyword_patterns.append(re.compile(rf\b{re.escape(line)}\b, re.IGNORECASE)) # 重写文件 with open(keywords.txt, w, encodingutf-8) as f: f.write(\n.join(keyword_lines)) await message.reply_text(f已移除关键词: {kw_to_remove}) else: await message.reply_text(f未找到关键词: {kw_to_remove}) elif text /pause: if LISTENING_ENABLED: LISTENING_ENABLED False await message.reply_text(⏸️ 监听已暂停。) else: await message.reply_text(监听已是暂停状态。) elif text /resume: if not LISTENING_ENABLED: LISTENING_ENABLED True await message.reply_text(▶️ 监听已恢复。) else: await message.reply_text(监听已是运行状态。) elif text /stats: # 这里可以从数据库查询当天的触发次数 # 示例假设有个stats表记录触发时间 reply 今日触发统计功能待实现。 await message.reply_text(reply) else: # 非命令消息可以忽略或回复帮助信息 help_text ( **管理命令列表**\n /keywords - 查看当前关键词\n /add 关键词 - 添加关键词\n /remove 关键词 - 移除关键词\n /pause - 暂停监听\n /resume - 恢复监听\n /stats - 查看统计 ) await message.reply_text(help_text) # 修改群组消息处理函数增加状态检查 app.on_message(filters.group ~filters.edited) async def handle_new_message(client: Client, message: Message): if not LISTENING_ENABLED: return # ... 原有的处理逻辑 ...4.2 消息去重与频率限制在活跃的群组中同一话题可能在短时间内产生多条包含关键词的消息导致警报轰炸。我们需要引入一些控制机制基于内容的去重短时间内如1分钟内相同的触发消息内容或来自同一用户的相同关键词消息只报警一次。频率限制对同一个群组或同一个关键词设置最小报警间隔如每5分钟一次。聚合报警将短时间内如30秒内的多次触发合并成一条摘要消息上报。这可以通过在内存中维护一个带有时间戳的缓存字典来实现并在send_alert函数前进行检查。4.3 支持更多消息类型与富文本除了纯文本我们还可以增强对以下内容的监听文档/文件匹配文件名message.document.file_name。投票匹配投票问题message.poll.question。地理位置通常不需要匹配但可以记录。富文本提取对于包含链接的消息可以尝试使用requests和BeautifulSoup简单抓取链接的title标签将其纳入匹配范围这能抓到更多“标题党”信息。5. 部署、运维与高级避坑指南让机器人7x24小时稳定运行才是真正的挑战。本地电脑运行不可靠我们需要将它部署到服务器上。5.1 服务器部署方案VPS选择选择一家稳定的云服务商如阿里云、腾讯云国际、AWS Lightsail、DigitalOcean等最低配置1核1G即可胜任。重点选择网络对Telegram连接友好的区域例如新加坡、日本、德国等。进程守护使用systemd或supervisor来管理Python进程确保崩溃后自动重启。systemd服务文件示例(/etc/systemd/system/tg-listener.service)[Unit] DescriptionTG Keyword Listener Bot Afternetwork.target [Service] Typesimple Userubuntu WorkingDirectory/path/to/your/tg-keyword-listener EnvironmentPATH/path/to/your/venv/bin ExecStart/path/to/your/venv/bin/python /path/to/your/tg-keyword-listener/listener.py Restartalways RestartSec10 [Install] WantedBymulti-user.target使用sudo systemctl enable tg-listener设置开机自启sudo systemctl start tg-listener启动服务。日志管理使用Python的logging模块将日志输出到文件并定期轮转便于排查问题。备份定期备份你的.session文件、数据库和配置文件。.session文件丢失意味着需要重新登录验证。5.2 关键避坑点与经验分享“普号”的风控与封禁风险这是使用MTProto方案最大的风险。Telegram对自动化行为有检测。为了降低风险行为模拟不要高频、快速地发送请求或加入群组。加入群组间隔最好在几分钟以上。减少主动行为监听机器人最好只“听”不说不要在群里发送任何消息。使用老账号新注册的账号风险较高使用一个注册时间较长、有过正常聊天记录的“老号”作为监听号更安全。准备备用号重要监听任务考虑准备1-2个备用账号和api_id/api_hash在主号被封后能快速切换。“Flood Wait”错误Telegram API有严格的频率限制。如果短时间内请求太多会收到FloodWaitError告诉你需要等待若干秒。你的代码必须捕获这个异常并等待指定时间后重试。Pyrogram内部已经处理了部分重试但对于密集操作自己实现一个带退避的重试机制更稳妥。会话管理与多设备登录一个.session文件只能在一个地方登录。如果你在服务器部署就不能同时在手机上用同一个账号登录Telegram否则会被踢下线。确保监听账号专号专用。消息抓取遗漏网络波动或机器人重启可能导致消息遗漏。我们的“断点续传”机制记录last_message_id只能保证重启后不重复处理但无法抓取掉线期间的消息。对于绝对不能遗漏的消息可以考虑使用Telegram的“频道”或“私密群组”将重要群组的消息同时转发到一个只有监听机器人的频道/群组因为频道消息具有唯一的message_id且更稳定。定期同步在机器人启动时主动拉取最近一段时间如过去5分钟的历史消息进行检查注意API限制。关键词误报与漏报误报比如关键词“Java”会匹配到“JavaScript”。使用单词边界\b可以缓解。更精细的控制需要更复杂的NLP或规则。漏报中英文混合、特殊字符、繁体简体、同义词等。考虑使用分词库、同义词库或更模糊的正则匹配。安全与隐私法律与合规仅监听你拥有权限或已公开的群组信息。尊重他人隐私切勿用于非法用途。代码安全如前所述保护好api_id、api_hash和.session文件。不要在公开代码库中提交这些信息。性能优化当监听群组过多如上百个时单进程可能成为瓶颈。可以考虑异步I/O确保全程使用async/await这是Pyrogram的优势。分片处理如果群组太多可以考虑运行多个机器人实例每个实例负责一部分群组。消息队列将消息接收、关键词匹配、通知发送解耦通过Redis等消息队列传递提高吞吐量和可靠性。这个项目从简单的脚本到一个健壮的监控系统有很多可以深化的方向。最重要的是先跑起来解决你最迫切的信息获取痛点然后在实际运行中不断迭代优化。监听机器人就像你的数字哨兵把它部署好你就能在信息的海洋中牢牢抓住那些对你真正重要的浪花。本文还有配套的精品资源点击获取 基于Pyrogram的Telegram关键词监听机器人：从架构到部署的完整实践
简介这是一套面向TelegramTG生态运营者的关键词监听机器人源码适用于社群营销、竞品监控与实时用户触达等场景尤其适合需要隐蔽部署、多群协同监听的中高级开发者或运营人员。资源包共52个文件以36个PHP核心脚本为主含server.php、work.php、bootstrap.php等主控与工作模块辅以3个配置类JSON与HTML说明文档、Docker容器化部署文件docker-compose.yml、start/stop/reload脚本及环境配置文件.env.docker、config.php整体仅100KB轻量易部署。已有511人学习下载资源结构清晰包含完整运行流程从监听关键词配置监听关键词.json、服务启停控制到人工介入触发私聊的逻辑闭环并附带《关键词监听教程.docx》《资源说明必看.txt》等实操指引。读者可直接复用该普号方案实现无痕监听——多个TG账号协同工作主号不露面有效规避群内识别与封禁风险。1. 项目缘起从被动搜索到主动感知的信息需求在信息爆炸的时代我们常常陷入一个矛盾一方面我们害怕错过任何重要的消息另一方面我们又无法忍受信息洪流带来的噪音。尤其是在一些即时通讯社群中比如TelegramTG上的技术交流群、行业动态群、项目社区等有价值的信息往往转瞬即逝被淹没在闲聊、表情包和无关链接里。手动爬楼翻看聊天记录效率低下且容易遗漏。这种“信息焦虑”催生了一个非常具体的需求能否有一个“哨兵”24小时不间断地帮我盯着群聊一旦出现我关心的关键词就立刻通知我这就是“关键词监听机器人”诞生的背景。它不是一个复杂的AI而是一个精准的“触发器”。其核心逻辑非常简单模拟一个真实的Telegram用户即“普号”普通账号加入目标群组持续监听所有新消息将消息文本与预设的关键词列表进行匹配。一旦匹配成功就通过预设的渠道如另一个TG私聊窗口、Webhook、邮件等将这条消息的上下文发送者、时间、内容、直达链接实时推送给监听者。我最初开发这类机器人是为了跟踪几个加密货币项目和开发者社区的动态。这些社群的讨论质量很高但消息刷屏速度极快人工盯盘几乎不可能。通过部署监听机器人我成功抓取到了多次重要的项目更新预告、漏洞披露和空投信息其价值远超投入。今天我就把自己在开发和运维这类“TG关键词监听机器人”过程中的核心思路、技术选型、避坑经验以及“人工实时监听”的增强方案毫无保留地分享出来。无论你是想追踪竞品动态、关注行业热点、收集用户反馈还是单纯不想错过群里的重要通知这套方案都能为你提供一个高可用的自动化解决方案。2. 核心架构解析一个监听机器人的五脏六腑一个稳定可用的TG关键词监听机器人远不止几行“if text in keywords”的代码。它需要一套健壮的架构来应对网络波动、TG API限制、消息格式多样性以及自身的稳定运行。下面我们来拆解它的核心组件。2.1 消息获取层与Telegram API的对话这是机器人的“眼睛”和“耳朵”。我们需要一个客户端与Telegram服务器建立连接并接收消息。主流有两种选择MTProto协议客户端这是Telegram的原生协议效率高、功能全。Telethon和Pyrogram是Python中最流行的两个MTProto库。它们需要你用一个真实的手机号注册一个Telegram应用api_id和api_hash这个账号就是你的“普号”。监听机器人就是用这个账号的身份在群里活动。Telethon更底层异步支持好功能强大但学习曲线稍陡。Pyrogram封装更友好同步/异步都支持文档清晰对新手更友好。选择建议对于监听这种I/O密集型任务异步是更好的选择。Pyrogram的异步模式asyncio写起来更简洁社区活跃是我目前的首选。Bot API通过BotFather创建的机器人账号。它无法以“普通成员”身份主动加入大多数需要审核的群组除非被设为管理员且无法读取机器人加入之前的消息历史。这对于“潜伏监听”公开或私密群组来说是一个致命缺陷。因此对于需要以普通成员身份监听任意群组的场景必须使用MTProto协议的“普号”方案。关键配置与初始化使用Pyrogram初始化一个客户端核心是妥善保管你的api_id,api_hash和会话文件。会话文件.session保存了登录状态避免每次运行都需重新登录和可能触发二次验证。from pyrogram import Client app Client( my_listener_session, # 会话文件名 api_idYOUR_API_ID, api_hashYOUR_API_HASH, phone_numberYOUR_PHONE_NUMBER # 国际格式如 8613800138000 ) async def main(): async with app: # 此时客户端已登录并准备就绪 # 可以在这里获取已加入的群组列表或等待消息 pass重要安全提示api_id和api_hash等同于你的账号密码绝对不要泄露或上传至公开仓库。会话文件也包含敏感信息需同等保护。建议使用环境变量或配置文件来管理这些凭证。2.2 消息处理与过滤层大脑中的模式匹配这是机器人的“大脑”。它需要处理原始消息并判断是否触发警报。这一步有几个关键考量消息类型处理Telegram消息不只有文本。还有图片、视频、文档、链接等。我们的关键词匹配主要针对文本。对于媒体消息我们需要提取其caption说明文字或文件名进行匹配。对于链接可能需要提取URL本身或预览标题。关键词策略精确匹配 vs 模糊匹配“空投”是精确匹配。但有时我们需要模糊匹配比如包含“漏洞”或“bug”的消息。这可以通过正则表达式实现例如r(漏洞|bug|安全問題)。关键词列表管理关键词不应硬编码在代码里。最好使用一个外部配置文件如JSON、YAML或数据库来存储支持动态增删改查无需重启机器人。排除干扰可以设置“排除词”列表例如当消息同时包含“广告”和“福利”时即使触发了“福利”关键词也不报警以减少垃圾信息干扰。上下文获取一条孤立的触发消息可能信息量不足。优秀的监听器应该能捕获这条消息的上下文例如消息链接生成一个可以直接点击跳转到该条消息的t.me链接需要chat_id和message_id。前后消息有时触发词出现在对话中需要结合前后几条消息才能理解。可以尝试捕获触发消息前后各2-3条消息内容一并上报。发送者信息记录发送者的ID、用户名如果有和显示名称有助于判断消息来源的可靠性。2.3 通知与上报层发出警报的嘴巴这是机器人的“嘴巴”。当匹配成功后需要以某种形式通知你。常见方式有Telegram私聊将触发消息转发到你的另一个TG账号或同一个账号的“Saved Messages”。这是最直接、最实时的方式。机器人可以调用send_message方法将格式化好的报警信息发给你。Webhook将触发消息的结构化数据JSON格式POST到一个你指定的服务器地址。这种方式非常灵活你可以用服务器对接钉钉、飞书、企业微信、Slack甚至存入数据库或触发其他自动化流程。邮件/SMS作为备用通知渠道但实时性较差。数据库存储所有触发消息都存入数据库如SQLite、MySQL、PostgreSQL便于后续统计分析、回溯和去重。一个简单的Telegram私聊上报示例async def send_alert(app, chat_id, message): 发送警报到指定私聊 alert_text f 关键词触发警报\n\n alert_text f**群组**{message.chat.title}\n alert_text f**发送者**{message.from_user.mention if message.from_user else N/A}\n alert_text f**时间**{message.date}\n alert_text f**内容**\n{message.text or message.caption or 无文本内容}\n alert_text f**直达链接**{message.link} # 假设你的个人用户ID是 123456789 await app.send_message(chat_id123456789, textalert_text)2.4 持久化与状态管理层机器人的记忆机器人需要“记住”自己的状态比如已处理消息ID避免网络重连或重启后重复处理旧消息。通常记录最新处理的消息ID只处理比它ID更大的消息。关键词列表动态加载最新的关键词配置。运行日志记录监听状态、错误信息便于排查问题。这些状态可以存储在内存中重启丢失但更可靠的做法是使用轻量级数据库如SQLite或者简单的文件如JSON来保存。3. 从零搭建手把手实现你的第一个监听机器人理论讲完我们进入实战。我将以Pyrogram异步为核心搭配SQLite和配置文件构建一个基础但功能完整的监听机器人。3.1 环境准备与依赖安装首先确保你的Python环境是3.7或更高版本。创建一个新的项目目录并建立虚拟环境。mkdir tg-keyword-listener cd tg-keyword-listener python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖pip install pyrogram[tgcrypto] aiosqlitepyrogram[tgcrypto]Pyrogram库及其加密依赖。aiosqlite异步版本的SQLite驱动避免数据库操作阻塞事件循环。3.2 获取Telegram API凭证登录 my.telegram.org 。选择 “API development tools”。填写应用信息任意如MyListener提交后你会获得api_id和api_hash。请妥善保存。3.3 项目结构与核心代码项目目录结构如下tg-keyword-listener/ ├── config.yaml # 配置文件 ├── keywords.txt # 关键词列表每行一个 ├── listener.py # 主程序 ├── database.py # 数据库操作 └── utils.py # 工具函数1. 配置文件 (config.yaml)api_id: YOUR_API_ID api_hash: YOUR_API_HASH phone_number: 8613800138000 # 你的手机号 alert_chat_id: 123456789 # 接收警报的TG用户ID session_name: listener_session # 可选需要监听的群组ID列表负数如果为空则监听所有已加入的群 target_chats: [-1001234567890, -1009876543210]如何获取chat_id你可以让机器人先运行起来在目标群组发一条消息然后在代码中打印出message.chat.id这个id就是chat_id。普通群组的id是负数。2. 关键词文件 (keywords.txt)每行一个关键词或正则表达式。空投 Airdrop (?i)bug # (?i)表示不区分大小写 漏洞 重要通知 紧急3. 数据库模块 (database.py)负责记录最新处理的消息ID实现断点续传。import aiosqlite import asyncio DB_PATH listener.db async def init_db(): 初始化数据库 async with aiosqlite.connect(DB_PATH) as db: await db.execute( CREATE TABLE IF NOT EXISTS last_message ( chat_id INTEGER PRIMARY KEY, message_id INTEGER NOT NULL ) ) await db.commit() async def update_last_msg_id(chat_id: int, message_id: int): 更新某个聊天最后处理的消息ID async with aiosqlite.connect(DB_PATH) as db: await db.execute( INSERT OR REPLACE INTO last_message (chat_id, message_id) VALUES (?, ?) , (chat_id, message_id)) await db.commit() async def get_last_msg_id(chat_id: int) - int: 获取某个聊天最后处理的消息ID如果没有记录则返回0 async with aiosqlite.connect(DB_PATH) as db: async with db.execute(SELECT message_id FROM last_message WHERE chat_id ?, (chat_id,)) as cursor: row await cursor.fetchone() return row[0] if row else 04. 主监听程序 (listener.py)这是机器人的核心逻辑。import yaml import re import asyncio from pyrogram import Client, filters from pyrogram.types import Message from database import init_db, update_last_msg_id, get_last_msg_id # 加载配置 with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 加载关键词 with open(keywords.txt, r, encodingutf-8) as f: keyword_lines [line.strip() for line in f if line.strip() and not line.startswith(#)] # 编译正则表达式普通关键词转为精确匹配的正则 keyword_patterns [] for line in keyword_lines: if line.startswith((?) or ( in line: # 简单判断是否为复杂正则 keyword_patterns.append(re.compile(line)) else: # 将普通关键词转换为单词边界匹配避免匹配到单词的一部分 keyword_patterns.append(re.compile(rf\b{re.escape(line)}\b, re.IGNORECASE)) app Client( config[session_name], api_idconfig[api_id], api_hashconfig[api_hash], phone_numberconfig[phone_number] ) def contains_keyword(text: str) - bool: 检查文本是否包含任何关键词 if not text: return False for pattern in keyword_patterns: if pattern.search(text): return True return False async def send_alert(message: Message): 格式化并发送警报 chat_title message.chat.title or 私聊/频道 sender message.from_user sender_info f{sender.first_name} {sender.last_name or } ({sender.username}) if sender else 未知 # 获取消息文本或说明 content message.text or message.caption or [非文本消息] # 截断过长的内容 preview (content[:200] ...) if len(content) 200 else content alert_msg ( f** 关键词触发警报**\n\n f**来源群组**{chat_title}\n f**发送者**{sender_info}\n f**时间**{message.date}\n f**消息预览**\n{preview}\n\n f[点击查看原消息]({message.link}) ) try: await app.send_message(chat_idconfig[alert_chat_id], textalert_msg, disable_web_page_previewTrue) print(f[警报已发送] {chat_title} - {message.date}) except Exception as e: print(f发送警报失败: {e}) app.on_message(filters.group ~filters.edited) # 监听群组新消息忽略编辑消息 async def handle_new_message(client: Client, message: Message): 处理新消息 chat_id message.chat.id # 如果配置了特定群组则检查是否在目标列表中 target_chats config.get(target_chats) if target_chats and chat_id not in target_chats: return # 获取该聊天最后处理的消息ID实现断点续传 last_msg_id await get_last_msg_id(chat_id) if message.id last_msg_id: return # 已处理过的旧消息跳过 # 检查消息内容是否触发关键词 text_to_check message.text or message.caption or if contains_keyword(text_to_check): await send_alert(message) # 更新最后处理的消息ID await update_last_msg_id(chat_id, message.id) async def main(): await init_db() # 初始化数据库 print(数据库初始化完成。) print(正在启动监听机器人...) async with app: # 获取所有已加入的对话可以打印出来确认 # async for dialog in app.get_dialogs(): # print(f{dialog.chat.title or dialog.chat.first_name} - ID: {dialog.chat.id}) print(监听已启动。按 CtrlC 停止。) await asyncio.Event().wait() # 永久运行等待事件 if __name__ __main__: try: asyncio.run(main()) except KeyboardInterrupt: print(\n监听已停止。)3.4 运行与测试将config.yaml中的api_id,api_hash,phone_number,alert_chat_id替换为你的真实信息。在keywords.txt中添加你想监听的关键词。在终端运行python listener.py。首次运行会要求你输入手机号收到的验证码和可能的两步验证密码。登录成功后会话文件listener_session.session会被保存下次运行无需再次验证。让这个账号加入你想要监听的Telegram群组。当群组中出现包含关键词的消息时你的alert_chat_id对应的账号就会收到警报。4. 进阶实现“人工实时监听”与增强功能基础的监听机器人已经能工作但“人工实时监听”意味着更高的交互性和可控性。我们可以将它升级为一个带有简单管理后台的机器人允许你动态管理关键词、查看触发记录、甚至临时暂停监听。4.1 构建一个简单的管理命令接口我们可以利用Pyrogram同样可以接收私聊消息的特性让监听机器人本身也响应你的命令。例如当你向监听账号私聊发送/keywords时它回复当前的关键词列表发送/add 漏洞时它动态添加关键词。实现思路在app.on_message处理函数中增加对私聊消息的过滤和命令解析。设计一套简单的命令如/keywords列出所有关键词。/add 关键词添加一个关键词。/remove 关键词移除一个关键词。/pause暂停监听。/resume恢复监听。/stats查看今日触发统计。将关键词的增删改查与文件keywords.txt或数据库实时同步。代码片段示例集成到主程序中from pyrogram import filters import shutil from datetime import datetime # 全局监听状态 LISTENING_ENABLED True app.on_message(filters.private filters.incoming) async def handle_private_command(client: Client, message: Message): 处理私聊命令 global LISTENING_ENABLED, keyword_lines, keyword_patterns if not message.text: return text message.text.strip() user_id message.from_user.id # 只允许特定的管理员用户你自己操作 if user_id ! config[alert_chat_id]: await message.reply_text(权限不足。) return if text /keywords: kw_list \n.join([f{i1}. {kw} for i, kw in enumerate(keyword_lines)]) reply f**当前关键词列表共{len(keyword_lines)}个:**\n{kw_list if kw_list else 暂无关键词} await message.reply_text(reply) elif text.startswith(/add ): new_kw text[5:].strip() if not new_kw: await message.reply_text(请输入要添加的关键词。) return if new_kw in keyword_lines: await message.reply_text(f关键词 {new_kw} 已存在。) else: keyword_lines.append(new_kw) # 更新正则列表 keyword_patterns.append(re.compile(rf\b{re.escape(new_kw)}\b, re.IGNORECASE)) # 写回文件 with open(keywords.txt, a, encodingutf-8) as f: f.write(f\n{new_kw}) await message.reply_text(f已添加关键词: {new_kw}) elif text.startswith(/remove ): kw_to_remove text[8:].strip() if kw_to_remove in keyword_lines: keyword_lines.remove(kw_to_remove) # 重建正则列表简单处理实际应更精细 keyword_patterns [] for line in keyword_lines: if line.startswith((?) or ( in line: keyword_patterns.append(re.compile(line)) else: keyword_patterns.append(re.compile(rf\b{re.escape(line)}\b, re.IGNORECASE)) # 重写文件 with open(keywords.txt, w, encodingutf-8) as f: f.write(\n.join(keyword_lines)) await message.reply_text(f已移除关键词: {kw_to_remove}) else: await message.reply_text(f未找到关键词: {kw_to_remove}) elif text /pause: if LISTENING_ENABLED: LISTENING_ENABLED False await message.reply_text(⏸️ 监听已暂停。) else: await message.reply_text(监听已是暂停状态。) elif text /resume: if not LISTENING_ENABLED: LISTENING_ENABLED True await message.reply_text(▶️ 监听已恢复。) else: await message.reply_text(监听已是运行状态。) elif text /stats: # 这里可以从数据库查询当天的触发次数 # 示例假设有个stats表记录触发时间 reply 今日触发统计功能待实现。 await message.reply_text(reply) else: # 非命令消息可以忽略或回复帮助信息 help_text ( **管理命令列表**\n /keywords - 查看当前关键词\n /add 关键词 - 添加关键词\n /remove 关键词 - 移除关键词\n /pause - 暂停监听\n /resume - 恢复监听\n /stats - 查看统计 ) await message.reply_text(help_text) # 修改群组消息处理函数增加状态检查 app.on_message(filters.group ~filters.edited) async def handle_new_message(client: Client, message: Message): if not LISTENING_ENABLED: return # ... 原有的处理逻辑 ...4.2 消息去重与频率限制在活跃的群组中同一话题可能在短时间内产生多条包含关键词的消息导致警报轰炸。我们需要引入一些控制机制基于内容的去重短时间内如1分钟内相同的触发消息内容或来自同一用户的相同关键词消息只报警一次。频率限制对同一个群组或同一个关键词设置最小报警间隔如每5分钟一次。聚合报警将短时间内如30秒内的多次触发合并成一条摘要消息上报。这可以通过在内存中维护一个带有时间戳的缓存字典来实现并在send_alert函数前进行检查。4.3 支持更多消息类型与富文本除了纯文本我们还可以增强对以下内容的监听文档/文件匹配文件名message.document.file_name。投票匹配投票问题message.poll.question。地理位置通常不需要匹配但可以记录。富文本提取对于包含链接的消息可以尝试使用requests和BeautifulSoup简单抓取链接的title标签将其纳入匹配范围这能抓到更多“标题党”信息。5. 部署、运维与高级避坑指南让机器人7x24小时稳定运行才是真正的挑战。本地电脑运行不可靠我们需要将它部署到服务器上。5.1 服务器部署方案VPS选择选择一家稳定的云服务商如阿里云、腾讯云国际、AWS Lightsail、DigitalOcean等最低配置1核1G即可胜任。重点选择网络对Telegram连接友好的区域例如新加坡、日本、德国等。进程守护使用systemd或supervisor来管理Python进程确保崩溃后自动重启。systemd服务文件示例(/etc/systemd/system/tg-listener.service)[Unit] DescriptionTG Keyword Listener Bot Afternetwork.target [Service] Typesimple Userubuntu WorkingDirectory/path/to/your/tg-keyword-listener EnvironmentPATH/path/to/your/venv/bin ExecStart/path/to/your/venv/bin/python /path/to/your/tg-keyword-listener/listener.py Restartalways RestartSec10 [Install] WantedBymulti-user.target使用sudo systemctl enable tg-listener设置开机自启sudo systemctl start tg-listener启动服务。日志管理使用Python的logging模块将日志输出到文件并定期轮转便于排查问题。备份定期备份你的.session文件、数据库和配置文件。.session文件丢失意味着需要重新登录验证。5.2 关键避坑点与经验分享“普号”的风控与封禁风险这是使用MTProto方案最大的风险。Telegram对自动化行为有检测。为了降低风险行为模拟不要高频、快速地发送请求或加入群组。加入群组间隔最好在几分钟以上。减少主动行为监听机器人最好只“听”不说不要在群里发送任何消息。使用老账号新注册的账号风险较高使用一个注册时间较长、有过正常聊天记录的“老号”作为监听号更安全。准备备用号重要监听任务考虑准备1-2个备用账号和api_id/api_hash在主号被封后能快速切换。“Flood Wait”错误Telegram API有严格的频率限制。如果短时间内请求太多会收到FloodWaitError告诉你需要等待若干秒。你的代码必须捕获这个异常并等待指定时间后重试。Pyrogram内部已经处理了部分重试但对于密集操作自己实现一个带退避的重试机制更稳妥。会话管理与多设备登录一个.session文件只能在一个地方登录。如果你在服务器部署就不能同时在手机上用同一个账号登录Telegram否则会被踢下线。确保监听账号专号专用。消息抓取遗漏网络波动或机器人重启可能导致消息遗漏。我们的“断点续传”机制记录last_message_id只能保证重启后不重复处理但无法抓取掉线期间的消息。对于绝对不能遗漏的消息可以考虑使用Telegram的“频道”或“私密群组”将重要群组的消息同时转发到一个只有监听机器人的频道/群组因为频道消息具有唯一的message_id且更稳定。定期同步在机器人启动时主动拉取最近一段时间如过去5分钟的历史消息进行检查注意API限制。关键词误报与漏报误报比如关键词“Java”会匹配到“JavaScript”。使用单词边界\b可以缓解。更精细的控制需要更复杂的NLP或规则。漏报中英文混合、特殊字符、繁体简体、同义词等。考虑使用分词库、同义词库或更模糊的正则匹配。安全与隐私法律与合规仅监听你拥有权限或已公开的群组信息。尊重他人隐私切勿用于非法用途。代码安全如前所述保护好api_id、api_hash和.session文件。不要在公开代码库中提交这些信息。性能优化当监听群组过多如上百个时单进程可能成为瓶颈。可以考虑异步I/O确保全程使用async/await这是Pyrogram的优势。分片处理如果群组太多可以考虑运行多个机器人实例每个实例负责一部分群组。消息队列将消息接收、关键词匹配、通知发送解耦通过Redis等消息队列传递提高吞吐量和可靠性。这个项目从简单的脚本到一个健壮的监控系统有很多可以深化的方向。最重要的是先跑起来解决你最迫切的信息获取痛点然后在实际运行中不断迭代优化。监听机器人就像你的数字哨兵把它部署好你就能在信息的海洋中牢牢抓住那些对你真正重要的浪花。本文还有配套的精品资源点击获取