You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何使用Telethon提取已加入群组/频道的置顶消息?

嘿,我来帮你搞定用Telethon提取所有已加入群组/频道置顶消息的事儿,步骤很清晰,跟着来就行:

准备工作
  • 先安装Telethon库:在终端执行 pip install telethon
  • 拿到Telegram API凭证:
    1. 登录Telegram官方的API管理页面,创建一个新应用(随便填名称和描述即可),之后就能拿到api_id(数字)和api_hash(字符串)
    2. 准备好你的Telegram绑定手机号,用于登录客户端
  • 确保你的账号没有被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

火山引擎 最新活动