Python Telegram Bot：调用get_chat_members_count规避Flood Limit及批量处理问题

我之前批量调用Telegram Bot API时也碰到过一模一样的问题——一次性查3000个聊天，直接触发了Flood Limit临时封禁。下面是我用装饰器解决这个问题的方案，亲测有效：

Telegram Bot API返回的FloodWait异常其实已经明确告诉你需要等待多久，我们可以写一个装饰器来捕获这个异常，自动等待后重试请求，完全不用手动处理限流逻辑。

1. 实现限流重试装饰器 #

这个装饰器会帮你自动捕获FloodWait异常，按照Telegram要求的时间暂停，然后重试请求，还能设置最大重试次数避免无限等待：

import time
from functools import wraps
from telegram.error import FloodWait

def retry_on_flood(max_retries=5):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    # 执行原请求
                    return func(*args, **kwargs)
                except FloodWait as e:
                    # 捕获到限流异常，打印提示并等待
                    print(f"触发Flood限制，需要等待{e.wait_time}秒，正在重试... (第{retries+1}/{max_retries}次)")
                    time.sleep(e.wait_time)
                    retries += 1
            # 超过最大重试次数则抛出异常
            raise Exception(f"已达到最大重试次数({max_retries})，仍无法完成请求")
        return wrapper
    return decorator

2. 把装饰器应用到你的请求中 #

你有两种方式使用这个装饰器，选适合自己的即可：

方式一：包装成安全调用方法 #

专门写一个带装饰器的安全方法，在循环中调用它：

# 给查询方法加上装饰器
@retry_on_flood(max_retries=10)
def get_chat_member_count_safe(bot, chat_id):
    return bot.get_chat_members_count(chat_id, timeout=60)

# 遍历你的聊天列表调用安全方法
for ico_tel_trim in chat_list:
    chat_id = '@' + ico_tel_trim
    try:
        ico_tel_memb = get_chat_member_count_safe(bot, chat_id)
        # 处理获取到的成员数
        print(f"聊天{chat_id}的成员数：{ico_tel_memb}")
    except Exception as e:
        print(f"获取{chat_id}成员数失败：{str(e)}")

方式二：直接装饰Bot的原生方法 #

如果你想让所有get_chat_members_count调用都自动处理限流，可以直接装饰Bot的方法，不用修改原有循环代码：

# 给bot的get_chat_members_count方法加上装饰器
bot.get_chat_members_count = retry_on_flood(max_retries=10)(bot.get_chat_members_count)

# 之后正常调用即可，和原来的代码完全一样
ico_tel_memb = bot.get_chat_members_count('@' + ico_tel_trim, timeout=60)

3. 额外优化建议 #

除了被动处理限流，你还可以主动降低触发概率：

• 添加主动延迟：在批量请求之间加1-2秒的延迟，减少短时间内的请求密度：

  for ico_tel_trim in chat_list:
      chat_id = '@' + ico_tel_trim
      ico_tel_memb = get_chat_member_count_safe(bot, chat_id)
      time.sleep(1)  # 主动延迟1秒
  

• 分批次处理：把3000个聊天分成100个一组，每组处理完后休息5分钟，进一步降低风险。
• 用异步API（推荐）：如果你用的是python-telegram-bot v13及以上版本，建议用异步版本配合asyncio.sleep()，实现非阻塞延迟，效率更高：

  import asyncio
  from telegram.error import FloodWait
  
  async def get_chat_member_count_safe(bot, chat_id):
      retries = 0
      while retries < 10:
          try:
              return await bot.get_chat_members_count(chat_id, timeout=60)
          except FloodWait as e:
              print(f"触发Flood限制，等待{e.wait_time}秒...")
              await asyncio.sleep(e.wait_time)
              retries += 1
      raise Exception("重试次数过多")
  

注意事项 #

• 严格按照FloodWait返回的wait_time等待，不要私自缩短时间，否则可能导致更久的封禁。
• 如果你的Bot完成了官方认证（绑定域名），Telegram会适当放宽限流限制，但依然要遵守基本频率规则。

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