拓冰网站优化 
简介本资源是一套面向PHP开发者与安全研究者的Telegram群组关键词监听机器人源码聚焦个人学习场景帮助理解即时通讯平台中隐蔽式内容监控的技术实现原理。系统采用普号普通账号伪装策略通过多账户潜伏于群组实时捕获含预设关键词的消息并触发人工响应流程具备高隐匿性与低感知度特点适用于安全攻防演练、协议分析及自动化消息处理机制研究。压缩包共57个文件以36个PHP核心逻辑文件含事件监听、日志记录、API封装、Swoole/Amp迁移适配等模块为主辅以9个备份文件.zbak、2个配置JSON、1个Docker编排文件及启动/重载/停止脚本等整体仅94KB轻量易部署。目前已有125人学习下载提供完整可运行架构、清晰分层的Observer模式设计、环境适配脚本及中文教程文档便于快速理解消息监听链路、事件分发机制与Telegram Bot API集成要点。1. 项目缘起从“信息焦虑”到“主动监听”的转变做社群运营或者项目管理的朋友大概都经历过这种场景你负责的某个重要社群每天消息成千上万你不可能24小时盯着屏幕。但你又生怕错过某个关键信息比如客户的重要反馈、竞品的最新动态、或者某个突发事件的讨论。这种“信息焦虑”在快节奏的线上协作中尤为突出。传统的做法是设置消息免打扰然后定时爬楼效率低下且容易遗漏。更高级一点的做法是依赖群管理员的“人肉”提醒但这又增加了人力成本且无法保证及时性。正是在这种需求背景下“关键词监听机器人”的概念应运而生。它本质上是一个自动化信息过滤器能够7x24小时驻留在指定的社群中像一位不知疲倦的哨兵安静地扫描每一条新消息。一旦发现预设的关键词组合它就能立即触发一系列动作通知你、记录到数据库、甚至自动回复或转发到其他协作平台。这不仅仅是简单的“全体成员”而是一种精准、静默、高效的“信息雷达”。我这次分享的就是基于Telegram平台以下简称TG构建这样一套系统的完整思路与核心源码实现。它不仅仅是一个简单的“机器人”而是一个包含“普号隐身监控”与“实时人工响应”的轻量级系统。所谓“普号”指的是使用普通的个人Telegram账号而非Bot API创建的机器人账号作为监听终端这带来了更高的隐蔽性和灵活性。“隐身监控”意味着机器人读取消息但尽量不发言、不改变群组状态降低被察觉的风险。而“实时人工响应”则是在机器人捕获到关键信息后通过接口即时通知到真人由真人进行决策和回复实现了“机筛人判”的协同模式。这套方案特别适合需要从公开或半公开的Telegram群组中追踪特定话题、竞品动态、用户反馈或者进行舆情监控的团队。接下来我将从技术选型、核心原理、代码实现、部署细节以及我踩过的几个大坑为你完整拆解这个项目。2. 技术栈选型与核心原理为什么是TelethonFlask构建一个TG机器人官方首推的是通过BotFather创建并使用Bot API。这种方式稳定、合规但有明显限制Bot无法读取普通群组的消息历史在需要“隐身”监听的大多数群组中基本无用武之地。因此要实现真正的“普号监控”我们必须使用用户账号User Account来模拟客户端登录。这就是Telethon库的核心价值所在。2.1 为什么选择TelethonTelethon是一个强大且底层的Python MTProto库。MTProto是Telegram客户端与服务器通信的私有协议。使用Telethon我们可以用编程方式控制一个真实的Telegram用户账号执行几乎所有手机或桌面客户端能做的操作登录、收发消息、获取群列表、读取历史消息等等。相较于更高层封装的python-telegram-bot仅支持Bot APITelethon给了我们潜入“深海”的能力。它的另一个巨大优势是异步asyncio原生支持。Telegram的通信是高度异步的使用异步框架可以让我们用单线程高效地处理多个群组的消息流而不必担心阻塞。这对于需要同时监控数十个群的场景至关重要。2.2 后端框架轻量级的Flask监听机器人本身是“事件驱动”的它持续运行等待消息事件。但我们需要一个控制面板和消息推送接口。这就是引入Flask的原因。Flask作为一个轻量级Web框架可以快速搭建几个API端点管理端点用于动态添加/删除监控的关键词、群组查看监控日志。回调端点Webhook当Telethon客户端监听到关键词时将消息详情通过HTTP POST请求发送到这个端点从而触发后续的“实时人工响应”流程比如发送通知到钉钉、飞书或企业内部系统。这种Telethon异步事件监听 Flask同步HTTP服务的架构是一种经典的生产者-消费者模式。两者通常运行在同一个进程的不同线程或通过消息队列如Redis解耦。在本项目的初始版本中为了简化部署我采用了线程内通信的方式。2.3 核心工作流程整个系统的运行流程可以概括为以下几步初始化与登录使用Telethon创建一个TelegramClient实例填入从Telegram官网申请到的api_id和api_hash。程序首次运行会要求输入手机号验证码登录成功后会话信息会保存在本地.session文件后续启动无需重复验证。加载监听任务从数据库或配置文件中读取需要监控的群组ID或用户名和对应的关键词列表。注册事件处理器为TelegramClient注册on.NewMessage事件监听器。每当其登录账号所在群组有新消息时这个回调函数就会被触发。消息过滤与处理在事件处理器中检查新消息的chat_id是否在监控列表内并检查消息文本是否包含任何预设关键词支持简单的模糊匹配或正则表达式。如果匹配则执行核心动作。触发响应核心动作包括将消息详情发送者、时间、内容、原始消息ID等格式化然后通过requests库同步调用本地Flask服务器提供的Webhook接口。人工响应闭环Flask的Webhook接口收到数据后可以将其推送至办公IM如通过钉钉机器人、飞书Webhook提醒相关责任人。责任人点击通知链接可以快速跳转到TG群对应的消息位置进行人工回复。注意使用用户账号进行自动化操作必须严格遵守Telegram的服务条款。过度频繁的消息获取或发送行为可能导致账号被暂时限制。因此代码中必须加入适当的延迟例如asyncio.sleep和错误处理模拟人类操作节奏这是实现“隐身”的关键技术点之一。3. 核心代码拆解从登录到消息推送下面我将分模块展示最核心的代码片段并解释关键逻辑。假设我们的项目结构如下tg_keyword_monitor/ ├── config.py # 配置文件 ├── monitor.py # Telethon监听主程序 ├── web_server.py # Flask Web服务器 ├── database.py # 数据库模型可选这里用文件代替 └── requirements.txt # 依赖包3.1 配置文件 (config.py)这里存放敏感信息和全局配置。切记不要将api_id,api_hash和.session文件提交到公开仓库# config.py import os # 从环境变量读取更安全 API_ID int(os.getenv(TG_API_ID, 你的_api_id)) API_HASH os.getenv(TG_API_HASH, 你的_api_hash) PHONE_NUMBER os.getenv(TG_PHONE, 861234567890) # 绑定的手机号 # Flask服务器配置 WEBHOOK_HOST 127.0.0.1 WEBHOOK_PORT 5000 WEBHOOK_URL fhttp://{WEBHOOK_HOST}:{WEBHOOK_PORT}/webhook # 监听配置实际应从数据库读取 MONITOR_CONFIG { # 群组ID或用户名: [关键词列表] -1001234567890: [bug, 故障, 无法登录, error], # 一个技术群 group_username: [价格, 优惠, 打折, 竞品名称], # 一个公开群 } # 请求间隔避免风控 REQUEST_DELAY 1.0 # 秒3.2 Telethon监听核心 (monitor.py)这是系统的大脑负责登录、监听和初步过滤。# monitor.py import asyncio import logging from telethon import TelegramClient, events from telethon.tl.types import PeerChannel, PeerChat import requests import json from config import API_ID, API_HASH, PHONE_NUMBER, WEBHOOK_URL, MONITOR_CONFIG, REQUEST_DELAY logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) client TelegramClient(PHONE_NUMBER, API_ID, API_HASH) def send_to_webhook(chat_title, sender_name, message_text, message_link): 将匹配到的消息发送到Flask Webhook payload { chat_title: chat_title, sender: sender_name, message: message_text, link: message_link, timestamp: asyncio.get_event_loop().time() } try: resp requests.post(WEBHOOK_URL, jsonpayload, timeout5) if resp.status_code 200: logger.info(fWebhook推送成功: {chat_title} - {sender_name}) else: logger.error(fWebhook推送失败: {resp.status_code}) except Exception as e: logger.error(f发送Webhook请求异常: {e}) client.on(events.NewMessage()) async def keyword_monitor(event): 核心事件处理器过滤并处理新消息 # 1. 获取聊天信息 chat await event.get_chat() chat_id event.chat_id # 2. 检查是否为目标监控群组 if chat_id not in MONITOR_CONFIG: return # 3. 获取消息文本处理纯文本和带链接/格式的消息 message_text event.raw_text or if not message_text: # 可能是图片、文件等可以根据需要处理这里只处理文本 return # 4. 关键词匹配 keywords MONITOR_CONFIG[chat_id] matched_keywords [] for kw in keywords: # 简单的大小写不敏感包含匹配可升级为正则 if kw.lower() in message_text.lower(): matched_keywords.append(kw) if not matched_keywords: return # 5. 匹配成功准备数据 sender await event.get_sender() sender_name f{sender.first_name or } {sender.last_name or }.strip() or sender.username or 未知用户 chat_title chat.title if hasattr(chat, title) else 私聊/频道 # 构造消息链接方便人工快速跳转 if hasattr(chat, username) and chat.username: message_link fhttps://t.me/{chat.username}/{event.id} else: # 对于私有群组链接无法直接跳转记录ID message_link fchat_id: {chat_id}, msg_id: {event.id} logger.info(f[命中] 群组「{chat_title}」- 用户「{sender_name}」- 关键词「{matched_keywords}」) logger.info(f 内容: {message_text[:100]}...) # 6. 触发Webhook推送消息详情 # 使用asyncio.to_thread将同步的requests调用放到线程池执行避免阻塞异步循环 await asyncio.to_thread(send_to_webhook, chat_title, sender_name, message_text, message_link) # 7. 礼貌性延迟模拟人类降低请求频率 await asyncio.sleep(REQUEST_DELAY) async def main(): 主异步函数 await client.start(phonePHONE_NUMBER) logger.info(监听机器人启动成功) # 打印已加入的群组方便配置 async for dialog in client.iter_dialogs(): if dialog.is_group or dialog.is_channel: logger.info(f群组/频道: {dialog.name} (ID: {dialog.id})) # 持续运行直到接收到停止信号 await client.run_until_disconnected() if __name__ __main__: # 启动异步事件循环 with client: client.loop.run_until_complete(main())3.3 Flask Web服务器 (web_server.py)这是一个简单的HTTP服务器接收监听器的推送并转发。# web_server.py from flask import Flask, request, jsonify import logging import json app Flask(__name__) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 这里可以集成钉钉、飞书、企业微信等机器人的推送函数 def send_to_dingtalk(webhook_url, content): 示例推送消息到钉钉群 import requests headers {Content-Type: application/json} data { msgtype: markdown, markdown: { title: TG关键词监控告警, text: content } } try: resp requests.post(webhook_url, headersheaders, datajson.dumps(data)) return resp.status_code 200 except Exception as e: logger.error(f钉钉推送失败: {e}) return False app.route(/webhook, methods[POST]) def handle_webhook(): 接收Telethon监听器推送的Webhook if not request.is_json: return jsonify({error: Invalid content type}), 400 data request.get_json() chat_title data.get(chat_title, N/A) sender data.get(sender, N/A) message data.get(message, N/A) link data.get(link, #) logger.info(f收到告警 - 来源: {chat_title}, 发送者: {sender}) # 构建推送内容 markdown_content f### TG监控告警 **群组**: {chat_title} **发送者**: {sender} **关键词消息**: {message[:200]} **快速跳转**: [查看原文]({link}) # 调用推送函数这里以钉钉为例需配置自己的Webhook dingtalk_webhook https://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN success send_to_dingtalk(dingtalk_webhook, markdown_content) if success: return jsonify({status: ok, msg: 推送成功}), 200 else: return jsonify({status: error, msg: 推送失败}), 500 app.route(/) def index(): return TG关键词监控系统 Webhook 服务运行中。 if __name__ __main__: # 注意在生产环境中不要使用debug模式并用WSGI服务器如Gunicorn运行 app.run(host0.0.0.0, port5000, debugFalse)4. 部署实战与隐身技巧让机器人“活”在背景里代码写好了但让这个机器人稳定、隐蔽地长期运行才是真正的挑战。直接在本机运行python monitor.py和python web_server.py不是长久之计。4.1 进程管理与持久化推荐使用systemdLinux或Supervisor来管理这两个进程。以systemd为例创建两个服务文件/etc/systemd/system/tg-monitor.service[Unit] DescriptionTG Keyword Monitor Daemon Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/tg_keyword_monitor EnvironmentPATH/usr/local/bin:/usr/bin ExecStart/usr/bin/python3 /path/to/tg_keyword_monitor/monitor.py Restartalways RestartSec10 StandardOutputsyslog StandardErrorsyslog SyslogIdentifiertg-monitor [Install] WantedBymulti-user.target/etc/systemd/system/tg-webhook.service[Unit] DescriptionTG Monitor Webhook Server Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/tg_keyword_monitor EnvironmentPATH/usr/local/bin:/usr/bin ExecStart/usr/bin/python3 /path/to/tg_keyword_monitor/web_server.py Restartalways RestartSec10 StandardOutputsyslog StandardErrorsyslog SyslogIdentifiertg-webhook [Install] WantedBymulti-user.target然后使用sudo systemctl daemon-reloadsudo systemctl start tg-monitor tg-webhook启动并sudo systemctl enable设置开机自启。这样即使服务器重启服务也会自动恢复。4.2 关键的“隐身”配置与风控规避使用普通账号进行自动化操作最大的风险就是被Telegram风控系统识别并限制。以下是我在实践中总结出的几条“军规”会话管理Telethon生成的.session文件是登录凭证。务必妥善保管并设置正确的文件权限如600防止泄露。一个.session文件理论上可以永久使用除非你在其他地方登录此账号导致其失效。速率限制代码中的REQUEST_DELAY至关重要。不要设置得太小如低于0.5秒。对于消息事件处理函数即使匹配到关键词也最好在函数末尾加一个短暂的sleep。Telethon内部有自动的洪水等待Flood Wait处理但我们主动放缓节奏是更友好的行为。避免敏感操作监听机器人绝对不要在监控的群组内主动发言、点赞、转发或修改群设置。它的行为模式应尽可能接近一个“只读”的隐身用户。如果需要测试请在自己的私人小群进行。IP地址稳定性尽量在固定的、干净的IP地址如云服务器IP下运行机器人。频繁切换IP尤其是数据中心IP和家庭IP混用容易触发安全警报。模拟人类行为可以随机化延迟时间比如await asyncio.sleep(1 random.random())。在启动时可以模拟人类偶尔翻看历史消息的行为使用client.get_messages但频率要低。准备备用号重要项目不要只依赖一个监控账号。可以准备2-3个备用号使用类似的代码但不同的.session文件运行监控相同的群组。即使一个号被限制其他号也能顶上。4.3 监控配置的动态化管理上面的示例将配置硬编码在config.py中这不利于维护。生产环境应该使用数据库如SQLite或PostgreSQL来管理监控任务。可以设计一张表monitor_tasksCREATE TABLE monitor_tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id BIGINT NOT NULL, -- 群组ID chat_name TEXT, -- 群组名称便于管理 keywords TEXT NOT NULL, -- JSON数组如 [bug, error] is_active BOOLEAN DEFAULT 1, -- 是否启用 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );然后在monitor.py中定期例如每5分钟从数据库拉取最新的有效任务列表更新内存中的MONITOR_CONFIG。Flask服务也可以提供简单的管理页面来增删改查这些监控任务。5. 踩坑实录与进阶优化在开发和运行这个系统的过程中我遇到了不少问题这里分享几个典型的坑和解决方案。5.1 坑一api_id和api_hash申请与使用很多人第一步就卡在这里。api_id和api_hash不是Bot Token它们代表一个“应用”用于用户账号登录。步骤访问 https://my.telegram.org 用你的手机号登录。进入“API development tools”填写应用信息随便填如MyMonitor提交后即可获得api_id和api_hash。坑点同一个IP频繁申请api_id可能会被限制。api_hash务必保密泄露可能导致账号安全风险。5.2 坑二无法获取私有群组的chat_id或监听不到消息这是最常见的问题。原因与解决你的监听账号必须已经是该群组的成员。对于公开群你可以用client.get_entity(‘群组用户名’)来解析。对于私有群最稳妥的方式是先用账号人工加入群组然后在monitor.py的main函数启动时它会遍历并打印所有对话的ID把这个ID记录下来填入配置。权限问题即使你在群里如果是被禁言状态或群组设置了严格的权限也可能无法读取消息。确保账号在群内有基本的“查看消息”权限。5.3 坑三asyncio事件循环冲突如果你在Flask这样的同步Web框架中直接调用Telethon的异步函数或者尝试在Jupyter Notebook中运行经常会遇到Event loop is closed或Event loop already running的错误。根本原因asyncio不允许在同一个线程中嵌套运行事件循环。解决方案分离进程就像本项目设计的一样将异步的监听程序(monitor.py)和同步的Web服务器(web_server.py)作为两个独立的进程运行。这是最清晰、最稳定的方式。使用asyncio.run()确保你的异步主函数被asyncio.run(main())调用。在复杂项目中可以使用nest_asyncio库修补但这通常是下策可能引入不确定性。线程隔离如果必须在同步代码中偶尔调用异步客户端例如在Flask路由里手动抓取一次消息可以使用asyncio.run_coroutine_threadsafe并配合一个全局的、长期运行的事件循环线程。5.4 进阶优化方向关键词策略升级目前的简单包含匹配误报率高。可以引入正则表达式实现更复杂的模式匹配如匹配特定格式的版本号v\d\.\d。分词与语义过滤结合jieba中文等分词库避免子串误匹配如“苹果”匹配到“苹果手机”是合理的但“代码”匹配到“密码”就不合理。负面词过滤定义“排除词”列表当消息同时包含关键词和排除词时忽略。消息上下文获取有时单条消息意义不明需要上下文。可以在匹配到关键词后用client.get_messages(chat_id, min_idmsg_id-5, max_idmsg_id5)来获取附近的消息一并推送给人工判断。消息去重同一个问题可能在群内被多人反复提及。可以基于消息内容的哈希值或相似度在一段时间内如10分钟进行去重避免轰炸通知。状态持久化与高可用将匹配记录、发送状态存入数据库。即使服务重启也能知道哪些消息已处理。可以考虑使用Redis作为消息队列将Telethon监听器作为生产者将推送逻辑作为多个消费者实现解耦和水平扩展。容器化部署使用Docker封装整个应用将配置、.session文件通过卷挂载可以极大简化部署和迁移流程。构建这样一个系统技术本身并不复杂真正的价值在于如何将它无缝融入你的工作流并持续稳定地运行。它解放了你的双眼让你从海量的信息噪音中抽身专注于那些真正需要你回应的信号。从“人找信息”到“信息找人”这小小的改变带来的效率提升是巨大的。本文还有配套的精品资源点击获取 基于Telethon与Flask构建Telegram关键词监听机器人：实现自动化信息监控与实时响应
简介本资源是一套面向PHP开发者与安全研究者的Telegram群组关键词监听机器人源码聚焦个人学习场景帮助理解即时通讯平台中隐蔽式内容监控的技术实现原理。系统采用普号普通账号伪装策略通过多账户潜伏于群组实时捕获含预设关键词的消息并触发人工响应流程具备高隐匿性与低感知度特点适用于安全攻防演练、协议分析及自动化消息处理机制研究。压缩包共57个文件以36个PHP核心逻辑文件含事件监听、日志记录、API封装、Swoole/Amp迁移适配等模块为主辅以9个备份文件.zbak、2个配置JSON、1个Docker编排文件及启动/重载/停止脚本等整体仅94KB轻量易部署。目前已有125人学习下载提供完整可运行架构、清晰分层的Observer模式设计、环境适配脚本及中文教程文档便于快速理解消息监听链路、事件分发机制与Telegram Bot API集成要点。1. 项目缘起从“信息焦虑”到“主动监听”的转变做社群运营或者项目管理的朋友大概都经历过这种场景你负责的某个重要社群每天消息成千上万你不可能24小时盯着屏幕。但你又生怕错过某个关键信息比如客户的重要反馈、竞品的最新动态、或者某个突发事件的讨论。这种“信息焦虑”在快节奏的线上协作中尤为突出。传统的做法是设置消息免打扰然后定时爬楼效率低下且容易遗漏。更高级一点的做法是依赖群管理员的“人肉”提醒但这又增加了人力成本且无法保证及时性。正是在这种需求背景下“关键词监听机器人”的概念应运而生。它本质上是一个自动化信息过滤器能够7x24小时驻留在指定的社群中像一位不知疲倦的哨兵安静地扫描每一条新消息。一旦发现预设的关键词组合它就能立即触发一系列动作通知你、记录到数据库、甚至自动回复或转发到其他协作平台。这不仅仅是简单的“全体成员”而是一种精准、静默、高效的“信息雷达”。我这次分享的就是基于Telegram平台以下简称TG构建这样一套系统的完整思路与核心源码实现。它不仅仅是一个简单的“机器人”而是一个包含“普号隐身监控”与“实时人工响应”的轻量级系统。所谓“普号”指的是使用普通的个人Telegram账号而非Bot API创建的机器人账号作为监听终端这带来了更高的隐蔽性和灵活性。“隐身监控”意味着机器人读取消息但尽量不发言、不改变群组状态降低被察觉的风险。而“实时人工响应”则是在机器人捕获到关键信息后通过接口即时通知到真人由真人进行决策和回复实现了“机筛人判”的协同模式。这套方案特别适合需要从公开或半公开的Telegram群组中追踪特定话题、竞品动态、用户反馈或者进行舆情监控的团队。接下来我将从技术选型、核心原理、代码实现、部署细节以及我踩过的几个大坑为你完整拆解这个项目。2. 技术栈选型与核心原理为什么是TelethonFlask构建一个TG机器人官方首推的是通过BotFather创建并使用Bot API。这种方式稳定、合规但有明显限制Bot无法读取普通群组的消息历史在需要“隐身”监听的大多数群组中基本无用武之地。因此要实现真正的“普号监控”我们必须使用用户账号User Account来模拟客户端登录。这就是Telethon库的核心价值所在。2.1 为什么选择TelethonTelethon是一个强大且底层的Python MTProto库。MTProto是Telegram客户端与服务器通信的私有协议。使用Telethon我们可以用编程方式控制一个真实的Telegram用户账号执行几乎所有手机或桌面客户端能做的操作登录、收发消息、获取群列表、读取历史消息等等。相较于更高层封装的python-telegram-bot仅支持Bot APITelethon给了我们潜入“深海”的能力。它的另一个巨大优势是异步asyncio原生支持。Telegram的通信是高度异步的使用异步框架可以让我们用单线程高效地处理多个群组的消息流而不必担心阻塞。这对于需要同时监控数十个群的场景至关重要。2.2 后端框架轻量级的Flask监听机器人本身是“事件驱动”的它持续运行等待消息事件。但我们需要一个控制面板和消息推送接口。这就是引入Flask的原因。Flask作为一个轻量级Web框架可以快速搭建几个API端点管理端点用于动态添加/删除监控的关键词、群组查看监控日志。回调端点Webhook当Telethon客户端监听到关键词时将消息详情通过HTTP POST请求发送到这个端点从而触发后续的“实时人工响应”流程比如发送通知到钉钉、飞书或企业内部系统。这种Telethon异步事件监听 Flask同步HTTP服务的架构是一种经典的生产者-消费者模式。两者通常运行在同一个进程的不同线程或通过消息队列如Redis解耦。在本项目的初始版本中为了简化部署我采用了线程内通信的方式。2.3 核心工作流程整个系统的运行流程可以概括为以下几步初始化与登录使用Telethon创建一个TelegramClient实例填入从Telegram官网申请到的api_id和api_hash。程序首次运行会要求输入手机号验证码登录成功后会话信息会保存在本地.session文件后续启动无需重复验证。加载监听任务从数据库或配置文件中读取需要监控的群组ID或用户名和对应的关键词列表。注册事件处理器为TelegramClient注册on.NewMessage事件监听器。每当其登录账号所在群组有新消息时这个回调函数就会被触发。消息过滤与处理在事件处理器中检查新消息的chat_id是否在监控列表内并检查消息文本是否包含任何预设关键词支持简单的模糊匹配或正则表达式。如果匹配则执行核心动作。触发响应核心动作包括将消息详情发送者、时间、内容、原始消息ID等格式化然后通过requests库同步调用本地Flask服务器提供的Webhook接口。人工响应闭环Flask的Webhook接口收到数据后可以将其推送至办公IM如通过钉钉机器人、飞书Webhook提醒相关责任人。责任人点击通知链接可以快速跳转到TG群对应的消息位置进行人工回复。注意使用用户账号进行自动化操作必须严格遵守Telegram的服务条款。过度频繁的消息获取或发送行为可能导致账号被暂时限制。因此代码中必须加入适当的延迟例如asyncio.sleep和错误处理模拟人类操作节奏这是实现“隐身”的关键技术点之一。3. 核心代码拆解从登录到消息推送下面我将分模块展示最核心的代码片段并解释关键逻辑。假设我们的项目结构如下tg_keyword_monitor/ ├── config.py # 配置文件 ├── monitor.py # Telethon监听主程序 ├── web_server.py # Flask Web服务器 ├── database.py # 数据库模型可选这里用文件代替 └── requirements.txt # 依赖包3.1 配置文件 (config.py)这里存放敏感信息和全局配置。切记不要将api_id,api_hash和.session文件提交到公开仓库# config.py import os # 从环境变量读取更安全 API_ID int(os.getenv(TG_API_ID, 你的_api_id)) API_HASH os.getenv(TG_API_HASH, 你的_api_hash) PHONE_NUMBER os.getenv(TG_PHONE, 861234567890) # 绑定的手机号 # Flask服务器配置 WEBHOOK_HOST 127.0.0.1 WEBHOOK_PORT 5000 WEBHOOK_URL fhttp://{WEBHOOK_HOST}:{WEBHOOK_PORT}/webhook # 监听配置实际应从数据库读取 MONITOR_CONFIG { # 群组ID或用户名: [关键词列表] -1001234567890: [bug, 故障, 无法登录, error], # 一个技术群 group_username: [价格, 优惠, 打折, 竞品名称], # 一个公开群 } # 请求间隔避免风控 REQUEST_DELAY 1.0 # 秒3.2 Telethon监听核心 (monitor.py)这是系统的大脑负责登录、监听和初步过滤。# monitor.py import asyncio import logging from telethon import TelegramClient, events from telethon.tl.types import PeerChannel, PeerChat import requests import json from config import API_ID, API_HASH, PHONE_NUMBER, WEBHOOK_URL, MONITOR_CONFIG, REQUEST_DELAY logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) client TelegramClient(PHONE_NUMBER, API_ID, API_HASH) def send_to_webhook(chat_title, sender_name, message_text, message_link): 将匹配到的消息发送到Flask Webhook payload { chat_title: chat_title, sender: sender_name, message: message_text, link: message_link, timestamp: asyncio.get_event_loop().time() } try: resp requests.post(WEBHOOK_URL, jsonpayload, timeout5) if resp.status_code 200: logger.info(fWebhook推送成功: {chat_title} - {sender_name}) else: logger.error(fWebhook推送失败: {resp.status_code}) except Exception as e: logger.error(f发送Webhook请求异常: {e}) client.on(events.NewMessage()) async def keyword_monitor(event): 核心事件处理器过滤并处理新消息 # 1. 获取聊天信息 chat await event.get_chat() chat_id event.chat_id # 2. 检查是否为目标监控群组 if chat_id not in MONITOR_CONFIG: return # 3. 获取消息文本处理纯文本和带链接/格式的消息 message_text event.raw_text or if not message_text: # 可能是图片、文件等可以根据需要处理这里只处理文本 return # 4. 关键词匹配 keywords MONITOR_CONFIG[chat_id] matched_keywords [] for kw in keywords: # 简单的大小写不敏感包含匹配可升级为正则 if kw.lower() in message_text.lower(): matched_keywords.append(kw) if not matched_keywords: return # 5. 匹配成功准备数据 sender await event.get_sender() sender_name f{sender.first_name or } {sender.last_name or }.strip() or sender.username or 未知用户 chat_title chat.title if hasattr(chat, title) else 私聊/频道 # 构造消息链接方便人工快速跳转 if hasattr(chat, username) and chat.username: message_link fhttps://t.me/{chat.username}/{event.id} else: # 对于私有群组链接无法直接跳转记录ID message_link fchat_id: {chat_id}, msg_id: {event.id} logger.info(f[命中] 群组「{chat_title}」- 用户「{sender_name}」- 关键词「{matched_keywords}」) logger.info(f 内容: {message_text[:100]}...) # 6. 触发Webhook推送消息详情 # 使用asyncio.to_thread将同步的requests调用放到线程池执行避免阻塞异步循环 await asyncio.to_thread(send_to_webhook, chat_title, sender_name, message_text, message_link) # 7. 礼貌性延迟模拟人类降低请求频率 await asyncio.sleep(REQUEST_DELAY) async def main(): 主异步函数 await client.start(phonePHONE_NUMBER) logger.info(监听机器人启动成功) # 打印已加入的群组方便配置 async for dialog in client.iter_dialogs(): if dialog.is_group or dialog.is_channel: logger.info(f群组/频道: {dialog.name} (ID: {dialog.id})) # 持续运行直到接收到停止信号 await client.run_until_disconnected() if __name__ __main__: # 启动异步事件循环 with client: client.loop.run_until_complete(main())3.3 Flask Web服务器 (web_server.py)这是一个简单的HTTP服务器接收监听器的推送并转发。# web_server.py from flask import Flask, request, jsonify import logging import json app Flask(__name__) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 这里可以集成钉钉、飞书、企业微信等机器人的推送函数 def send_to_dingtalk(webhook_url, content): 示例推送消息到钉钉群 import requests headers {Content-Type: application/json} data { msgtype: markdown, markdown: { title: TG关键词监控告警, text: content } } try: resp requests.post(webhook_url, headersheaders, datajson.dumps(data)) return resp.status_code 200 except Exception as e: logger.error(f钉钉推送失败: {e}) return False app.route(/webhook, methods[POST]) def handle_webhook(): 接收Telethon监听器推送的Webhook if not request.is_json: return jsonify({error: Invalid content type}), 400 data request.get_json() chat_title data.get(chat_title, N/A) sender data.get(sender, N/A) message data.get(message, N/A) link data.get(link, #) logger.info(f收到告警 - 来源: {chat_title}, 发送者: {sender}) # 构建推送内容 markdown_content f### TG监控告警 **群组**: {chat_title} **发送者**: {sender} **关键词消息**: {message[:200]} **快速跳转**: [查看原文]({link}) # 调用推送函数这里以钉钉为例需配置自己的Webhook dingtalk_webhook https://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN success send_to_dingtalk(dingtalk_webhook, markdown_content) if success: return jsonify({status: ok, msg: 推送成功}), 200 else: return jsonify({status: error, msg: 推送失败}), 500 app.route(/) def index(): return TG关键词监控系统 Webhook 服务运行中。 if __name__ __main__: # 注意在生产环境中不要使用debug模式并用WSGI服务器如Gunicorn运行 app.run(host0.0.0.0, port5000, debugFalse)4. 部署实战与隐身技巧让机器人“活”在背景里代码写好了但让这个机器人稳定、隐蔽地长期运行才是真正的挑战。直接在本机运行python monitor.py和python web_server.py不是长久之计。4.1 进程管理与持久化推荐使用systemdLinux或Supervisor来管理这两个进程。以systemd为例创建两个服务文件/etc/systemd/system/tg-monitor.service[Unit] DescriptionTG Keyword Monitor Daemon Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/tg_keyword_monitor EnvironmentPATH/usr/local/bin:/usr/bin ExecStart/usr/bin/python3 /path/to/tg_keyword_monitor/monitor.py Restartalways RestartSec10 StandardOutputsyslog StandardErrorsyslog SyslogIdentifiertg-monitor [Install] WantedBymulti-user.target/etc/systemd/system/tg-webhook.service[Unit] DescriptionTG Monitor Webhook Server Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/tg_keyword_monitor EnvironmentPATH/usr/local/bin:/usr/bin ExecStart/usr/bin/python3 /path/to/tg_keyword_monitor/web_server.py Restartalways RestartSec10 StandardOutputsyslog StandardErrorsyslog SyslogIdentifiertg-webhook [Install] WantedBymulti-user.target然后使用sudo systemctl daemon-reloadsudo systemctl start tg-monitor tg-webhook启动并sudo systemctl enable设置开机自启。这样即使服务器重启服务也会自动恢复。4.2 关键的“隐身”配置与风控规避使用普通账号进行自动化操作最大的风险就是被Telegram风控系统识别并限制。以下是我在实践中总结出的几条“军规”会话管理Telethon生成的.session文件是登录凭证。务必妥善保管并设置正确的文件权限如600防止泄露。一个.session文件理论上可以永久使用除非你在其他地方登录此账号导致其失效。速率限制代码中的REQUEST_DELAY至关重要。不要设置得太小如低于0.5秒。对于消息事件处理函数即使匹配到关键词也最好在函数末尾加一个短暂的sleep。Telethon内部有自动的洪水等待Flood Wait处理但我们主动放缓节奏是更友好的行为。避免敏感操作监听机器人绝对不要在监控的群组内主动发言、点赞、转发或修改群设置。它的行为模式应尽可能接近一个“只读”的隐身用户。如果需要测试请在自己的私人小群进行。IP地址稳定性尽量在固定的、干净的IP地址如云服务器IP下运行机器人。频繁切换IP尤其是数据中心IP和家庭IP混用容易触发安全警报。模拟人类行为可以随机化延迟时间比如await asyncio.sleep(1 random.random())。在启动时可以模拟人类偶尔翻看历史消息的行为使用client.get_messages但频率要低。准备备用号重要项目不要只依赖一个监控账号。可以准备2-3个备用号使用类似的代码但不同的.session文件运行监控相同的群组。即使一个号被限制其他号也能顶上。4.3 监控配置的动态化管理上面的示例将配置硬编码在config.py中这不利于维护。生产环境应该使用数据库如SQLite或PostgreSQL来管理监控任务。可以设计一张表monitor_tasksCREATE TABLE monitor_tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id BIGINT NOT NULL, -- 群组ID chat_name TEXT, -- 群组名称便于管理 keywords TEXT NOT NULL, -- JSON数组如 [bug, error] is_active BOOLEAN DEFAULT 1, -- 是否启用 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );然后在monitor.py中定期例如每5分钟从数据库拉取最新的有效任务列表更新内存中的MONITOR_CONFIG。Flask服务也可以提供简单的管理页面来增删改查这些监控任务。5. 踩坑实录与进阶优化在开发和运行这个系统的过程中我遇到了不少问题这里分享几个典型的坑和解决方案。5.1 坑一api_id和api_hash申请与使用很多人第一步就卡在这里。api_id和api_hash不是Bot Token它们代表一个“应用”用于用户账号登录。步骤访问 https://my.telegram.org 用你的手机号登录。进入“API development tools”填写应用信息随便填如MyMonitor提交后即可获得api_id和api_hash。坑点同一个IP频繁申请api_id可能会被限制。api_hash务必保密泄露可能导致账号安全风险。5.2 坑二无法获取私有群组的chat_id或监听不到消息这是最常见的问题。原因与解决你的监听账号必须已经是该群组的成员。对于公开群你可以用client.get_entity(‘群组用户名’)来解析。对于私有群最稳妥的方式是先用账号人工加入群组然后在monitor.py的main函数启动时它会遍历并打印所有对话的ID把这个ID记录下来填入配置。权限问题即使你在群里如果是被禁言状态或群组设置了严格的权限也可能无法读取消息。确保账号在群内有基本的“查看消息”权限。5.3 坑三asyncio事件循环冲突如果你在Flask这样的同步Web框架中直接调用Telethon的异步函数或者尝试在Jupyter Notebook中运行经常会遇到Event loop is closed或Event loop already running的错误。根本原因asyncio不允许在同一个线程中嵌套运行事件循环。解决方案分离进程就像本项目设计的一样将异步的监听程序(monitor.py)和同步的Web服务器(web_server.py)作为两个独立的进程运行。这是最清晰、最稳定的方式。使用asyncio.run()确保你的异步主函数被asyncio.run(main())调用。在复杂项目中可以使用nest_asyncio库修补但这通常是下策可能引入不确定性。线程隔离如果必须在同步代码中偶尔调用异步客户端例如在Flask路由里手动抓取一次消息可以使用asyncio.run_coroutine_threadsafe并配合一个全局的、长期运行的事件循环线程。5.4 进阶优化方向关键词策略升级目前的简单包含匹配误报率高。可以引入正则表达式实现更复杂的模式匹配如匹配特定格式的版本号v\d\.\d。分词与语义过滤结合jieba中文等分词库避免子串误匹配如“苹果”匹配到“苹果手机”是合理的但“代码”匹配到“密码”就不合理。负面词过滤定义“排除词”列表当消息同时包含关键词和排除词时忽略。消息上下文获取有时单条消息意义不明需要上下文。可以在匹配到关键词后用client.get_messages(chat_id, min_idmsg_id-5, max_idmsg_id5)来获取附近的消息一并推送给人工判断。消息去重同一个问题可能在群内被多人反复提及。可以基于消息内容的哈希值或相似度在一段时间内如10分钟进行去重避免轰炸通知。状态持久化与高可用将匹配记录、发送状态存入数据库。即使服务重启也能知道哪些消息已处理。可以考虑使用Redis作为消息队列将Telethon监听器作为生产者将推送逻辑作为多个消费者实现解耦和水平扩展。容器化部署使用Docker封装整个应用将配置、.session文件通过卷挂载可以极大简化部署和迁移流程。构建这样一个系统技术本身并不复杂真正的价值在于如何将它无缝融入你的工作流并持续稳定地运行。它解放了你的双眼让你从海量的信息噪音中抽身专注于那些真正需要你回应的信号。从“人找信息”到“信息找人”这小小的改变带来的效率提升是巨大的。本文还有配套的精品资源点击获取