如何使用Telethon提取已加入群组/频道的置顶消息?
嘿,我来帮你搞定用Telethon提取所有已加入群组/频道置顶消息的事儿,步骤很清晰,跟着来就行:
准备工作
- 先安装Telethon库:在终端执行
pip install telethon - 拿到Telegram API凭证:
- 登录Telegram官方的API管理页面,创建一个新应用(随便填名称和描述即可),之后就能拿到
api_id(数字)和api_hash(字符串) - 准备好你的Telegram绑定手机号,用于登录客户端
- 登录Telegram官方的API管理页面,创建一个新应用(随便填名称和描述即可),之后就能拿到
- 确保你的账号没有被Telegram限制API调用权限
具体实现步骤
1. 初始化Telethon客户端
先写基础的客户端初始化代码,替换成你自己的API信息和手机号:
from telethon import TelegramClient from telethon.tl.types import PeerChannel, PeerChat import asyncio import json # 替换成你的实际信息 api_id = 123456 # 你的api_id api_hash = 'your_api_hash_string_here' # 你的api_hash phone_number = '+8613xxxxxxxxx' # 你的Telegram绑定手机号
2. 获取所有已加入的群组和频道
写一个异步函数,遍历所有对话,过滤出群组和频道(排除私人聊天):
async def fetch_all_groups_and_channels(client): dialogs = await client.get_dialogs() target_chats = [] for dialog in dialogs: # 判断当前对话是群组(PeerChat)还是频道(PeerChannel) if isinstance(dialog.entity, (PeerChat, PeerChannel)): target_chats.append({ "chat_id": dialog.entity.id, "chat_title": dialog.title, "entity": dialog.entity }) return target_chats
3. 提取每个对话的置顶消息
Telethon提供了直接获取置顶消息的参数,我们可以批量提取每个群组/频道的置顶内容:
async def extract_pinned_messages(client, target_chats): pinned_msg_collection = [] for chat in target_chats: try: # 获取最多10条置顶消息(如果有多个的话),可以根据需求调整limit值 pinned_messages = await client.get_messages(chat["entity"], limit=10, pinned=True) for msg in pinned_messages: # 整理消息内容,处理无文本的媒体消息 msg_content = msg.text if msg.text else "该置顶消息为媒体内容(图片/视频等)" pinned_msg_collection.append({ "群组/频道名称": chat["chat_title"], "群组/频道ID": chat["chat_id"], "消息ID": msg.id, "消息内容": msg_content, "置顶时间": str(msg.date) }) print(f"✅ 成功提取「{chat['chat_title']}」的置顶消息") except Exception as e: print(f"❌ 提取「{chat['chat_title']}」置顶消息失败: {str(e)}") # 遇到权限问题或其他错误时跳过,继续处理下一个对话 continue return pinned_msg_collection
4. 整合代码并运行
把所有函数整合起来,执行登录、提取、保存操作:
async def main(): # 初始化客户端,session_name是本地会话文件的名称,不用改 client = TelegramClient('telegram_session', api_id, api_hash) await client.start(phone_number) print("🎉 Telegram客户端登录成功") # 获取所有目标对话 target_chats = await fetch_all_groups_and_channels(client) print(f"🔍 共找到 {len(target_chats)} 个群组/频道") # 提取置顶消息 pinned_messages = await extract_pinned_messages(client, target_chats) # 保存结果到JSON文件,方便查看 with open('telegram_pinned_messages.json', 'w', encoding='utf-8') as f: json.dump(pinned_messages, f, ensure_ascii=False, indent=4) print("📦 所有置顶消息已保存到 telegram_pinned_messages.json 文件") await client.disconnect() if __name__ == "__main__": asyncio.run(main())
注意事项
- API调用限制:Telegram对API请求有频率限制,如果你的群组/频道数量很多,建议在循环里加个小延迟(比如
await asyncio.sleep(1)),避免触发限制导致账号被临时封禁 - 权限问题:有些群组/频道可能设置了隐私权限,你没有权限查看内容,代码里已经加了异常捕获,会自动跳过这些对话
- 媒体消息处理:如果置顶消息是图片、视频等媒体,代码里会标记为「媒体内容」,如果需要提取媒体文件,可以参考Telethon的
download_media方法 - 会话文件:运行代码后会生成一个
telegram_session.session文件,下次运行时不需要再输入验证码(除非换设备)
内容的提问来源于stack exchange,提问作者vinay kusuma




