基于Telethon的Telegram群聊关键词监听机器人实现指南

简介这是一套面向TelegramTG运营者与自动化工具开发者的关键词监听机器人源码适用于私域引流、社群监控及实时营销响应等场景。资源采用PHP为主语言实现兼顾轻量部署与隐蔽性设计支持在多个电报群或频道中静默监听指定关键词并通过备用账号主动私聊触发用户实现无人值守式拉新与转化。压缩包共52个文件含36个核心PHP脚本如server.php、work.php、bootstrap.php、3个配置类文本config.php、.env.docker、监听关键词.json、3个HTML说明文档及Docker相关编排文件docker-compose.yml、start/stop/reload脚本等整体仅100KB结构紧凑、模块职责清晰。目前已有511人学习下载配套提供《关键词监听教程.docx》《资源说明必看.txt》及多份环境配置与运行指引涵盖部署流程、关键词规则配置、会话持久化机制与人工介入监听模式可直接用于测试与二次开发。1. 项目概述一个能“听懂”群聊的机器人最近在折腾一个挺有意思的东西一个专门用于Telegram简称TG群组的消息监听机器人。它的核心功能很直接像一个不知疲倦的“耳朵”24小时挂在指定的群组里实时监听所有聊天记录。但它不是简单地记录而是能“听懂”——它会根据你预先设定好的“关键词”进行过滤和匹配。一旦群聊中出现了你关心的词汇或短语它就会立刻向你发出警报甚至可以把包含关键词的原始消息、发言人、时间等信息推送给你实现一种“人工实时监听”的效果。这个项目的价值点在于它极大地解放了人力。无论是用于社区运营、舆情监控、竞品信息追踪还是特定话题的收集你都不需要再手动去“爬楼”翻看成千上万条聊天记录。机器人帮你完成了最枯燥的筛选工作让你能把精力集中在有价值的信息分析和决策上。而且它基于“普号”普通Telegram用户账号运行这意味着它的部署门槛相对较低不需要依赖特定的Bot API Token在某些场景下可能更具灵活性和隐蔽性。简单来说这就是一个信息过滤和提纯的工具。在信息爆炸的今天尤其是在Telegram这种信息流动极快的平台上拥有这样一个工具就相当于给自己配备了一个专属的情报官。接下来我会详细拆解这个项目的实现思路、核心代码、部署细节以及我踩过的一些坑希望能给有类似需求的朋友提供一个完整的参考方案。2. 核心思路与技术选型解析2.1 为什么选择“普号”而非官方Bot API在Telegram生态中主要有两种方式与平台交互一是通过官方Bot API使用Bot Token二是模拟普通用户客户端使用用户账号即“普号”。这个项目选择了后者这背后有几个关键的考量1. 监听权限与限制官方Bot在加入群组时通常需要被授予管理员权限或者该群组必须是公开群。对于许多私密群、需要邀请链接的群Bot可能无法加入。而使用普通用户账号只要该账号能被邀请进群就拥有了和普通成员一样的消息读取权限适用性更广。2. 消息获取的完整性某些群组可能禁用了“全体消息预览”等功能这可能会影响Bot对历史消息或特定类型消息如大文件的获取。用户客户端通常能获取到更接近真人视角的完整消息流。3. 灵活性与隐蔽性用户账号的行为模式更接近真人在某些对自动化工具敏感的环境下可能比一个明显的Bot账号更不容易被察觉或限制。当然这必须严格遵守平台规则避免滥用。技术实现权衡使用“普号”的代价是技术实现上更复杂。你需要处理用户登录包括二步验证、维护会话、模拟客户端行为等。而Bot API是HTTP-based的更简单稳定。因此这个选择是在“功能灵活性”和“开发复杂度”之间做的权衡。对于需要深入各种群组进行监听的需求“普号”方案往往是更优解。2.2 整体架构设计整个机器人的工作流可以抽象为以下几个核心模块它们共同协作完成从登录到报警的闭环用户账号登录与会话管理 ↓ 加入目标群组 ↓ 持续监听消息流 ↓ 实时消息过滤引擎 ↓ 关键词匹配与触发 ↓ 警报通知与消息推送1. 客户端模拟模块这是基石。我们需要一个库来模拟Telegram客户端。Telethon或Pyrogram是Python中两个最主流的选择。它们都基于MTProto协议能很好地模拟用户行为。本项目示例将使用Telethon因为它异步特性好文档丰富社区活跃。2. 消息监听与处理模块客户端成功运行后需要注册事件处理器。对于Telethon就是为events.NewMessage事件添加监听函数。这个函数会在机器人所在群组每收到一条新消息时被调用。3. 关键词管理与匹配模块这是大脑。我们需要一个高效、灵活的方式来管理关键词。不能只是简单的字符串相等判断要考虑模糊匹配、多关键词逻辑AND/OR、正则表达式支持等。关键词列表应该易于配置和动态更新。4. 警报推送模块这是输出。匹配成功后如何通知用户常见方式有向一个指定的“警报接收群”或私聊发送消息、通过Telegram Bot发送通知、甚至集成到钉钉/飞书/webhook。设计时需要兼顾即时性和可靠性。3. 核心代码实现与实操要点3.1 环境准备与依赖安装首先你需要一个Python环境建议3.8。然后安装核心库pip install telethon此外你可能还需要python-dotenv来管理配置re正则表达式Python内置用于高级匹配asyncioPython内置用于处理异步事件。接下来你需要一个Telegram账号作为机器人的“宿主”。非常重要请务必使用一个专门的小号不要用你的主账号因为自动化操作存在风险可能导致账号被暂时限制。准备好这个账号的手机号并确保它能正常接收登录验证码。3.2 核心监听引擎的实现下面是一个最精简但功能完整的监听机器人骨架代码我加入了大量注释来解释每一步import asyncio import re from telethon import TelegramClient, events from telethon.tl.types import PeerChannel, PeerChat import logging # 配置日志方便调试 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 1. 初始化客户端 # api_id 和 api_hash 需要从 https://my.telegram.org/apps 申请 # 它们代表你的“应用”而不是账号本身。 api_id 你的api_id api_hash ‘你的api_hash’ client TelegramClient(‘session_name’, api_id, api_hash) # session_name 用于保存登录会话 # 2. 定义关键词列表和匹配逻辑 # 这里使用一个简单的列表后续可以扩展为从文件或数据库读取 KEYWORDS [‘紧急通知’ ‘bug’ ‘漏洞’ ‘今晚发布’ r‘版本v\d\.\d’] # 支持正则表达式 # 匹配模式默认任何关键词出现即触发 def contains_keywords(text): “””检查文本是否包含任何关键词””” if not text: return False for kw in KEYWORDS: # 尝试将关键词作为正则表达式匹配 if re.search(kw, text, re.IGNORECASE): # 忽略大小写 return True return False # 3. 定义事件处理函数——机器人的“耳朵”和“大脑” client.on(events.NewMessage()) # 监听所有新消息事件 async def message_handler(event): # 只处理来自群组或频道的消息可选也可以监听私聊 chat await event.get_chat() if not (hasattr(chat, ‘broadcast’) or hasattr(chat, ‘megagroup’) or isinstance(event.peer_id, (PeerChannel, PeerChat))): return message_text event.message.text or event.message.message sender await event.get_sender() sender_name sender.username or sender.first_name or ‘未知用户’ chat_title chat.title logger.info(f“收到消息 - 来自 [{chat_title}] 的 {sender_name}: {message_text[:50]}...“) # 4. 关键词匹配判断 if message_text and contains_keywords(message_text): logger.warning(f“!!! 关键词匹配触发 !!!“) # 5. 警报推送逻辑 alert_msg f“ **关键词警报** \n\n” alert_msg f“**群组:** {chat_title}\n” alert_msg f“**发送者:** {sender_name} (ID: {sender.id})\n” alert_msg f“**时间:** {event.message.date}\n” alert_msg f“**匹配内容:** \n{message_text}\n\n” alert_msg f“原始消息链接: https://t.me/c/{event.chat_id}/{event.message.id}” # 注意这个链接格式需要根据chat_id调整 # 这里演示两种推送方式 # 方式一发送到指定的“警报接收群”或“个人收藏” alarm_chat_username ‘你的警报接收群用户名或ID‘ try: await client.send_message(alarm_chat_username, alert_msg) logger.info(f“警报已发送至 {alarm_chat_username}“) except Exception as e: logger.error(f“发送警报失败: {e}“) # 方式二打印到控制台开发调试用 print(“\n“ “”*50) print(alert_msg) print(““*50 “\n“) # 6. 主函数启动客户端 async def main(): # 启动客户端会提示输入手机号和验证码首次登录 await client.start() logger.info(“客户端启动成功开始监听...“) # 获取所有已加入的对话确认机器人所在的群组 async for dialog in client.iter_dialogs(): logger.info(f‘已加入对话: {dialog.name} (ID: {dialog.id})‘) # 保持运行直到手动停止 await client.run_until_disconnected() if __name__ ‘__main__‘: asyncio.run(main())关键点解析与注意事项api_id和api_hash这是Telegram官方用于识别第三方应用的身份凭证与你使用的Telegram账号是绑定的。你必须在 https://my.telegram.org/apps 用你的“宿主”账号登录并创建一个应用来获取它们。这不是Bot Token请勿混淆。session_name这个参数非常重要。Telethon会将登录后的会话包括授权密钥保存在一个名为session_name.session的文件中。只要这个文件存在且有效下次启动时就无需再次输入手机号和验证码实现了“免登录”运行。务必保管好这个.session文件它等同于你的账号权限事件过滤器client.on(events.NewMessage())这里注册了一个全局消息处理器。为了提升效率并避免干扰强烈建议添加过滤器。例如只监听特定群组# 假设你的目标群组ID是 -1001234567890 (通常为负数) TARGET_CHAT_ID -1001234567890 client.on(events.NewMessage(chatsTARGET_CHAT_ID))群组ID可以通过临时打印event.chat_id来获取。消息内容获取event.message.text获取纯文本。但Telegram消息可能是图片、文件、带格式文本等。event.message.message属性通常也能获取文本内容但更健壮的做法是使用event.raw_text。对于复杂消息可能需要额外处理。链接生成代码中尝试生成消息链接。注意https://t.me/c/...这种格式适用于频道和某些类型的超级群组且c/后面的数字需要是原始ID通常需要去掉-100前缀。对于普通群组链接格式不同。最可靠的方式是直接使用event.message.id和chat_id进行记录而非生成可点击链接。3.3 关键词匹配引擎的进阶优化上面的contains_keywords函数是最简单的实现。在实际应用中你可能需要更强大的匹配逻辑1. 支持多种匹配模式class KeywordMatcher: def __init__(self): self.keywords [] self.regex_patterns [] def load_keywords(self, keyword_list): for kw in keyword_list: if kw.startswith(‘regex:‘): # 处理正则表达式 pattern kw[6:] # 去掉 ‘regex:‘ 前缀 try: self.regex_patterns.append(re.compile(pattern, re.IGNORECASE)) except re.error as e: logger.error(f“无效的正则表达式 ‘{pattern}‘: {e}“) else: # 处理普通关键词支持简单通配符 self.keywords.append(kw.lower()) def match(self, text): if not text: return False text_lower text.lower() # 检查普通关键词 for kw in self.keywords: if kw in text_lower: return True # 检查正则表达式 for pattern in self.regex_patterns: if pattern.search(text): return True return False # 使用示例 matcher KeywordMatcher() matcher.load_keywords([‘紧急’ ‘重要通知’ ‘regex:版本v\\d\\.\\d\\.\\d’ ‘regex:(?i)urgent|asap‘])2. 关键词组与逻辑关系有时你需要监控“同时包含A和B”或“包含A但不包含B”的消息。可以设计一个规则引擎rules [ {‘name’: ‘版本发布’ ‘keywords’: [‘发布’ ‘版本’] ‘logic’: ‘AND‘}, # 必须同时出现 {‘name’: ‘问题反馈’ ‘keywords’: [‘bug’ ‘错误’ ‘无法运行’] ‘logic’: ‘OR‘}, # 出现任意一个 {‘name’: ‘非正式公告’ ‘keywords’: [‘内部’ ‘勿外传’] ‘exclude’: [‘公开’] ‘logic’: ‘AND‘}, # 包含“内部”且不包含“公开” ]3. 动态更新关键词你可以通过监听特定的管理命令来实时添加或删除关键词而无需重启机器人。例如在私聊中向机器人发送!add_keyword 新关键词。3.4 部署与长期运行方案在本地电脑上运行python bot.py只是测试。要让机器人7x24小时稳定运行你需要一个服务器。1. 服务器选择推荐使用海外的VPS虚拟专用服务器因为Telethon的连接对网络环境有一定要求。常见的供应商有DigitalOcean、Vultr、Linode等选择离你目标用户群较近的区域。2. 进程管理不能简单地用ssh运行然后关闭终端。你需要一个进程管理器来保证程序崩溃后能自动重启。Systemd (Linux推荐)创建一份 service 文件 (/etc/systemd/system/tg-monitor.service)[Unit] DescriptionTelegram Keyword Monitor Bot Afternetwork.target [Service] Typesimple Userubuntu WorkingDirectory/path/to/your/bot ExecStart/usr/bin/python3 /path/to/your/bot/bot.py Restartalways RestartSec10 StandardOutputsyslog StandardErrorsyslog EnvironmentPYTHONUNBUFFERED1 [Install] WantedBymulti-user.target然后使用sudo systemctl start tg-monitor启动sudo systemctl enable tg-monitor设置开机自启。Docker将你的代码和依赖打包成Docker镜像可以更好地隔离环境方便迁移。3. 日志管理将日志输出到文件并定期轮转方便后续排查问题。可以使用Python的logging.handlers.RotatingFileHandler。4. 会话文件安全.session文件是你的“账号护照”。在服务器上务必将其放在安全的目录并设置严格的文件权限如chmod 600 session_name.session。考虑定期备份此文件。4. 常见问题排查与实战经验在实际开发和运营过程中我遇到了不少问题这里总结一下最常见的几个及其解决方案。4.1 登录与授权问题问题运行代码后提示输入手机号但收不到验证码或提示FloodWaitError。排查手机号格式确保输入的是国际格式例如8613800138000。验证码接收有时验证码会通过Telegram应用本身发送如果该账号已在其他客户端登录请检查已登录设备上的消息。如果始终收不到可能是号码问题或触发了风控。FloodWaitError这意味着Telegram限制了你这个IP或账号的登录频率。千万不要反复重试错误信息里会包含需要等待的秒数如FloodWaitError: A wait of 86000 seconds is required。耐心等待指定时间后再试。首次登录或更换IP后较常见。二步验证如果账号设置了两步验证密码需要在代码中处理。Telethon的start()方法可以传入password参数或者通过交互式提示输入。经验首次登录建议在本地网络环境下进行成功生成.session文件后再将这个文件上传到服务器。这样可以避免服务器IP可能存在的风控问题。4.2 监听不到消息或反应迟钝问题机器人成功登录也加入了群组但收不到任何消息或者消息延迟很高。排查群组ID是否正确确认你监听的是正确的chat_id。chat_id对于普通群和超级群/频道是不同的。最准确的方法是在事件处理函数里打印event.chat_id和event.message.id。事件过滤器是否过窄检查client.on(events.NewMessage(chats...))中的chats参数确保它包含了所有你想监听的群组ID或用户名。网络连接问题服务器到Telegram服务器的网络可能不稳定。可以尝试在服务器上ping一下google.com测试基本连通性。对于海外VPS通常问题不大。客户端连接状态Telethon会自动处理重连但网络剧烈波动时可能断开。可以在代码中添加连接状态回调函数进行监控和日志记录。被群组禁言或限制虽然作为普通成员能看消息但如果账号被部分限制也可能影响接收。检查账号在群内的状态。经验在事件处理函数开头加一句日志记录所有收到的消息的chat_id和chat_title这是最直接的调试手段。4.3 关键词匹配不准确或漏报问题设置了关键词但该触发时没触发或者不该触发时乱触发。排查文本编码与清洗Telegram消息可能包含换行符、多余空格、或不可见字符。在匹配前可以对文本进行简单的清洗text ‘ ‘.join(event.raw_text.split())。大小写问题使用re.IGNORECASE标志进行忽略大小写的匹配如代码示例所示。关键词边界简单使用in判断会导致误匹配例如关键词“发布”会匹配到“本软件发布于2023年”。如果需要精确匹配单词应使用正则表达式的单词边界\b例如r‘\b发布\b‘。中文分词考虑对于复杂的中文语义匹配简单字符串匹配可能不够。例如“我喜欢苹果手机”和“我买了一个苹果”如果关键词是“苹果”两者都会匹配但语义不同。对于高要求场景可能需要引入简单的分词库如jieba进行更精确的匹配。消息类型你的代码可能只处理了event.message.text。但消息可能是图片带标题、投票、链接预览等。event.raw_text或event.message.message通常能获取到更多的文本内容。对于图片标题可以检查event.message.photo并获取其caption。4.4 账号安全与风控应对问题账号被限制无法发言、被要求验证手机号、甚至被封禁。排查与预防这是最重要的一部分滥用自动化工具严重违反Telegram服务条款可能导致永久封号。行为模拟真人化不要高频发送消息警报推送是合理的但切忌让这个账号在群内进行高频交互如自动回复每条消息。加入休眠间隔在消息处理函数中可以加入随机的、短暂的asyncio.sleep(random.uniform(0.5, 2))模拟真人阅读消息的延迟。避免同时加入过多群组短时间内用同一个账号加入几十个群组是明显的机器人行为。使用专用小号绝对不要用你的主账号。使用一个一次性或专门用于此目的的账号。准备备用方案如果主监听账号被封应有备用账号和快速切换的机制如切换.session文件。尊重隐私与合规仅在你有权访问的群组内运行监听机器人。在公开群组或频道中进行信息收集通常问题不大但在私人群组中务必谨慎并了解相关法律法规。我的实战心得运行这类监听机器人核心原则是“低调”和“只读”。让它像一个安静的观察者。我运行时间最久的一个实例超过一年其行为模式只有登录、同步消息、关键词匹配时向另一个固定的管理账号发送一条警报消息。除此之外不做任何其他操作。稳定性非常高。4.5 性能与可靠性优化当监听的群组非常多比如几十个消息量巨大时需要考虑性能。异步处理确保你的代码是纯异步的。不要在async函数内部调用阻塞式的同步IO操作如某些同步的数据库查询。如果必须用使用asyncio.to_thread将其放到线程池中执行。消息队列将消息接收和消息处理特别是关键词匹配、数据库写入、网络推送解耦。可以使用内存队列asyncio.Queue或外部消息队列如Redis。监听事件只负责快速将消息对象放入队列然后由单独的消费者任务进行处理避免因为某个处理步骤慢而阻塞消息接收。数据库写入优化如果需要将所有消息存库不要每条消息都单独提交一次事务。可以批量写入例如每10条或每秒集中提交一次。异常捕获与自我修复用try...except包裹关键逻辑特别是网络请求如发送警报。确保一个环节的失败不会导致整个事件循环崩溃。Telethon自身的网络错误通常会被库捕获并尝试重连但你的业务逻辑错误需要自己处理。这个项目从技术上看是MTProto协议应用、异步编程和简单规则引擎的结合。它的威力不在于技术有多高深而在于其解决实际问题的精准性。通过合理的配置和谨慎的操作它可以成为一个非常高效的信息感知工具。最后再次强调工具本身无对错请务必在法律和平台规则允许的范围内负责任地使用它。本文还有配套的精品资源点击获取