Python Telegram Bot 无法一次从用户获取多个图像

Question

我正在编写一个Python脚本来创建一个电报机器人，因此该机器人的功能之一是从用户收集图像，当用户选择多个图像并同时发送它们时，机器人无法正确到达那里，就会出现问题文件 id 这里是代码：

# Function to handle picture collection
def collect_pictures(message):
    user_id = message.chat.id
    if message.photo:
        # Iterate through each photo in the array
        for photo_data in message.photo:
            file_id = photo_data.file_id
            user_data[user_id]['pictures'].append(file_id)

            # Print information about the received picture
            print(f"User {message.from_user.username} sent a picture with file_id: {file_id}")

        # Total number of pictures received
        total_pictures = len(user_data[user_id]['pictures'])
        print(f"User {message.from_user.username} has sent {total_pictures} pictures in total.")

        # You can proceed with further processing or confirmation here
        markup = create_confirm_keyboard()
        bot.send_message(message.chat.id, "Do you want to submit the collected information and pictures?", reply_markup=markup)
        bot.register_next_step_handler(message, handle_additional_pictures)
    else:
        bot.send_message(message.chat.id, "Invalid input. Please send pictures.")
        bot.register_next_step_handler(message, collect_pictures)

1. 我尝试相应地获取 file_id 并使用 for 循环，但它没有按预期运行，它获取了上传图片的数量，但在迭代并获取 id 后，它将第一个图像 file_id 存储在所有其他图像 file_id 中

Answer 1

所以基本上在电报中 - 组合媒体是视觉上分组的单独消息。您可以做的就是使用中间件在一个处理程序中捕获它们。

import asyncio
from abc import ABC
from typing import Callable, Dict, Any, Awaitable
from aiogram.types import Message

from aiogram import BaseMiddleware
from aiogram import types
from aiogram.dispatcher.event.bases import CancelHandler


class AlbumMiddleware(BaseMiddleware, ABC):
    """This middleware is for capturing media groups."""

    album_data: dict = {}

    def __init__(self, latency: int | float = 0.01):
        """
        You can provide custom latency to make sure
        albums are handled properly in highload.
        """
        self.latency = latency
        super().__init__()

    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any],
    ) -> Any:
        if not event.media_group_id:
            return

        try:
            self.album_data[event.media_group_id].append(event)
            return  # Tell aiogram to cancel handler for this group element
        except KeyError:
            self.album_data[event.media_group_id] = [event]
            await asyncio.sleep(self.latency)

            event.model_config["is_last"] = True
            data["album"] = self.album_data[event.media_group_id]

            result = await handler(event, data)

            if event.media_group_id and event.model_config.get("is_last"):
                del self.album_data[event.media_group_id]

            return result

然后在处理程序中使用它：

@router.message(F.content_type.in_({'photo', 'document'}))
async def receive_album(message: Message, state: FSMContext, album: dict[Message] = None):
    await message.answer(album)

要在启动机器人之前注册中间件，您可以使用：

middleware = AlbumMiddleware()

dp.message.outer_middleware(middleware)
dp.callback_query.outer_middleware(middleware)