尧图网站建设 
简介这是一套面向TelegramTG群组运营者与私域流量操盘手的关键词监听机器人源码适用于需隐蔽监控多群消息、实现自动化响应与人工介入结合的营销或客服场景。资源基于PHP开发支持普通账号部署规避被识别为机器人风险兼顾隐蔽性与实时性。压缩包共52个文件含36个核心PHP脚本如server.php、work.php、bootstrap.php、3个配置类JSON与TXT文件含监听关键词.json、.env.docker、Docker容器化部署相关文件docker-compose.yml、start/stop/reload脚本及文档说明教程.docx、README.txt、免责声明等整体仅100KB轻量易部署。已有511人学习下载提供完整运行环境配置、关键词规则管理、消息路由逻辑与人工接管接口目录结构清晰runtime与src模块分离便于二次开发与调试。1. 项目概述一个能“听懂”群聊的自动化助手最近在折腾一个挺有意思的自动化项目核心目标很简单让一个机器人能7x24小时蹲守在指定的群组里自动“监听”所有聊天消息一旦发现预设的关键词被触发就立刻通知我。这听起来有点像舆情监控或者信息筛选但应用场景其实非常广泛。比如你是一个项目管理者需要第一时间知道群里谁提到了“BUG”或者“延期”你是一个电商运营想实时捕捉用户对某个“新品”的讨论或者你只是某个兴趣社群的活跃分子不想错过任何关于“周末活动”的召集信息。手动爬楼既低效又容易遗漏而这个机器人就是为了解决这个痛点而生的。这个项目的特别之处在于它强调了对“普号”普通账号的支持以及“人工实时监听”的能力。这意味着它并非依赖需要特殊申请、有诸多限制的官方Bot API而是可以直接使用一个普通的Telegram用户账号来运行模拟真人操作。这样做的好处是灵活性极高几乎可以接入任何类型的群组包括需要验证的私密群、频道评论等并且能够实现真正意义上的“实时”响应因为它的行为模式和一个真实用户别无二致。整个项目的核心就是一套能够稳定运行、精准过滤、及时通知的自动化脚本源码。2. 核心方案选型与技术栈解析2.1 为什么选择“普号”方案而非官方Bot在Telegram生态中获取信息主要有两种途径官方Bot API和使用用户账号的MTProto协议。官方Bot功能强大、稳定但限制也很明显它无法读取普通群组的历史消息除非被设为管理员无法在需要管理员审核才能发言的群组中自动通过验证更无法访问一些设置为私密的超级群组。而我们的需求恰恰经常涉及这些场景。因此采用基于MTProto协议的“普号”方案成为了必然选择。这相当于我们编写了一个脚本用它来登录一个真实的Telegram账号并模拟用户的所有操作登录、接收消息、解析消息、关键词匹配、发送通知。这个方案的灵魂在于一个名为Telethon的Python异步库。Telethon是对Telegram MTProto协议的完整封装它允许我们以编程方式控制一个用户账户功能全面且社区活跃是完成此类任务的绝佳工具。注意使用用户账号进行自动化操作需要严格遵守Telegram的服务条款。过度频繁的请求、发送垃圾信息或滥用行为可能导致账号被限制甚至封禁。因此我们的代码必须设计得“像人一样”加入合理的延迟和错误处理。2.2 整体架构与数据流设计整个机器人的运行逻辑可以清晰地分为几个模块形成一个高效的数据处理流水线客户端初始化与登录模块负责加载配置API ID、API Hash、手机号完成账号的登录验证包括处理二次验证码。这是机器人的启动入口。消息监听与获取模块持续监听指定对话群组或频道的新消息。这里的关键是异步事件驱动机器人不会阻塞等待而是有消息到来时才触发处理函数。消息过滤与解析模块这是核心逻辑层。收到原始消息后需要对其进行清洗去除多余空格、换行、解析获取发送者、消息内容、时间戳等然后与预设的关键词列表进行匹配。匹配策略可以是精确匹配、模糊匹配或正则表达式匹配。通知触发与发送模块一旦匹配成功就需要将结果通知给“监听者”。通知方式可以多样化向另一个指定的私聊窗口比如你自己的账号发送一条格式化好的摘要消息或者将信息写入本地日志文件、数据库更高级的可以对接Webhook推送到钉钉、飞书或Slack。配置与状态管理模块管理关键词列表、监听群组列表、通知对象等配置信息。这些配置最好设计成可以动态热加载这样在机器人运行期间也能修改关键词而无需重启。数据流大致如下新消息事件 - 消息内容文本 - 文本清洗与解析 - 关键词匹配引擎 - 匹配成功 - 格式化通知消息 - 通过Telethon发送给指定接收者。3. 环境准备与核心依赖部署3.1 基础Python环境搭建首先你需要一个Python运行环境。推荐使用Python 3.8或更高版本因为Telethon对异步特性的支持在新版本中更好。使用虚拟环境是一个好习惯可以避免包依赖冲突。# 创建项目目录并进入 mkdir tg_keyword_monitor cd tg_keyword_monitor # 创建虚拟环境以venv为例 python3 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate # 升级pip pip install --upgrade pip3.2 关键库的安装与说明接下来安装核心依赖。最主要的库就是Telethon。此外我们可能还需要python-dotenv来管理敏感配置loguru或标准库logging来记录日志方便排查问题。pip install telethon python-dotenv loguruTelethon这是项目的基石。它不仅处理了底层的MTProto协议通信还提供了非常友好的高层API比如client.on(events.NewMessage)这样的事件监听器让我们可以用很少的代码实现强大的功能。python-dotenv我们的Telegram API凭证api_id,api_hash以及手机号是敏感信息不应该硬编码在脚本里。这个库帮助我们从.env文件中加载这些环境变量。Loguru一个比标准库logging更友好、功能更强大的日志库。它支持彩色输出、结构化日志、异常捕获等能极大提升开发调试效率。3.3 获取Telegram API凭证这是使用“普号”方案的前提。你需要访问 my.telegram.org 并用你的手机号登录。点击 “API development tools”。填写应用信息应用名称、简称、平台等这些可以随意填写仅用于标识。提交后你将获得至关重要的两样东西api_id和api_hash。请务必妥善保管它们是你账号的编程接入凭证。实操心得建议为这个监控项目单独注册一个Telegram账号而不是使用你的主号。这样既能隔离风险也方便管理。注册时最好使用一个长期有效的邮箱并记好密码。4. 核心代码实现与分步拆解4.1 项目结构与配置文件我们先来搭建一个清晰的项目结构tg_keyword_monitor/ ├── .env # 存储敏感配置需加入.gitignore ├── config.py # 应用配置关键词、群组ID等 ├── monitor.py # 主监控逻辑脚本 ├── utils/ # 工具函数目录 │ └── notifier.py # 通知发送器 └── logs/ # 日志目录首先创建.env文件填入你的凭证# .env API_ID你的api_id API_HASH你的api_hash PHONE_NUMBER8612345678901 # 你的Telegram账号带国家代码然后是config.py这里存放非敏感的、可能经常变动的配置# config.py # 监听的目标群组/频道用户名或ID可以是列表监听多个源 TARGET_ENTITIES [ “group_username”, # 公开群的用户名 -1001234567890, # 私有群/频道的数字ID通常为负数 ] # 需要监控的关键词列表支持正则表达式 KEYWORDS [ “BUG”, “故障”, “紧急”, “r’上线|发布’”, # 正则示例匹配“上线”或“发布” “今晚开会”, ] # 通知接收者的用户名或ID通常是你的个人账号 NOTIFY_TO “your_private_username”4.2 客户端初始化与安全登录在monitor.py中我们编写初始化和登录逻辑。Telethon的TelegramClient会处理会话session文件首次登录需要输入验证码之后就可以自动登录。# monitor.py import asyncio import re from datetime import datetime from telethon import TelegramClient, events from telethon.tl.types import Message from dotenv import load_dotenv import os import sys sys.path.append(‘.’) from config import TARGET_ENTITIES, KEYWORDS, NOTIFY_TO # 加载环境变量 load_dotenv() api_id int(os.getenv(‘API_ID’)) api_hash os.getenv(‘API_HASH’) phone os.getenv(‘PHONE_NUMBER’) # 创建客户端session文件名为‘monitor_session’ client TelegramClient(‘monitor_session’, api_id, api_hash) async def main(): # 连接并登录 await client.start(phonephone) print(f“[{datetime.now()}] 登录成功账号{await client.get_me().username}”) # 这里将添加事件监听器 # ... 监听代码将在下一节添加 # 保持客户端运行持续监听 print(f“[{datetime.now()}] 开始监听目标{TARGET_ENTITIES}”) await client.run_until_disconnected() # 编译关键词正则提升匹配效率 compiled_keywords [] for kw in KEYWORDS: if kw.startswith(‘r’’’) and kw.endswith(‘’’’): # 简单判断是否为原始字符串表示的正则 pattern kw[2:-1] # 去除 r’’ 引号 compiled_keywords.append(re.compile(pattern)) else: # 对于普通字符串我们进行子串匹配可以忽略大小写 compiled_keywords.append(re.compile(re.escape(kw), re.IGNORECASE))注意事项首次运行client.start()时脚本会在终端提示你输入手机收到的验证码。如果账号开启了二次密码2FA还需要输入密码。登录成功后会生成一个.session文件以后运行就无需再次验证。务必保管好这个.session文件它等同于登录凭证。4.3 消息监听与关键词匹配引擎这是最核心的部分。我们使用Telethon的事件装饰器来监听新消息。为了精准控制监听范围我们需要将事件限定在TARGET_ENTITIES指定的对话中。# 接上面的 main 函数内部 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): “”“处理新消息进行关键词匹配”“” message: Message event.message sender await event.get_sender() chat await event.get_chat() # 获取干净的纯文本内容忽略媒体、链接预览等 text message.text or message.raw_text or “” if not text: return # 忽略纯媒体消息 # 简单的文本清洗 cleaned_text text.strip() # 关键词匹配检查 matched_keywords [] for pattern in compiled_keywords: if pattern.search(cleaned_text): # 如果是正则对象找到匹配的原始词 if isinstance(pattern, re.Pattern): # 这里简单记录模式串实际可更精细 matched_keywords.append(pattern.pattern) else: matched_keywords.append(pattern.pattern) if not matched_keywords: return # 没有匹配到任何关键词忽略此消息 # 匹配成功准备通知信息 print(f“[{datetime.now()}] 检测到关键词来源{chat.title} 发送者{sender.username or sender.id} 内容摘要{cleaned_text[:100]}...”) # 构建通知消息 notify_msg f“ **关键词触发警报**\n\n” notify_msg f“**群组**{chat.title}\n” notify_msg f“**发送者**{sender.first_name}” if sender.username: notify_msg f“ ({sender.username})” notify_msg f“\n**触发词**{‘ ’.join(matched_keywords)}\n” notify_msg f“**消息时间**{message.date.strftime(‘%Y-%m-%d %H:%M:%S’)}\n” notify_msg f“**消息内容**\n{cleaned_text}\n” notify_msg f“\n[点击查看上下文](https://t.me/c/{chat.id}/{message.id})” # 发送通知这里先简单打印下一节实现真实发送 print(notify_msg) # await send_notification(notify_msg) # 后续实现 # 然后运行主函数 if __name__ ‘__main__’: with client: client.loop.run_until_complete(main())这段代码做了几件关键事client.on(events.NewMessage(chats...))将处理函数限定在特定聊天列表避免监听所有对话造成干扰和性能浪费。从事件中提取消息、发送者、聊天对象等详细信息。对消息文本进行预处理。遍历编译好的关键词正则列表进行匹配。这里采用re.IGNORECASE实现不区分大小写的匹配更符合实际需求。匹配成功后格式化一条包含所有关键信息的通知消息。其中构造了Telegram的深链接t.me/c/chat.id/message.id点击可以直接跳转到原消息上下文这是“人工实时监听”中“人工介入”的关键极大提升了处理效率。4.4 实现多通道通知系统仅仅打印日志是不够的。我们需要将警报可靠地推送到指定位置。我们在utils/notifier.py中实现一个通知器它可以支持多种后端。# utils/notifier.py import asyncio from telethon import TelegramClient import logging logger logging.getLogger(__name__) class NotificationManager: def __init__(self, tg_client: TelegramClient, notify_to): self.client tg_client self.notify_to notify_to # 可以是一个用户名/ID也可以是一个列表 self._notification_queue asyncio.Queue() # 可以在此初始化其他通知渠道如邮件、Webhook等 async def send_telegram_alert(self, message: str): “”“通过Telegram发送警报”“” try: # 确保接收者是列表形式 targets self.notify_to if isinstance(self.notify_to, list) else [self.notify_to] for target in targets: await self.client.send_message(target, message, parse_mode‘markdown’, link_previewFalse) logger.info(f“Telegram警报已发送至 {target}”) await asyncio.sleep(0.5) # 发送间隔避免触发风控 except Exception as e: logger.error(f“发送Telegram警报失败{e}”) async def put_notification(self, message: str): “”“将通知放入队列由后台任务处理”“” await self._notification_queue.put(message) async def _worker(self): “”“后台工作线程从队列中取出通知并发送”“” while True: message await self._notification_queue.get() await self.send_telegram_alert(message) self._notification_queue.task_done() async def start(self): “”“启动后台通知任务”“” asyncio.create_task(self._worker()) logger.info(“通知管理器已启动”)然后在主程序monitor.py中集成这个通知管理器# 在 monitor.py 顶部导入 from utils.notifier import NotificationManager # 在 main 函数开始登录成功后初始化通知管理器 async def main(): await client.start(phonephone) print(f“登录成功...”) # 初始化通知管理器 notify_mgr NotificationManager(client, NOTIFY_TO) await notify_mgr.start() # 启动后台工作线程 # 修改 keyword_handler 函数中的通知部分 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): # ... 前面的匹配逻辑不变 ... if matched_keywords: # ... 构建 notify_msg ... # 将通知放入队列非阻塞式发送 await notify_mgr.put_notification(notify_msg)实操心得使用消息队列 (asyncio.Queue) 来处理通知发送是一个好习惯。它解耦了消息接收和发送即使网络波动导致发送暂时失败或者短时间内触发大量关键词消息也不会丢失而是排队等待发送。同时在send_message中加入短暂延迟 (await asyncio.sleep(0.5)) 是模拟人类操作、避免被Telegram风控系统误判为垃圾消息的关键技巧。5. 高级功能与优化实践5.1 动态更新关键词与群组列表让机器人重启才能修改配置太不灵活。我们可以实现一个简单的命令监听让管理员通过向机器人账号发送私聊命令来动态更新配置。# 在 monitor.py 中添加新的时间处理器 client.on(events.NewMessage(from_usersNOTIFY_TO)) # 只接收来自管理员的消息 async def admin_command_handler(event): “”“处理管理命令”“” text event.message.text.strip() if text.startswith(‘/add_keyword’): _, new_keyword text.split(maxsplit1) if new_keyword: KEYWORDS.append(new_keyword) # 重新编译正则列表需将compiled_keywords设为全局或可修改对象 # ... 更新逻辑 ... await event.reply(f“✅ 已添加关键词{new_keyword}”) elif text.startswith(‘/list_keywords’): await event.reply(f“当前监控关键词\n” “\n”.join(KEYWORDS)) # 可以添加更多命令如 /add_chat, /remove_keyword 等5.2 消息去重与频率限制在活跃的群组中同一话题可能在短时间内被反复提及导致警报轰炸。我们需要引入简单的去重和频率限制机制。# 简易的去重缓存可使用redis等持久化存储以应对重启 message_cache set() CACHE_TTL 300 # 5分钟内相同内容不重复警报 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): # ... 获取 cleaned_text ... # 内容去重检查 cache_key f“{chat.id}:{cleaned_text[:50]}” # 用聊天ID和内容前50字符做键 if cache_key in message_cache: return # 匹配关键词... if matched_keywords: # 发送通知... # 加入缓存并设置异步任务在TTL后移除 message_cache.add(cache_key) asyncio.create_task(remove_from_cache(cache_key)) async def remove_from_cache(key): await asyncio.sleep(CACHE_TTL) message_cache.discard(key)5.3 支持媒体消息与复杂内容解析有时关键词可能出现在图片的OCR文字、文档名称或链接预览中。Telethon提供了访问这些属性的能力。async def extract_full_text(event): “”“尝试从消息中提取尽可能多的文本内容”“” message event.message text message.text or message.raw_text or “” # 检查消息是否包含文档/图片/视频并获取其文件名/标题 if message.document: text f“\n[文件] {message.document.attributes[0].file_name if message.document.attributes else ‘未知文件’}” if message.photo: text “\n[图片]” # 检查是否有链接预览web_preview if message.web_preview: text f“\n[链接预览] {message.web_preview.title or ‘’} - {message.web_preview.description or ‘’}” return text.strip()在keyword_handler中调用extract_full_text(event)来获取更全面的文本内容进行匹配可以显著提高监控覆盖率。6. 部署、运维与常见问题排查6.1 服务器部署与后台运行开发完成后你需要让脚本在服务器上7x24小时运行。使用systemd或supervisor是可靠的选择。使用 systemd 服务示例创建文件/etc/systemd/system/tg-monitor.service[Unit] DescriptionTelegram Keyword Monitor Bot Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/your/tg_keyword_monitor Environment“PATH/path/to/your/venv/bin” ExecStart/path/to/your/venv/bin/python monitor.py Restartalways RestartSec10 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable tg-monitor sudo systemctl start tg-monitor sudo systemctl status tg-monitor # 查看状态6.2 常见问题与解决方案速查表问题现象可能原因排查步骤与解决方案启动时报ApiIdInvalidErrorapi_id或api_hash错误1. 检查.env文件中的值是否正确前后有无空格。2. 前往 my.telegram.org 重新创建应用并获取新凭证。登录时收不到验证码账号被限制或手机号格式错误1. 确认手机号格式为国际格式如8613800138000。2. 尝试在官方App上先登录一次解除可能的风控。3. 如果使用86号码尝试切换成86 13800138000格式有空格。运行一段时间后失去连接网络波动或Telegram服务器中断1.Telethon内置了重连机制通常会自动恢复。检查日志确认。2. 确保服务器网络稳定特别是国际出口。3. 在TelegramClient初始化时调整连接参数如设置代理。无法监听某些私有群组账号未加入该群组或没有访问权限1. 确保运行机器人的账号已经加入了目标群组。2. 对于“禁止转发消息”的群组机器人依然可以读取消息内容并触发关键词但无法生成消息链接。触发关键词后没有通知通知接收者设置错误或发送失败1. 检查NOTIFY_TO配置是否正确确保机器人账号已与接收者账号建立私聊发过一条消息即可。2. 查看程序日志确认send_message是否报错如权限不足、对方已屏蔽等。3. 检查通知队列是否正常工作。误报或漏报率高关键词匹配策略太宽泛或太严格1. 优化关键词列表使用更精确的词语或正则表达式。2. 考虑在匹配前对文本进行更细致的清洗如去除标点、统一大小写。3. 实现上下文匹配例如要求关键词“故障”附近出现“系统”一词时才报警。账号被限制或封禁行为过于频繁被判定为滥用1.最重要在所有send_message和get_entity等可能产生频繁请求的操作前后增加随机延迟 (await asyncio.sleep(random.uniform(1, 3)))。2. 避免在短时间内加入大量群组或发送大量消息。3. 如果被封尝试通过Telegram官方渠道申诉解封。6.3 性能优化与扩展思路当监控的群组非常多、消息量巨大时需要考虑性能优化异步IO充分利用确保所有网络请求如发送通知都是异步的避免阻塞主监听循环。关键词匹配算法优化对于海量关键词可以考虑使用Aho-Corasick算法等多模式匹配算法来替代简单的循环正则匹配效率可提升数个量级。数据库持久化将匹配到的消息、发送的通知记录到SQLite或PostgreSQL中便于后续查询、分析和去重。分布式监听如果单个账号监听上千个群组有压力可以考虑用多个账号多个session分担监听任务并通过一个中心服务聚合通知。规则引擎超越简单关键词实现更复杂的规则如“关键词A与关键词B在10分钟内出现”、“用户X发言中包含关键词C”等这需要引入一个规则引擎和状态管理。这个“关键词监听机器人”项目从技术上看是MTProto协议应用、异步编程和简单规则引擎的结合从实用角度看它是一个高度定制化的信息过滤与提纯工具。它的价值不在于技术有多高深而在于它能切实地将人从繁杂的实时信息流中解放出来精准捕捉到那些真正重要的信号。你可以根据自己的需求轻松地修改关键词列表、调整通知格式、甚至将报警对接到你日常使用的办公软件中让它成为你数字生活里一个沉默而可靠的哨兵。本文还有配套的精品资源点击获取 基于Telethon的Telegram群聊关键词实时监控机器人开发指南
简介这是一套面向TelegramTG群组运营者与私域流量操盘手的关键词监听机器人源码适用于需隐蔽监控多群消息、实现自动化响应与人工介入结合的营销或客服场景。资源基于PHP开发支持普通账号部署规避被识别为机器人风险兼顾隐蔽性与实时性。压缩包共52个文件含36个核心PHP脚本如server.php、work.php、bootstrap.php、3个配置类JSON与TXT文件含监听关键词.json、.env.docker、Docker容器化部署相关文件docker-compose.yml、start/stop/reload脚本及文档说明教程.docx、README.txt、免责声明等整体仅100KB轻量易部署。已有511人学习下载提供完整运行环境配置、关键词规则管理、消息路由逻辑与人工接管接口目录结构清晰runtime与src模块分离便于二次开发与调试。1. 项目概述一个能“听懂”群聊的自动化助手最近在折腾一个挺有意思的自动化项目核心目标很简单让一个机器人能7x24小时蹲守在指定的群组里自动“监听”所有聊天消息一旦发现预设的关键词被触发就立刻通知我。这听起来有点像舆情监控或者信息筛选但应用场景其实非常广泛。比如你是一个项目管理者需要第一时间知道群里谁提到了“BUG”或者“延期”你是一个电商运营想实时捕捉用户对某个“新品”的讨论或者你只是某个兴趣社群的活跃分子不想错过任何关于“周末活动”的召集信息。手动爬楼既低效又容易遗漏而这个机器人就是为了解决这个痛点而生的。这个项目的特别之处在于它强调了对“普号”普通账号的支持以及“人工实时监听”的能力。这意味着它并非依赖需要特殊申请、有诸多限制的官方Bot API而是可以直接使用一个普通的Telegram用户账号来运行模拟真人操作。这样做的好处是灵活性极高几乎可以接入任何类型的群组包括需要验证的私密群、频道评论等并且能够实现真正意义上的“实时”响应因为它的行为模式和一个真实用户别无二致。整个项目的核心就是一套能够稳定运行、精准过滤、及时通知的自动化脚本源码。2. 核心方案选型与技术栈解析2.1 为什么选择“普号”方案而非官方Bot在Telegram生态中获取信息主要有两种途径官方Bot API和使用用户账号的MTProto协议。官方Bot功能强大、稳定但限制也很明显它无法读取普通群组的历史消息除非被设为管理员无法在需要管理员审核才能发言的群组中自动通过验证更无法访问一些设置为私密的超级群组。而我们的需求恰恰经常涉及这些场景。因此采用基于MTProto协议的“普号”方案成为了必然选择。这相当于我们编写了一个脚本用它来登录一个真实的Telegram账号并模拟用户的所有操作登录、接收消息、解析消息、关键词匹配、发送通知。这个方案的灵魂在于一个名为Telethon的Python异步库。Telethon是对Telegram MTProto协议的完整封装它允许我们以编程方式控制一个用户账户功能全面且社区活跃是完成此类任务的绝佳工具。注意使用用户账号进行自动化操作需要严格遵守Telegram的服务条款。过度频繁的请求、发送垃圾信息或滥用行为可能导致账号被限制甚至封禁。因此我们的代码必须设计得“像人一样”加入合理的延迟和错误处理。2.2 整体架构与数据流设计整个机器人的运行逻辑可以清晰地分为几个模块形成一个高效的数据处理流水线客户端初始化与登录模块负责加载配置API ID、API Hash、手机号完成账号的登录验证包括处理二次验证码。这是机器人的启动入口。消息监听与获取模块持续监听指定对话群组或频道的新消息。这里的关键是异步事件驱动机器人不会阻塞等待而是有消息到来时才触发处理函数。消息过滤与解析模块这是核心逻辑层。收到原始消息后需要对其进行清洗去除多余空格、换行、解析获取发送者、消息内容、时间戳等然后与预设的关键词列表进行匹配。匹配策略可以是精确匹配、模糊匹配或正则表达式匹配。通知触发与发送模块一旦匹配成功就需要将结果通知给“监听者”。通知方式可以多样化向另一个指定的私聊窗口比如你自己的账号发送一条格式化好的摘要消息或者将信息写入本地日志文件、数据库更高级的可以对接Webhook推送到钉钉、飞书或Slack。配置与状态管理模块管理关键词列表、监听群组列表、通知对象等配置信息。这些配置最好设计成可以动态热加载这样在机器人运行期间也能修改关键词而无需重启。数据流大致如下新消息事件 - 消息内容文本 - 文本清洗与解析 - 关键词匹配引擎 - 匹配成功 - 格式化通知消息 - 通过Telethon发送给指定接收者。3. 环境准备与核心依赖部署3.1 基础Python环境搭建首先你需要一个Python运行环境。推荐使用Python 3.8或更高版本因为Telethon对异步特性的支持在新版本中更好。使用虚拟环境是一个好习惯可以避免包依赖冲突。# 创建项目目录并进入 mkdir tg_keyword_monitor cd tg_keyword_monitor # 创建虚拟环境以venv为例 python3 -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate # 升级pip pip install --upgrade pip3.2 关键库的安装与说明接下来安装核心依赖。最主要的库就是Telethon。此外我们可能还需要python-dotenv来管理敏感配置loguru或标准库logging来记录日志方便排查问题。pip install telethon python-dotenv loguruTelethon这是项目的基石。它不仅处理了底层的MTProto协议通信还提供了非常友好的高层API比如client.on(events.NewMessage)这样的事件监听器让我们可以用很少的代码实现强大的功能。python-dotenv我们的Telegram API凭证api_id,api_hash以及手机号是敏感信息不应该硬编码在脚本里。这个库帮助我们从.env文件中加载这些环境变量。Loguru一个比标准库logging更友好、功能更强大的日志库。它支持彩色输出、结构化日志、异常捕获等能极大提升开发调试效率。3.3 获取Telegram API凭证这是使用“普号”方案的前提。你需要访问 my.telegram.org 并用你的手机号登录。点击 “API development tools”。填写应用信息应用名称、简称、平台等这些可以随意填写仅用于标识。提交后你将获得至关重要的两样东西api_id和api_hash。请务必妥善保管它们是你账号的编程接入凭证。实操心得建议为这个监控项目单独注册一个Telegram账号而不是使用你的主号。这样既能隔离风险也方便管理。注册时最好使用一个长期有效的邮箱并记好密码。4. 核心代码实现与分步拆解4.1 项目结构与配置文件我们先来搭建一个清晰的项目结构tg_keyword_monitor/ ├── .env # 存储敏感配置需加入.gitignore ├── config.py # 应用配置关键词、群组ID等 ├── monitor.py # 主监控逻辑脚本 ├── utils/ # 工具函数目录 │ └── notifier.py # 通知发送器 └── logs/ # 日志目录首先创建.env文件填入你的凭证# .env API_ID你的api_id API_HASH你的api_hash PHONE_NUMBER8612345678901 # 你的Telegram账号带国家代码然后是config.py这里存放非敏感的、可能经常变动的配置# config.py # 监听的目标群组/频道用户名或ID可以是列表监听多个源 TARGET_ENTITIES [ “group_username”, # 公开群的用户名 -1001234567890, # 私有群/频道的数字ID通常为负数 ] # 需要监控的关键词列表支持正则表达式 KEYWORDS [ “BUG”, “故障”, “紧急”, “r’上线|发布’”, # 正则示例匹配“上线”或“发布” “今晚开会”, ] # 通知接收者的用户名或ID通常是你的个人账号 NOTIFY_TO “your_private_username”4.2 客户端初始化与安全登录在monitor.py中我们编写初始化和登录逻辑。Telethon的TelegramClient会处理会话session文件首次登录需要输入验证码之后就可以自动登录。# monitor.py import asyncio import re from datetime import datetime from telethon import TelegramClient, events from telethon.tl.types import Message from dotenv import load_dotenv import os import sys sys.path.append(‘.’) from config import TARGET_ENTITIES, KEYWORDS, NOTIFY_TO # 加载环境变量 load_dotenv() api_id int(os.getenv(‘API_ID’)) api_hash os.getenv(‘API_HASH’) phone os.getenv(‘PHONE_NUMBER’) # 创建客户端session文件名为‘monitor_session’ client TelegramClient(‘monitor_session’, api_id, api_hash) async def main(): # 连接并登录 await client.start(phonephone) print(f“[{datetime.now()}] 登录成功账号{await client.get_me().username}”) # 这里将添加事件监听器 # ... 监听代码将在下一节添加 # 保持客户端运行持续监听 print(f“[{datetime.now()}] 开始监听目标{TARGET_ENTITIES}”) await client.run_until_disconnected() # 编译关键词正则提升匹配效率 compiled_keywords [] for kw in KEYWORDS: if kw.startswith(‘r’’’) and kw.endswith(‘’’’): # 简单判断是否为原始字符串表示的正则 pattern kw[2:-1] # 去除 r’’ 引号 compiled_keywords.append(re.compile(pattern)) else: # 对于普通字符串我们进行子串匹配可以忽略大小写 compiled_keywords.append(re.compile(re.escape(kw), re.IGNORECASE))注意事项首次运行client.start()时脚本会在终端提示你输入手机收到的验证码。如果账号开启了二次密码2FA还需要输入密码。登录成功后会生成一个.session文件以后运行就无需再次验证。务必保管好这个.session文件它等同于登录凭证。4.3 消息监听与关键词匹配引擎这是最核心的部分。我们使用Telethon的事件装饰器来监听新消息。为了精准控制监听范围我们需要将事件限定在TARGET_ENTITIES指定的对话中。# 接上面的 main 函数内部 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): “”“处理新消息进行关键词匹配”“” message: Message event.message sender await event.get_sender() chat await event.get_chat() # 获取干净的纯文本内容忽略媒体、链接预览等 text message.text or message.raw_text or “” if not text: return # 忽略纯媒体消息 # 简单的文本清洗 cleaned_text text.strip() # 关键词匹配检查 matched_keywords [] for pattern in compiled_keywords: if pattern.search(cleaned_text): # 如果是正则对象找到匹配的原始词 if isinstance(pattern, re.Pattern): # 这里简单记录模式串实际可更精细 matched_keywords.append(pattern.pattern) else: matched_keywords.append(pattern.pattern) if not matched_keywords: return # 没有匹配到任何关键词忽略此消息 # 匹配成功准备通知信息 print(f“[{datetime.now()}] 检测到关键词来源{chat.title} 发送者{sender.username or sender.id} 内容摘要{cleaned_text[:100]}...”) # 构建通知消息 notify_msg f“ **关键词触发警报**\n\n” notify_msg f“**群组**{chat.title}\n” notify_msg f“**发送者**{sender.first_name}” if sender.username: notify_msg f“ ({sender.username})” notify_msg f“\n**触发词**{‘ ’.join(matched_keywords)}\n” notify_msg f“**消息时间**{message.date.strftime(‘%Y-%m-%d %H:%M:%S’)}\n” notify_msg f“**消息内容**\n{cleaned_text}\n” notify_msg f“\n[点击查看上下文](https://t.me/c/{chat.id}/{message.id})” # 发送通知这里先简单打印下一节实现真实发送 print(notify_msg) # await send_notification(notify_msg) # 后续实现 # 然后运行主函数 if __name__ ‘__main__’: with client: client.loop.run_until_complete(main())这段代码做了几件关键事client.on(events.NewMessage(chats...))将处理函数限定在特定聊天列表避免监听所有对话造成干扰和性能浪费。从事件中提取消息、发送者、聊天对象等详细信息。对消息文本进行预处理。遍历编译好的关键词正则列表进行匹配。这里采用re.IGNORECASE实现不区分大小写的匹配更符合实际需求。匹配成功后格式化一条包含所有关键信息的通知消息。其中构造了Telegram的深链接t.me/c/chat.id/message.id点击可以直接跳转到原消息上下文这是“人工实时监听”中“人工介入”的关键极大提升了处理效率。4.4 实现多通道通知系统仅仅打印日志是不够的。我们需要将警报可靠地推送到指定位置。我们在utils/notifier.py中实现一个通知器它可以支持多种后端。# utils/notifier.py import asyncio from telethon import TelegramClient import logging logger logging.getLogger(__name__) class NotificationManager: def __init__(self, tg_client: TelegramClient, notify_to): self.client tg_client self.notify_to notify_to # 可以是一个用户名/ID也可以是一个列表 self._notification_queue asyncio.Queue() # 可以在此初始化其他通知渠道如邮件、Webhook等 async def send_telegram_alert(self, message: str): “”“通过Telegram发送警报”“” try: # 确保接收者是列表形式 targets self.notify_to if isinstance(self.notify_to, list) else [self.notify_to] for target in targets: await self.client.send_message(target, message, parse_mode‘markdown’, link_previewFalse) logger.info(f“Telegram警报已发送至 {target}”) await asyncio.sleep(0.5) # 发送间隔避免触发风控 except Exception as e: logger.error(f“发送Telegram警报失败{e}”) async def put_notification(self, message: str): “”“将通知放入队列由后台任务处理”“” await self._notification_queue.put(message) async def _worker(self): “”“后台工作线程从队列中取出通知并发送”“” while True: message await self._notification_queue.get() await self.send_telegram_alert(message) self._notification_queue.task_done() async def start(self): “”“启动后台通知任务”“” asyncio.create_task(self._worker()) logger.info(“通知管理器已启动”)然后在主程序monitor.py中集成这个通知管理器# 在 monitor.py 顶部导入 from utils.notifier import NotificationManager # 在 main 函数开始登录成功后初始化通知管理器 async def main(): await client.start(phonephone) print(f“登录成功...”) # 初始化通知管理器 notify_mgr NotificationManager(client, NOTIFY_TO) await notify_mgr.start() # 启动后台工作线程 # 修改 keyword_handler 函数中的通知部分 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): # ... 前面的匹配逻辑不变 ... if matched_keywords: # ... 构建 notify_msg ... # 将通知放入队列非阻塞式发送 await notify_mgr.put_notification(notify_msg)实操心得使用消息队列 (asyncio.Queue) 来处理通知发送是一个好习惯。它解耦了消息接收和发送即使网络波动导致发送暂时失败或者短时间内触发大量关键词消息也不会丢失而是排队等待发送。同时在send_message中加入短暂延迟 (await asyncio.sleep(0.5)) 是模拟人类操作、避免被Telegram风控系统误判为垃圾消息的关键技巧。5. 高级功能与优化实践5.1 动态更新关键词与群组列表让机器人重启才能修改配置太不灵活。我们可以实现一个简单的命令监听让管理员通过向机器人账号发送私聊命令来动态更新配置。# 在 monitor.py 中添加新的时间处理器 client.on(events.NewMessage(from_usersNOTIFY_TO)) # 只接收来自管理员的消息 async def admin_command_handler(event): “”“处理管理命令”“” text event.message.text.strip() if text.startswith(‘/add_keyword’): _, new_keyword text.split(maxsplit1) if new_keyword: KEYWORDS.append(new_keyword) # 重新编译正则列表需将compiled_keywords设为全局或可修改对象 # ... 更新逻辑 ... await event.reply(f“✅ 已添加关键词{new_keyword}”) elif text.startswith(‘/list_keywords’): await event.reply(f“当前监控关键词\n” “\n”.join(KEYWORDS)) # 可以添加更多命令如 /add_chat, /remove_keyword 等5.2 消息去重与频率限制在活跃的群组中同一话题可能在短时间内被反复提及导致警报轰炸。我们需要引入简单的去重和频率限制机制。# 简易的去重缓存可使用redis等持久化存储以应对重启 message_cache set() CACHE_TTL 300 # 5分钟内相同内容不重复警报 client.on(events.NewMessage(chatsTARGET_ENTITIES)) async def keyword_handler(event: events.NewMessage.Event): # ... 获取 cleaned_text ... # 内容去重检查 cache_key f“{chat.id}:{cleaned_text[:50]}” # 用聊天ID和内容前50字符做键 if cache_key in message_cache: return # 匹配关键词... if matched_keywords: # 发送通知... # 加入缓存并设置异步任务在TTL后移除 message_cache.add(cache_key) asyncio.create_task(remove_from_cache(cache_key)) async def remove_from_cache(key): await asyncio.sleep(CACHE_TTL) message_cache.discard(key)5.3 支持媒体消息与复杂内容解析有时关键词可能出现在图片的OCR文字、文档名称或链接预览中。Telethon提供了访问这些属性的能力。async def extract_full_text(event): “”“尝试从消息中提取尽可能多的文本内容”“” message event.message text message.text or message.raw_text or “” # 检查消息是否包含文档/图片/视频并获取其文件名/标题 if message.document: text f“\n[文件] {message.document.attributes[0].file_name if message.document.attributes else ‘未知文件’}” if message.photo: text “\n[图片]” # 检查是否有链接预览web_preview if message.web_preview: text f“\n[链接预览] {message.web_preview.title or ‘’} - {message.web_preview.description or ‘’}” return text.strip()在keyword_handler中调用extract_full_text(event)来获取更全面的文本内容进行匹配可以显著提高监控覆盖率。6. 部署、运维与常见问题排查6.1 服务器部署与后台运行开发完成后你需要让脚本在服务器上7x24小时运行。使用systemd或supervisor是可靠的选择。使用 systemd 服务示例创建文件/etc/systemd/system/tg-monitor.service[Unit] DescriptionTelegram Keyword Monitor Bot Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/your/tg_keyword_monitor Environment“PATH/path/to/your/venv/bin” ExecStart/path/to/your/venv/bin/python monitor.py Restartalways RestartSec10 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable tg-monitor sudo systemctl start tg-monitor sudo systemctl status tg-monitor # 查看状态6.2 常见问题与解决方案速查表问题现象可能原因排查步骤与解决方案启动时报ApiIdInvalidErrorapi_id或api_hash错误1. 检查.env文件中的值是否正确前后有无空格。2. 前往 my.telegram.org 重新创建应用并获取新凭证。登录时收不到验证码账号被限制或手机号格式错误1. 确认手机号格式为国际格式如8613800138000。2. 尝试在官方App上先登录一次解除可能的风控。3. 如果使用86号码尝试切换成86 13800138000格式有空格。运行一段时间后失去连接网络波动或Telegram服务器中断1.Telethon内置了重连机制通常会自动恢复。检查日志确认。2. 确保服务器网络稳定特别是国际出口。3. 在TelegramClient初始化时调整连接参数如设置代理。无法监听某些私有群组账号未加入该群组或没有访问权限1. 确保运行机器人的账号已经加入了目标群组。2. 对于“禁止转发消息”的群组机器人依然可以读取消息内容并触发关键词但无法生成消息链接。触发关键词后没有通知通知接收者设置错误或发送失败1. 检查NOTIFY_TO配置是否正确确保机器人账号已与接收者账号建立私聊发过一条消息即可。2. 查看程序日志确认send_message是否报错如权限不足、对方已屏蔽等。3. 检查通知队列是否正常工作。误报或漏报率高关键词匹配策略太宽泛或太严格1. 优化关键词列表使用更精确的词语或正则表达式。2. 考虑在匹配前对文本进行更细致的清洗如去除标点、统一大小写。3. 实现上下文匹配例如要求关键词“故障”附近出现“系统”一词时才报警。账号被限制或封禁行为过于频繁被判定为滥用1.最重要在所有send_message和get_entity等可能产生频繁请求的操作前后增加随机延迟 (await asyncio.sleep(random.uniform(1, 3)))。2. 避免在短时间内加入大量群组或发送大量消息。3. 如果被封尝试通过Telegram官方渠道申诉解封。6.3 性能优化与扩展思路当监控的群组非常多、消息量巨大时需要考虑性能优化异步IO充分利用确保所有网络请求如发送通知都是异步的避免阻塞主监听循环。关键词匹配算法优化对于海量关键词可以考虑使用Aho-Corasick算法等多模式匹配算法来替代简单的循环正则匹配效率可提升数个量级。数据库持久化将匹配到的消息、发送的通知记录到SQLite或PostgreSQL中便于后续查询、分析和去重。分布式监听如果单个账号监听上千个群组有压力可以考虑用多个账号多个session分担监听任务并通过一个中心服务聚合通知。规则引擎超越简单关键词实现更复杂的规则如“关键词A与关键词B在10分钟内出现”、“用户X发言中包含关键词C”等这需要引入一个规则引擎和状态管理。这个“关键词监听机器人”项目从技术上看是MTProto协议应用、异步编程和简单规则引擎的结合从实用角度看它是一个高度定制化的信息过滤与提纯工具。它的价值不在于技术有多高深而在于它能切实地将人从繁杂的实时信息流中解放出来精准捕捉到那些真正重要的信号。你可以根据自己的需求轻松地修改关键词列表、调整通知格式、甚至将报警对接到你日常使用的办公软件中让它成为你数字生活里一个沉默而可靠的哨兵。本文还有配套的精品资源点击获取