Telegram聊天记录采集、活跃参与者识别及代码故障排查

我需要采集Telegram聊天记录并导出到Excel，收集字段包括：user id、发送日期、消息文本（含媒体的话标注类型，比如语音消息）。实验阶段计划采集最近1000条消息，但运行代码时提示无法识别指定聊天（我是该群管理员）。另外想确认代码里的延迟设置能不能避免被Telegram封禁，还有怎么识别群里最活跃的参与者。

附上尝试运行的代码：

from telethon.sync import TelegramClient
import pandas as pd
import asyncio
import time

api_id = '******'  # 替换为你的API ID
api_hash = '********'  # 替换为你的API HASH
phone = '+*******'  # 替换为你的手机号

client = TelegramClient('anon', api_id, api_hash)

data = {
    'user_id' : [],
    'message_timestamp' : [],
    'message_text' : [],
    'message_media_type' : []
}

async def collect_messages(client):
    chat = await client.get_input_entity('-*******')  # 替换为实际聊天ID

    # 初始化变量存储最后一条消息ID
    last_msg_id = None
    limit = 100  # 每次获取100条消息

    for _ in range(10):  # 10*100=1000条消息
        # 获取聊天消息
        messages = await client.get_messages(chat, limit=limit, max_id=last_msg_id)
        time.sleep(1)  # 遵守速率限制

        for message in messages:
            user_id = message.sender_id
            message_timestamp = message.date
            message_text = message.text

            if message.media is not None:
                # 不下载文件的情况下获取媒体信息
                message_media_type = message.media.__class__.__name__
            else:
                message_media_type = None

            data['user_id'].append(user_id)
            data['message_timestamp'].append(message_timestamp)
            data['message_text'].append(message_text)
            data['message_media_type'].append(message_media_type)

        # 保存最后一条消息ID，用于获取下一批消息
        last_msg_id = messages[-1].id

# 运行异步函数
with client:
    client.loop.run_until_complete(collect_messages(client))

df = pd.DataFrame(data)
df.to_excel('telegram_messages.xlsx', index=False)

我运行上述代码时，程序无法找到指定聊天。

一、解决无法识别指定聊天的问题 #

1. 确认聊天ID的正确性 #

不要直接硬编码-*******，先通过代码列出所有聊天获取目标实体：

async def collect_messages(client):
    # 列出所有聊天，获取目标聊天的正确ID/用户名
    dialogs = await client.get_dialogs()
    for dialog in dialogs:
        print(f"聊天标题: {dialog.title}, ID: {dialog.id}, 用户名: {dialog.entity.username}")

运行后从输出复制目标聊天的ID（超级群/频道通常是-100xxxxxxxxx格式）或username替换原代码中的-*******。

2. 优化get_input_entity调用 #

• 若目标聊天有公开用户名，直接传入@xxx格式更可靠：

  chat = await client.get_input_entity('@your_group_username')
  

• 确保使用完整的群ID前缀，超级群/频道必须带-100，普通群是负数ID。

3. 确认客户端登录状态 #

首次运行时需完成验证码登录流程，确保代码中的phone与注册Telegram API时的手机号一致。

二、延迟设置与防封禁说明 #

1. 修复异步代码中的延迟问题 #

原代码用time.sleep(1)会阻塞异步事件循环，应替换为异步延迟：

await asyncio.sleep(1)  # 替代time.sleep(1)

2. 安全的速率控制方案 #

• 单次请求消息数控制在100条以内，每次请求后延迟1-2秒，采集1000条消息总耗时15-20秒，完全符合Telegram的速率限制规则，不会触发封禁。
• Telethon内置了基础的速率限制处理，手动添加延迟可进一步降低风险。

三、识别最活跃的参与者 #

导出Excel后，用Pandas直接统计：

1. 统计消息发送次数 #

# 读取采集好的消息数据
df = pd.read_excel('telegram_messages.xlsx')

# 统计每个用户的消息数量，排除无user_id的系统消息
active_users = df[df['user_id'].notna()]['user_id'].value_counts().reset_index()
active_users.columns = ['user_id', '消息总数']

# 按消息数降序排列
active_users = active_users.sort_values(by='消息总数', ascending=False)

# 保存统计结果
active_users.to_excel('活跃参与者统计.xlsx', index=False)

# 打印前10名活跃用户
print("最活跃的10名参与者：")
print(active_users.head(10))

2. 可选：关联用户名 #

若需要显示用户名而非ID，可在采集阶段同步保存：

# 在消息循环中添加
if message.sender:
    username = message.sender.username or message.sender.first_name
else:
    username = None

data['username'].append(username)

后续统计时即可直接使用用户名，结果更直观。

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