如何通过消息关键词查找Telegram频道及相关消息？

官方功能局限性 #

Telegram官方自带的搜索功能无法满足全局遍历所有公开频道并抓取关键词消息的需求：

• 自带搜索仅能在你已加入的频道内检索，或通过关键词搜索相关频道，但无法批量提取频道内的目标消息。
• 没有官方API支持直接批量扫描所有公开频道的内容。

用Telethon实现的方案 #

Telethon是Telegram的第三方Python库，可以通过API实现这个需求，以下是具体步骤和代码：

1. 准备工作 #

• 安装Telethon：pip install telethon
• 获取你的API ID和API hash：前往Telegram官方开发者页面注册应用后即可获得。

2. 核心代码实现 #

from telethon import TelegramClient
from telethon.tl.functions.messages import SearchGlobalRequest
from telethon.tl.functions.channels import GetFullChannelRequest
import asyncio

# 替换为你的API信息
API_ID = 你的API_ID数字
API_HASH = '你的API_HASH字符串'
SESSION_NAME = 'keyword_search'
TARGET_KEYWORD = 'Batman'

async def fetch_keyword_messages():
    async with TelegramClient(SESSION_NAME, API_ID, API_HASH) as client:
        offset_id = 0
        # 循环分页获取全局搜索结果
        while True:
            search_results = await client(SearchGlobalRequest(
                q=TARGET_KEYWORD,
                offset_id=offset_id,
                limit=100  # 每次请求的结果数量，可调整但不要过大
            ))
            
            if not search_results.messages:
                break  # 没有更多结果时退出循环
            
            for msg in search_results.messages:
                # 仅处理频道消息（排除私聊、群组等）
                if hasattr(msg.peer_id, 'channel_id'):
                    # 获取频道完整信息，包括名称
                    channel_info = await client(GetFullChannelRequest(msg.peer_id.channel_id))
                    channel_name = channel_info.full_chat.title
                    # 格式化消息发布时间
                    publish_time = msg.date.strftime('%Y-%m-%d %H:%M:%S')
                    # 提取消息内容（处理非文本消息）
                    msg_content = msg.text if msg.text else '[非文本消息（图片/视频等）]'
                    
                    # 输出结果
                    print(f"频道名称: {channel_name}")
                    print(f"发布时间: {publish_time}")
                    print(f"消息内容: {msg_content}\n{'-'*50}")
            
            # 更新偏移量，继续下一页搜索
            offset_id = search_results.messages[-1].id

if __name__ == '__main__':
    asyncio.run(fetch_keyword_messages())

3. 常见问题及解决 #

• FloodWaitError：Telegram API有请求频率限制，遇到该错误时，可添加asyncio.sleep()延时，或使用Telethon自带的重试机制（在创建Client时设置retry_delay参数）。
• 部分频道无法访问：若遇到地区限制或权限问题，可在创建Client时配置代理（如socks5://代理）。
• 搜索结果不全：Telegram的全局搜索本身有范围限制，无法保证抓取到所有公开频道的相关消息，这是平台的固有约束。

为什么之前的Stack Overflow方案没生效？ #

可能的原因包括：

• 未处理分页：只抓取了第一页搜索结果，遗漏后续内容。
• 未正确获取频道信息：直接使用消息中的peer ID而未调用GetFullChannelRequest获取频道名称。
• 忽略了API频率限制：频繁请求导致被限制，未添加延时或重试逻辑。

内容的提问来源于stack exchange，提问作者AskingForAnswers