使用Telethon开发频道订阅Bot时如何规避FloodWaitError？

问题 #

使用Telethon开发Telegram机器人批量订阅频道时，频繁触发FloodWaitError，即使设置了3秒的订阅间隔也无法解决，甚至出现过要求等待8小时的情况。

原代码 #

client = TelegramClient(name, id, hash)


def is_private(url: str) -> bool:
    return "/+" in url

async def join_channel():
    for links in channels:
        await asyncio.sleep(3)
        try:
            if is_private(links):
                await client(ImportChatInviteRequest(links.split("+")[-1]))
                print(0)
            else:
                await client(functions.channels.JoinChannelRequest(channel=links))
                print(1)
        except ValueError:
            print(f"ValueError cant get entity from: {links}")
        except UserAlreadyParticipantError:
            print(f"Already a participant of {links}")
        except InviteRequestSentError:
            print(f"Successfully requested to join this chat: {links}")
        except telethon.errors.rpcerrorlist.InviteHashExpiredError:
            print("afk")


client.start()
client.loop.run_until_complete(join_channel())
print("Bot start!")
client.run_until_disconnected()

报错信息 #

Traceback (most recent call last):
  File "F:\proj\PYTHON\TElethonTGbot\main.py", line 107, in <module>
    client.loop.run_until_complete(join_channel())
  File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 646, in run_until_complete
    return future.result()
  File "F:\proj\PYTHON\TElethonTGbot\main.py", line 90, in join_channel
    await client(ImportChatInviteRequest(links.split("+")[-1]))
  File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\site-packages\telethon\client\users.py", line 30, in __call__
    return await self._call(self._sender, request, ordered=ordered)
  File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\site-packages\telethon\client\users.py", line 83, in _call
    result = await future
telethon.errors.rpcerrorlist.FloodWaitError: A wait of 232 seconds is required (caused by ImportChatInviteRequest)

解决方案 #

核心思路 #

Telegram对批量加频道/群组的限流规则非常严格，尤其是私有邀请链接。需要针对性处理限流错误、动态调整操作间隔、分批次降低检测风险。

修改后的代码示例 #

import asyncio
from telethon import TelegramClient, functions
from telethon.errors import (
    ValueError, UserAlreadyParticipantError,
    InviteRequestSentError, InviteHashExpiredError,
    FloodWaitError
)

client = TelegramClient(name, id, hash)

def is_private(url: str) -> bool:
    return "/+" in url

async def join_channel():
    batch_size = 8  # 每批次处理8个频道
    batch_rest_time = 300  # 批次间休息5分钟（300秒）
    base_wait = {"public": 10, "private": 15}  # 公开/私有频道基础间隔

    for idx, link in enumerate(channels):
        # 批次间休息
        if idx != 0 and idx % batch_size == 0:
            print(f"已处理{idx}个频道，休息{batch_rest_time//60}分钟")
            await asyncio.sleep(batch_rest_time)
        
        # 循环重试当前链接，直到成功或遇到不可恢复错误
        while True:
            try:
                # 根据链接类型设置基础间隔
                await asyncio.sleep(base_wait["private"] if is_private(link) else base_wait["public"])
                
                if is_private(link):
                    await client(functions.messages.ImportChatInviteRequest(link.split("+")[-1]))
                    print(f"✅ 成功加入私有频道: {link}")
                else:
                    await client(functions.channels.JoinChannelRequest(channel=link))
                    print(f"✅ 成功加入公开频道: {link}")
                break  # 成功后处理下一个链接
            except FloodWaitError as e:
                print(f"⚠️ 触发限流，自动等待{e.wait_time}秒")
                await asyncio.sleep(e.wait_time)
            except UserAlreadyParticipantError:
                print(f"ℹ️ 已订阅频道: {link}")
                break
            except InviteRequestSentError:
                print(f"ℹ️ 已发送加入请求: {link}")
                break
            except InviteHashExpiredError:
                print(f"❌ 邀请链接已过期: {link}")
                break
            except ValueError:
                print(f"❌ 无法获取频道实体: {link}")
                break

client.start()
client.loop.run_until_complete(join_channel())
print("批量订阅完成!")
client.run_until_disconnected()

额外注意事项 #

• 新账号或活跃度低的账号限流更严格，建议先手动进行正常操作（发消息、浏览频道）提升账号活跃度后再运行批量任务。
• 不要过度缩短操作间隔，Telegram的反垃圾机制会根据账号历史行为动态调整限制，过度操作可能导致账号封禁。

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