基于Telethon实现Python的Telegram全局消息关键词搜索
问题描述
我希望无需加入目标群组/频道,就能通过关键词检索Telegram中的消息。例如,在某个我并非成员的群组/频道中存在消息:"Hello, this is some information about banana",当我搜索关键词banana时,应能获取这条完整消息。但我编写的脚本仅能搜索群组名称,无法检索消息,脚本代码如下:
from telethon import functions import os, sys try: api_id = api_id api_hash = api_hash phone = phone client = TelegramClient(phone, api_id, api_hash) except KeyError: os.system("clear") print(re + "[!] run python3 setup.py first !!\n") sys.exit(1) client.connect() if not client.is_user_authorized(): client.send_code_request(phone) os.system("clear") client.sign_in(phone, input(gr + "[+] Enter the code: " + re)) os.system("clear") last_date = None chunk_size = 200 results = client( functions.contacts.SearchRequest(q="banana", limit=100) ) print(results.stringify())
解决方案
你当前使用的functions.contacts.SearchRequest仅用于搜索联系人、群组和频道的名称/用户名,无法检索消息内容。要实现无需加入群组/频道就能搜索消息的需求,需要使用functions.messages.SearchGlobalRequest(全局消息搜索)或者针对特定公开频道的functions.messages.SearchRequest。
方法1:全局消息搜索(覆盖所有可访问的公开内容)
这个方法会搜索Telegram中所有你能访问到的公开消息(包括未加入的公开群组/频道),匹配指定关键词:
from telethon import functions, TelegramClient import os, sys try: api_id = api_id # 替换为你的API ID api_hash = api_hash # 替换为你的API Hash phone = phone # 替换为你的手机号 client = TelegramClient(phone, api_id, api_hash) except KeyError: os.system("clear") print("[!] 先运行python3 setup.py !!\n") sys.exit(1) async def search_global_messages(): await client.start() # 全局搜索关键词"banana",限制返回100条结果 results = await client(functions.messages.SearchGlobalRequest( q="banana", limit=100, offset_date=None, offset_id=0, offset_peer=client.get_input_entity('me'), min_date=None, max_date=None, filter=None )) # 遍历结果,提取消息内容和来源 for msg in results.messages: if msg.message: # 获取消息所在的频道/群组名称 peer = await client.get_entity(msg.peer_id) source = peer.title if hasattr(peer, 'title') else peer.username print(f"来源: {source}") print(f"消息内容: {msg.message}\n") with client: client.loop.run_until_complete(search_global_messages())
方法2:针对特定公开频道搜索
如果已经知道目标公开频道的用户名,直接对该频道搜索消息即可,无需加入:
from telethon import functions, TelegramClient import os, sys try: api_id = api_id api_hash = api_hash phone = phone client = TelegramClient(phone, api_id, api_hash) except KeyError: os.system("clear") print("[!] 先运行python3 setup.py !!\n") sys.exit(1) async def search_specific_channel(): await client.start() # 指定目标公开频道的用户名(比如@example_channel) target_channel = await client.get_input_entity('@example_channel') # 在该频道内搜索关键词"banana" results = await client(functions.messages.SearchRequest( peer=target_channel, q="banana", limit=100, offset_date=None, offset_id=0, min_date=None, max_date=None, filter=None )) # 输出搜索到的消息 for msg in results.messages: if msg.message: print(f"消息内容: {msg.message}") with client: client.loop.run_until_complete(search_specific_channel())
注意事项
- 仅能搜索公开的群组/频道消息,私有群组/频道的消息无法检索——Telegram不允许未授权访问私有内容。
- 确保Telethon版本是最新的,避免API函数变更导致的问题。
- 替换代码中的
api_id、api_hash和phone为你自己的Telegram开发者信息。
内容的提问来源于stack exchange,提问作者Aamir Siddiqui
相关产品推荐
相关产品推荐

