如何解决大文件上传时的FLOOD_PREMIUM_WAIT_3错误？

问题描述 #

上传2GB视频时遇到以下错误：

telethon.errors.rpcbaseerrors.FloodError: RPCError 420:
FLOOD_PREMIUM_WAIT_3 (caused by SaveBigFilePartRequest)

尝试修改分块大小无效，本人是编程新手。

原代码：

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

async def upload_file_with_retry(client, file_path, part_size_kb=512):
    """Uploads a file in parts with FloodWaitError handling."""
    part_size = part_size_kb * 1024
    file_size = os.path.getsize(file_path)
    parts_count = (file_size + part_size - 1) // part_size

    logging.info(f"Starting file upload: {file_path}, size: {file_size} bytes, parts: {parts_count}")

    with open(file_path, 'rb') as f:
        uploaded_file = None
        for part_num in range(parts_count):
            logging.info(f"Uploading part {part_num + 1} of {parts_count}")
            part_start = part_num * part_size
            f.seek(part_start)
            part_data = f.read(part_size)

            wait_time = 5
            while True:
                try:
                    uploaded_file = await client.upload_file(
                        bytes(part_data),  
                        file_name=os.path.basename(file_path),
                        file_size=file_size  
                    )
                    logging.info(f"Part {part_num + 1} successfully uploaded")
                    break
                except FloodWaitError as e:
                    logging.warning(
                        f"Flood error during upload! File: {file_path}, part {part_num + 1}, waiting {wait_time} seconds, error: {e}")
                    await asyncio.sleep(wait_time)
                    wait_time *= 2
        return uploaded_file

async def main():
    async with TelegramClient("session_name", api_id, api_hash) as client:
        caption = "caption"

        media_group = [
            types.InputMediaUploadedDocument(
                file=await upload_file_with_retry(client, "video.mp4"),
                thumb=await upload_file_with_retry(client, "image.jpg"),
                mime_type='video/mp4', 
                attributes=[
                    types.DocumentAttributeVideo(
                        duration=2940,
                        w=1920,
                        h=1080,
                        supports_streaming=True
                    )
                ]
            )
        ]

        for i in range(2, 11):
            video_file = f"video{i}.mp4"
            image_file = f"image{i}.jpg"
            if os.path.exists(video_file) and os.path.exists(image_file):
                media_group.append(
                    types.InputMediaUploadedDocument(
                        file=await upload_file_with_retry(client, video_file),
                        thumb=await upload_file_with_retry(client, image_file),
                        mime_type='video/mp4',
                        attributes=[
                            types.DocumentAttributeVideo(
                                duration=2940,
                                w=1920,
                                h=1080,
                                supports_streaming=True
                            )
                        ]
                    )
                )
            else:
                print(f"File not found: {video_file}")

        try:
            await client.send_file(
                " ",
                media_group,
                caption=caption 
            )
        except FloodWaitError as e:
            print(f"Flood error during sending! Waiting for {e.seconds} seconds...")
            await asyncio.sleep(e.seconds)

asyncio.run(main())

解决办法 #

这个错误是Telegram针对非Premium用户设置的大文件上传频率限制，不是Telethon的bug，不用等更新，按以下步骤修改即可：

1. 捕获正确的异常类型 #

你当前只处理了FloodWaitError，但FLOOD_PREMIUM_WAIT_3属于更通用的FloodError，需要把这个异常也加入捕获逻辑，直接使用错误返回的等待时间（Telethon的FloodError会包含seconds属性）。

2. 移除手动分块逻辑 #

client.upload_file本身已经内置了高效的分块上传机制，你手动拆分文件会导致请求次数翻倍，更容易触发限制。直接传入文件路径让Telethon处理即可。

3. 给多文件上传添加间隔 #

一次性上传10个大文件会连续发送大量请求，每个文件上传后加几秒延迟，降低触发限制的概率。

修正后的代码 #

import logging
import os
import asyncio
from telethon import TelegramClient, types
from telethon.errors import FloodWaitError, FloodError

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

async def upload_file_with_retry(client, file_path):
    """上传文件并处理限流异常"""
    logging.info(f"开始上传文件: {file_path}")
    while True:
        try:
            # 直接传入文件路径，让Telethon处理分块
            uploaded_file = await client.upload_file(
                file_path,
                file_name=os.path.basename(file_path)
            )
            logging.info(f"文件 {file_path} 上传成功")
            return uploaded_file
        except (FloodWaitError, FloodError) as e:
            # 提取错误中的等待时间，没有则默认等3秒
            wait_time = getattr(e, 'seconds', 3)
            logging.warning(f"上传触发限流！文件: {file_path}, 等待 {wait_time} 秒, 错误信息: {e}")
            await asyncio.sleep(wait_time)

async def main():
    async with TelegramClient("session_name", api_id, api_hash) as client:
        caption = "caption"
        media_group = []

        # 逐个处理文件，添加上传间隔
        for i in range(1, 11):
            video_file = f"video{i}.mp4"
            image_file = f"image{i}.jpg"
            if os.path.exists(video_file) and os.path.exists(image_file):
                # 上传视频和缩略图
                video_upload = await upload_file_with_retry(client, video_file)
                thumb_upload = await upload_file_with_retry(client, image_file)
                
                media_group.append(
                    types.InputMediaUploadedDocument(
                        file=video_upload,
                        thumb=thumb_upload,
                        mime_type='video/mp4',
                        attributes=[
                            types.DocumentAttributeVideo(
                                duration=2940,
                                w=1920,
                                h=1080,
                                supports_streaming=True
                            )
                        ]
                    )
                )
                # 每个文件上传后加3秒延迟
                await asyncio.sleep(3)
            else:
                print(f"文件不存在: {video_file}")

        try:
            await client.send_file(
                "目标对话ID或用户名",  # 这里替换成实际的对话ID或用户名
                media_group,
                caption=caption
            )
            logging.info("媒体组发送成功")
        except (FloodWaitError, FloodError) as e:
            wait_time = getattr(e, 'seconds', 3)
            print(f"发送触发限流！等待 {wait_time} 秒...")
            await asyncio.sleep(wait_time)
            # 等待后尝试重新发送
            await client.send_file(
                "目标对话ID或用户名",
                media_group,
                caption=caption
            )

asyncio.run(main())

额外提示 #

• 如果经常需要上传大文件，建议使用Telegram Premium账号，它有更高的上传频率和文件大小限制。
• 不要随意修改分块大小，Telethon默认的分块大小已经是最优的，手动修改反而可能触发限制。

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