使用for循环调用Telegram API触发FloodWaitError如何优化代码

问题根因 #

你遇到的FloodWaitError确实是Telegram API的速率限制触发的，报错原因里的ResolveUsernameRequest说明你遍历的list_groups里存的是群组/频道的用户名，每次调用get_messages时都会自动发起用户名解析请求，700+次高频解析直接触发了风控，要求等待8万多秒（约23小时）。

优化方案 #

• 预解析所有群组的Peer对象，避免每次循环重复发起用户名解析请求，从根源减少高频ResolveUsernameRequest调用
• 全局捕获FloodWaitError，遇到限流时自动按照API要求的时长等待后重试，不用手动中断程序
• 拆分消息拉取逻辑，不要一次性拉取全量消息，分批拉取且每轮请求之间加短延迟，平滑请求频率
• 优化存储逻辑，每处理完一个群组就持久化数据，避免程序意外中断时所有数据丢失

修改后的完整代码 #

import time
import random
import pandas as pd
from telethon.errors import FloodWaitError

# 此处保留你原有client初始化的代码

# 先预解析所有群组的Peer对象，避免后续重复解析用户名
processed_groups = []
for x in list_groups:
    while True:
        try:
            # 解析为entity对象，后续直接传entity就不会再触发用户名解析
            entity = client.get_entity(x)
            processed_groups.append(entity)
            # 每次解析完加1-2秒随机延迟，避免解析请求太频繁
            time.sleep(random.uniform(1,2))
            break
        except FloodWaitError as e:
            # 遇到限流直接等API要求的时长
            print(f"解析用户名触发限流，等待{e.seconds}秒")
            time.sleep(e.seconds)

# 存储变量的列表，注意把原time变量改名避免和内置time模块重名
message_id = []
message = []
channel_id =[]
reply_to =[] 
time_list = []
retweet_count = []
view_count = []
user_id = []
retweet = []

# 遍历预解析后的群组拉取消息
for group in processed_groups:
    while True:
        try:
            offset_id = 0
            # 每次拉取200条，是Telegram单页最大返回条数，不用直接拉取全量
            while True:
                chats = client.get_messages(group, limit=200, offset_id=offset_id)
                if not chats:
                    # 没有更多消息就退出当前群组的拉取
                    break
                for chat in chats:
                    message_id.append(chat.id)
                    message.append(chat.message)
                    channel_id.append(chat.peer_id)
                    reply_to.append(chat.reply_to_msg_id)
                    time_list.append(chat.date)
                    retweet_count.append(chat.forwards)
                    view_count.append(chat.views)
                    user_id.append(chat.from_id)
                    retweet.append(chat.fwd_from)
                # 更新偏移量，下次拉取更早的消息
                offset_id = chats[-1].id
                # 每拉取一页加2-3秒随机延迟
                time.sleep(random.uniform(2,3))
            # 单个群组处理完先备份一次，避免程序崩溃丢数据
            temp_df = pd.DataFrame({
                'message_id':message_id, 'message':message, 'channel_id': channel_id, 
                'reply_to_msg_id':reply_to, 'time':time_list, 'retweet_count':retweet_count, 
                'view_count':view_count, 'user_id':user_id, 'retweet':retweet
            })
            temp_df.to_csv("tg_data_backup.csv", index=False, encoding="utf-8-sig")
            # 处理完一个群组加1-2秒延迟，切换群组时降低请求频率
            time.sleep(random.uniform(1,2))
            break
        except FloodWaitError as e:
            print(f"拉取消息触发限流，等待{e.seconds}秒")
            time.sleep(e.seconds)

# 最终生成完整DataFrame
df = pd.DataFrame({
    'message_id':message_id, 'message':message, 'channel_id': channel_id, 
    'reply_to_msg_id':reply_to, 'time':time_list, 'retweet_count':retweet_count, 
    'view_count':view_count, 'user_id':user_id, 'retweet':retweet
})
# 最终落地存储
df.to_csv("tg_data_final.csv", index=False, encoding="utf-8-sig")

额外效率提升提示 #

• 如果有异步代码基础，可以改用Telethon的异步接口拉取，在限流等待的同时处理其他群组的拉取任务，速度会提升30%以上，新手不建议轻易尝试
• 可以设置拉取的时间范围，比如只拉取最近1年的消息，减少不必要的拉取请求
• 本地记录已经完成的群组ID，程序中断后可以直接跳过已处理的部分，不用重复爬取

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