如何完整下载Telegram群组消息？代码仅获25条的问题排查

代码可正常运行无报错，但仅能获取约25条消息而非全部。尝试修改limit参数、在循环内添加await asyncio.sleep(2)均无效。

原代码：

import csv
import asyncio
from telethon import TelegramClient
from telethon.sessions import StringSession

api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

session = StringSession()

async def print_messages():
    async with TelegramClient(session, api_id, api_hash) as client:
        with open('messages.csv', 'w', newline='', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow(['Sender ID', 'Text'])

            channel_username = 'channelusername'
            entity = await client.get_entity(channel_username)

            async for message in client.iter_messages(entity, limit=None):
                writer.writerow([message.sender_id, message.text])

task = asyncio.create_task(print_messages())

await task

问题原因 #

• 未验证会话限制：StringSession属于匿名会话，未完成手机号验证，Telegram对这类会话的消息获取权限做了限制，仅允许获取最新的少量消息。
• 群组权限不足：如果目标是私有群组，账号可能未完全加入，或者群组设置为仅允许成员查看加入后的消息，无法访问更早的历史内容。
• API访问限制：部分公开群组也会对未验证的客户端设置消息获取额度，仅返回最新的几十条消息。

解决方法 #

1. 使用已验证的文件会话 #

替换StringSession为常规的文件会话，首次运行时完成手机号和验证码验证，验证后的会话会保存到本地文件，解除权限限制：

import csv
import asyncio
from telethon import TelegramClient

api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

async def print_messages():
    # 使用文件会话，首次运行会提示输入手机号和验证码
    async with TelegramClient('telegram_session', api_id, api_hash) as client:
        with open('messages.csv', 'w', newline='', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow(['Sender ID', 'Text'])

            channel_username = 'channelusername'
            entity = await client.get_entity(channel_username)

            # 从最早的消息开始遍历，确保获取全部内容
            async for message in client.iter_messages(entity, limit=None, reverse=True):
                writer.writerow([message.sender_id, message.text])

asyncio.run(print_messages())

2. 确认账号群组权限 #

• 确保账号已成功加入目标群组，且群组未设置「仅允许查看加入后的消息」权限（可通过Telegram客户端查看群组设置）。
• 对于超大群组，可尝试按日期分段获取消息：

import csv
import asyncio
from datetime import datetime
from telethon import TelegramClient

api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

async def print_messages():
    async with TelegramClient('telegram_session', api_id, api_hash) as client:
        with open('messages.csv', 'w', newline='', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow(['Sender ID', 'Text'])

            channel_username = 'channelusername'
            entity = await client.get_entity(channel_username)
            
            # 设置起始日期，比如2020年1月1日
            offset_date = datetime(2020, 1, 1)
            while True:
                messages = await client.get_messages(entity, limit=100, offset_date=offset_date)
                if not messages:
                    break
                for msg in messages:
                    writer.writerow([msg.sender_id, msg.text])
                # 更新偏移日期为最后一条消息的日期
                offset_date = messages[-1].date

asyncio.run(print_messages())

3. 处理速率限制（可选） #

如果验证后仍遇到速率限制，可捕获FloodWaitError并自动等待：

import csv
import asyncio
from telethon import TelegramClient
from telethon.errors import FloodWaitError

api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'

async def print_messages():
    async with TelegramClient('telegram_session', api_id, api_hash) as client:
        with open('messages.csv', 'w', newline='', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow(['Sender ID', 'Text'])

            channel_username = 'channelusername'
            entity = await client.get_entity(channel_username)
            
            offset_id = 0
            while True:
                try:
                    messages = await client.get_messages(entity, limit=100, offset_id=offset_id)
                    if not messages:
                        break
                    for msg in messages:
                        writer.writerow([msg.sender_id, msg.text])
                    offset_id = messages[-1].id
                except FloodWaitError as e:
                    print(f"触发速率限制，等待{e.seconds}秒")
                    await asyncio.sleep(e.seconds)

asyncio.run(print_messages())

内容的提问来源于stack exchange，提问作者Foad S. Farimani