如何通过Pyrogram获取Telegram超级群Slow mode剩余等待时长?
解决Telegram慢模式剩余时长检测与发送限制问题的思路
1. 捕获错误提取等待时长并缓存
当触发420 SLOWMODE_WAIT_X错误时,Pyrogram会在异常信息里返回需要等待的秒数。可以捕获这个SlowmodeWait异常,提取时长后,给对应聊天ID记录下次允许发送的时间戳(当前时间+等待时长)。后续发送前先检查缓存,若当前时间未到记录时间,就等待至对应时间再执行发送。
代码示例:
from pyrogram.errors import SlowmodeWait import time from datetime import datetime, timedelta # 缓存聊天ID对应的下次可发送时间 chat_slowmode_cache = {} async def send_message_with_slowmode(client, chat_id, text): try: # 检查缓存,若未到发送时间则等待 if chat_id in chat_slowmode_cache: next_send_time = chat_slowmode_cache[chat_id] if datetime.now() < next_send_time: wait_seconds = (next_send_time - datetime.now()).total_seconds() time.sleep(wait_seconds) await client.send_message(chat_id, text) # 发送成功后清除该群的缓存记录 chat_slowmode_cache.pop(chat_id, None) except SlowmodeWait as e: # 提取错误中的等待秒数,更新缓存 wait_seconds = e.x next_send_time = datetime.now() + timedelta(seconds=wait_seconds) chat_slowmode_cache[chat_id] = next_send_time # 等待后重试发送 time.sleep(wait_seconds) await send_message_with_slowmode(client, chat_id, text)
2. 预配置已知慢模式群的固定时长
既然你已经明确部分群的慢模式为15分钟(900秒)或60分钟(3600秒),可以提前把这些群的ID和对应时长存入配置字典。发送前根据群ID获取固定时长,对比上次发送时间与当前时间的间隔,若不足时长则等待补足。
代码示例:
import time # 预配置群慢模式时长(chat_id: 秒数) chat_slowmode_config = { -1001234567890: 900, # 15分钟慢模式群 -1000987654321: 3600 # 60分钟慢模式群 } # 记录每个群的上次发送时间戳 last_send_timestamps = {} async def send_message(client, chat_id, text): current_time = time.time() if chat_id in chat_slowmode_config: slow_duration = chat_slowmode_config[chat_id] # 检查上次发送时间,计算需要等待的时长 if chat_id in last_send_timestamps: elapsed = current_time - last_send_timestamps[chat_id] if elapsed < slow_duration: time.sleep(slow_duration - elapsed) # 更新该群的上次发送时间 last_send_timestamps[chat_id] = time.time() await client.send_message(chat_id, text)
这种方式无需依赖错误捕获,适合已明确慢模式时长的群,效率更高。
3. 调用Telegram底层API获取剩余时长
Pyrogram的Chat对象仅暴露slowmode_enabled属性,但可以直接调用Telegram底层的messages.getChat方法,获取包含slowmode_remaining字段的完整聊天信息,从而拿到剩余等待时长。
代码示例:
from pyrogram.raw.functions.messages import GetChat import time async def get_slowmode_remaining(client, chat_id): try: # 调用底层API获取聊天详情 peer = await client.resolve_peer(chat_id) chat = await client.invoke(GetChat(chat_id=peer)) # 返回剩余时长,无慢模式则返回0 return getattr(chat, 'slowmode_remaining', 0) except Exception as e: print(f"获取慢模式剩余时长失败: {str(e)}") return 0 async def send_message(client, chat_id, text): remaining = await get_slowmode_remaining(client, chat_id) if remaining > 0: time.sleep(remaining) await client.send_message(chat_id, text)
注意:该方法依赖Telegram底层API的返回格式,需测试验证兼容性,部分场景可能因权限或API版本变化失效。
4. 混合策略:预配置+错误捕获兜底
将前两种方法结合,对已知时长的群用预配置规则,未知时长的群第一次发送触发错误后,把获取到的时长存入缓存,后续发送直接使用缓存的时长判断。这种方式兼顾效率与兼容性,覆盖所有场景。
内容的提问来源于stack exchange,提问作者Unthought
相关产品推荐
相关产品推荐

