基于Pyrogram实现Telegram消息爬取的Flood处理与断点续爬

我通过Pyrogram获取Telegram超级群组主题中的message_id与文件名，并将结果保存至results.txt文件。当待处理文件数量较多时，程序会触发Telegram的FloodWait（420）错误，导致程序中断且无法从上次处理的位置恢复执行。

我需要实现：程序触发Flood错误时，自动等待要求的时长后，从最后处理的message_id处继续爬取，避免重复执行全部任务。例如本次中断后应从Message ID:142451的位置继续。

原代码可支持单次运行处理超1000次请求，但我为实现每500次请求暂停20秒、追加写入results.txt而修改的代码，仅执行1次请求就触发了Flood错误，需解决该问题。

错误日志 #

Traceback (most recent call last):
  File "C:\Users\Peter\Desktop\script\messagge_id_telegram\temp\tg_id_fileditesto.py", line 42, in <module>
    app.run(main())
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\methods\utilities\run.py", line 77, in run
    run(coroutine)
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Users\Peter\Desktop\script\messagge_id_telegram\temp\tg_id_fileditesto.py", line 18, in main
    async for message in app.get_discussion_replies(chat_id=group_id, message_id=topic_id):
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\methods\messages\get_discussion_replies.py", line 59, in get_discussion_replies
    r = await self.invoke(
        ^^^^^^^^^^^^^^^^^^
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\methods\advanced\invoke.py", line 79, in invoke
    r = await self.session.invoke(
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\session\session.py", line 389, in invoke
    return await self.send(query, timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\session\session.py", line 357, in send
    RPCError.raise_it(result, type(data))
  File "C:\Users\Peter\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyrogram\errors\rpc_error.py", line 91, in raise_it
    raise getattr(
pyrogram.errors.exceptions.flood_420.FloodWait: Telegram says: [420 FLOOD_WAIT_X] - A wait of 16 seconds is required (caused by "messages.GetReplies")
PS C:\Users\Peter\Desktop\script\messagge_id_telegram\temp>

修改后的测试代码 #

from pyrogram import Client
import time

app = Client(
    name="@Peter_LongX",
    api_id=*******,
    api_hash="******************",
    phone_number="+3********",
    password="" or None
)

group_id = -1001867911973
topic_id = 665
msg_file_dict = {}
last_processed_message_id = 0  # Initialize the variable here

async def main():
    global last_processed_message_id  # Declare it as global

    async with app:
        count = 0
        file = open("results.txt", "a", encoding="utf-8")
        async for message in app.get_discussion_replies(chat_id=group_id, message_id=topic_id):
            if message.id <= last_processed_message_id:
                continue

            last_processed_message_id = message.id  # Update the ID of the last processed message

            print(f"Message ID: {message.id}")

            file_name = None
            if message.video or (message.document and message.document.mime_type.endswith(("rar", "zip", "pdf", "epub", "cbr"))):
                file = message.video or message.document
                print("File video o rar/zip/pdf/epub/cbr trovato")

                file_name = file.file_name or f"VID_{message.id}_{file.file_unique_id}.{file.mime_type.split('/')[-1]}"
                print(file_name)
                msg_file_dict[message.id] = file_name

            # Add the values to the text file
            with open("results.txt", "a", encoding="utf-8") as file:
                file.write(f"Message ID: {message.id}\n")
                file.write(f"File Name: {file_name}\n\n")

            count += 1
            if count == 500:
                # Take a 20 second break
                time.sleep(20)
                count = 0

app.run(main())

# Print the collected data
print(msg_file_dict.keys())  # List of Message ID
print(msg_file_dict.values())  # List of File Name

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