基于Telethon的Telegram群聊关键词实时监控机器人开发指南

简介：这是一套面向Telegram（TG）群组运营者与私域流量操盘手的关键词监听机器人源码，适用于需隐蔽监控多群消息、实现自动化响应与人工介入结合的营销或客服场景。资源基于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 整体架构与数据流设计

整个机器人的运行逻辑可以清晰地分为几个模块，形成一个高效的数据处理流水线：

1. 客户端初始化与登录模块：负责加载配置（API ID、API Hash、手机号），完成账号的登录验证（包括处理二次验证码）。这是机器人的启动入口。
2. 消息监听与获取模块：持续监听指定对话（群组或频道）的新消息。这里的关键是异步事件驱动，机器人不会阻塞等待，而是有消息到来时才触发处理函数。
3. 消息过滤与解析模块：这是核心逻辑层。收到原始消息后，需要对其进行清洗（去除多余空格、换行）、解析（获取发送者、消息内容、时间戳等），然后与预设的关键词列表进行匹配。匹配策略可以是精确匹配、模糊匹配或正则表达式匹配。
4. 通知触发与发送模块：一旦匹配成功，就需要将结果通知给“监听者”。通知方式可以多样化：向另一个指定的私聊窗口（比如你自己的账号）发送一条格式化好的摘要消息；或者将信息写入本地日志文件、数据库；更高级的可以对接Webhook，推送到钉钉、飞书或Slack。
5. 配置与状态管理模块：管理关键词列表、监听群组列表、通知对象等配置信息。这些配置最好设计成可以动态热加载，这样在机器人运行期间也能修改关键词，而无需重启。

数据流大致如下：新消息事件 -> 消息内容文本 -> 文本清洗与解析 -> 关键词匹配引擎 -> 匹配成功 -> 格式化通知消息 -> 通过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 pip

3.2 关键库的安装与说明

接下来安装核心依赖。最主要的库就是Telethon。此外，我们可能还需要python-dotenv来管理敏感配置，loguru或标准库logging来记录日志，方便排查问题。

pip install telethon python-dotenv loguru

• Telethon：这是项目的基石。它不仅处理了底层的MTProto协议通信，还提供了非常友好的高层API，比如client.on(events.NewMessage)这样的事件监听器，让我们可以用很少的代码实现强大的功能。
• python-dotenv：我们的Telegram API凭证（api_id,api_hash）以及手机号是敏感信息，不应该硬编码在脚本里。这个库帮助我们从.env文件中加载这些环境变量。
• Loguru：一个比标准库logging更友好、功能更强大的日志库。它支持彩色输出、结构化日志、异常捕获等，能极大提升开发调试效率。

3.3 获取Telegram API凭证

这是使用“普号”方案的前提。你需要访问 my.telegram.org 并用你的手机号登录。

1. 点击 “API development tools”。
2. 填写应用信息（应用名称、简称、平台等，这些可以随意填写，仅用于标识）。
3. 提交后，你将获得至关重要的两样东西：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_NUMBER=+8612345678901 # 你的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(phone=phone) 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(chats=TARGET_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())

这段代码做了几件关键事：

1. @client.on(events.NewMessage(chats=...))将处理函数限定在特定聊天列表，避免监听所有对话造成干扰和性能浪费。
2. 从事件中提取消息、发送者、聊天对象等详细信息。
3. 对消息文本进行预处理。
4. 遍历编译好的关键词正则列表进行匹配。这里采用re.IGNORECASE实现不区分大小写的匹配，更符合实际需求。
5. 匹配成功后，格式化一条包含所有关键信息的通知消息。其中，构造了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_preview=False) 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(phone=phone) print(f“登录成功...”) # 初始化通知管理器 notify_mgr = NotificationManager(client, NOTIFY_TO) await notify_mgr.start() # 启动后台工作线程 # 修改 keyword_handler 函数中的通知部分 @client.on(events.NewMessage(chats=TARGET_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_users=NOTIFY_TO)) # 只接收来自管理员的消息 async def admin_command_handler(event): “”“处理管理命令”“” text = event.message.text.strip() if text.startswith(‘/add_keyword’): _, new_keyword = text.split(maxsplit=1) 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(chats=TARGET_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] Description=Telegram Keyword Monitor Bot After=network.target [Service] Type=simple User=your_username WorkingDirectory=/path/to/your/tg_keyword_monitor Environment=“PATH=/path/to/your/venv/bin” ExecStart=/path/to/your/venv/bin/python monitor.py Restart=always RestartSec=10 StandardOutput=journal StandardError=journal [Install] WantedBy=multi-user.target

然后执行：

sudo systemctl daemon-reload sudo systemctl enable tg-monitor sudo systemctl start tg-monitor sudo systemctl status tg-monitor # 查看状态

6.2 常见问题与解决方案速查表

6.3 性能优化与扩展思路

当监控的群组非常多、消息量巨大时，需要考虑性能优化：

1. 异步IO充分利用：确保所有网络请求（如发送通知）都是异步的，避免阻塞主监听循环。
2. 关键词匹配算法优化：对于海量关键词，可以考虑使用Aho-Corasick算法等多模式匹配算法来替代简单的循环正则匹配，效率可提升数个量级。
3. 数据库持久化：将匹配到的消息、发送的通知记录到SQLite或PostgreSQL中，便于后续查询、分析和去重。
4. 分布式监听：如果单个账号监听上千个群组有压力，可以考虑用多个账号（多个session）分担监听任务，并通过一个中心服务聚合通知。
5. 规则引擎：超越简单关键词，实现更复杂的规则，如“关键词A与关键词B在10分钟内出现”、“用户X发言中包含关键词C”等，这需要引入一个规则引擎和状态管理。

这个“关键词监听机器人”项目，从技术上看是MTProto协议应用、异步编程和简单规则引擎的结合；从实用角度看，它是一个高度定制化的信息过滤与提纯工具。它的价值不在于技术有多高深，而在于它能切实地将人从繁杂的实时信息流中解放出来，精准捕捉到那些真正重要的信号。你可以根据自己的需求，轻松地修改关键词列表、调整通知格式、甚至将报警对接到你日常使用的办公软件中，让它成为你数字生活里一个沉默而可靠的哨兵。

本文还有配套的精品资源，点击获取