Small Liba for processing albums aiogram3

Не знаю, необходимо ли это вообще) Обычно все публичные решения выполнены с использованием Middleware. Если кому-то надо - вот решение через декоратор.
Кода вышло слишком мало, чтобы создавать отдельную репу под это, поэтому вот:

import asyncio
from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, TypeAlias, Union

from aiogram.types import (
InputMedia,
InputMediaAudio,
InputMediaDocument,
InputMediaPhoto,
InputMediaVideo,
Message,
)


Handler: TypeAlias = Callable[..., Awaitable[Any]]

def is_media_message(message: Message) -> bool:
"""
Проверяет, если сообщение содержит медиа.

Принимает:
- message: сообщение Telegram

Возвращает:
- bool: есть ли в сообщении фото/видео/документ/аудио
"""
return bool(
getattr(message, "photo", None)
or getattr(message, "video", None)
or getattr(message, "document", None)
or getattr(message, "audio", None)
)


def _caption_kwargs(message: Message) -> Dict[str, Any]:
return {
"caption": getattr(message, "caption", None),
"caption_entities": getattr(message, "caption_entities", None),
}


def to_input_media(messages: Sequence[Message]) -> List[InputMedia]:
"""
Преобразует сообщения альбома в список InputMedia.

Принимает:
- messages: сообщения с медиа (обычно из альбома)

Возвращает:
- list[InputMedia]: готово для answer_media_group
"""
result: List[InputMedia] = []
for msg in messages:
kwargs = _caption_kwargs(msg)
if getattr(msg, "photo", None):
result.append(InputMediaPhoto(media=msg.photo[-1].file_id, **kwargs))
elif getattr(msg, "video", None):
result.append(InputMediaVideo(media=msg.video.file_id, **kwargs))
elif getattr(msg, "document", None):
result.append(InputMediaDocument(media=msg.document.file_id, **kwargs))
elif getattr(msg, "audio", None):
result.append(InputMediaAudio(media=msg.audio.file_id, **kwargs))
else:
continue

return result


def album_handler(
*,
delay: Union[int, float] = 0.6,
pass_input_media: bool = False,
singles: bool = True,
) -> Callable[[Handler], Handler]:
"""
Декоратор, принимающий сообщения одного альбома.
Для одиночного медиа при singles=True вызывает обработчик с album=[message].

Принимает:
- delay: задержка перед вызовом хендлера для сбора альбома
- pass_input_media: добавлять ли input_media в аргументы хендлера (для отправки альбома)
- singles: включать ли одиночные медиа

Требования к обработчику:
- принимает аргументы message, album, input_media (если pass_input_media=True)
"""

def decorator(func: Handler) -> Handler:
buffers: Dict[str, List[Message]] = {}
scheduled: Dict[str, bool] = {}

async def wrapper(*args: Any, **kwargs: Any) -> Any:
message: Optional[Message] = None
for value in args:
if isinstance(value, Message):
message = value
break

if message is None:
return await func(*args, **kwargs)

group_id = getattr(message, "media_group_id", None)
if group_id:
if group_id not in buffers:
buffers[group_id] = []
buffers[group_id].append(message)

if not scheduled.get(group_id):
scheduled[group_id] = True
await asyncio.sleep(delay)
album_messages = buffers.pop(group_id, [])
scheduled.pop(group_id, None)
if not album_messages:
return None
if pass_input_media:
return await func(message, album_messages, to_input_media(album_messages))
return await func(message, album_messages)
return None

if singles and is_media_message(message):
single_album = [message]
if pass_input_media:
return await func(message, single_album, to_input_media(single_album))
return await func(message, single_album)

return await func(message)

return wrapper

return decorator

Python

import asyncio

from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, TypeAlias, Union



from aiogram.types import (

    InputMedia,

    InputMediaAudio,

    InputMediaDocument,

    InputMediaPhoto,

    InputMediaVideo,

    Message,

)





Handler: TypeAlias = Callable[..., Awaitable[Any]]



def is_media_message(message: Message) -> bool:

    """

    Проверяет, если сообщение содержит медиа.



    Принимает:

    - message: сообщение Telegram



    Возвращает:

    - bool: есть ли в сообщении фото/видео/документ/аудио

    """

    return bool(

        getattr(message, "photo", None)

        or getattr(message, "video", None)

        or getattr(message, "document", None)

        or getattr(message, "audio", None)

    )





def _caption_kwargs(message: Message) -> Dict[str, Any]:

    return {

        "caption": getattr(message, "caption", None),

        "caption_entities": getattr(message, "caption_entities", None),

    }





def to_input_media(messages: Sequence[Message]) -> List[InputMedia]:

    """

    Преобразует сообщения альбома в список InputMedia.



    Принимает:

    - messages: сообщения с медиа (обычно из альбома)



    Возвращает:

    - list[InputMedia]: готово для answer_media_group

    """

    result: List[InputMedia] = []

    for msg in messages:

        kwargs = _caption_kwargs(msg)

        if getattr(msg, "photo", None):

            result.append(InputMediaPhoto(media=msg.photo[-1].file_id, **kwargs))

        elif getattr(msg, "video", None):

            result.append(InputMediaVideo(media=msg.video.file_id, **kwargs))

        elif getattr(msg, "document", None):

            result.append(InputMediaDocument(media=msg.document.file_id, **kwargs))

        elif getattr(msg, "audio", None):

            result.append(InputMediaAudio(media=msg.audio.file_id, **kwargs))

        else:

            continue

      

    return result





def album_handler(

    *,

    delay: Union[int, float] = 0.6,

    pass_input_media: bool = False,

    singles: bool = True,

) -> Callable[[Handler], Handler]:

    """

    Декоратор, принимающий сообщения одного альбома.

    Для одиночного медиа при singles=True вызывает обработчик с album=[message].



    Принимает:

    - delay: задержка перед вызовом хендлера для сбора альбома

    - pass_input_media: добавлять ли input_media в аргументы хендлера (для отправки альбома)

    - singles: включать ли одиночные медиа



    Требования к обработчику:

    - принимает аргументы message, album, input_media (если pass_input_media=True)

    """



    def decorator(func: Handler) -> Handler:

        buffers: Dict[str, List[Message]] = {}

        scheduled: Dict[str, bool] = {}



        async def wrapper(*args: Any, **kwargs: Any) -> Any:

            message: Optional[Message] = None

            for value in args:

                if isinstance(value, Message):

                    message = value

                    break



            if message is None:

                return await func(*args, **kwargs)



            group_id = getattr(message, "media_group_id", None)

            if group_id:

                if group_id not in buffers:

                    buffers[group_id] = []

                buffers[group_id].append(message)



                if not scheduled.get(group_id):

                    scheduled[group_id] = True

                    await asyncio.sleep(delay)

                    album_messages = buffers.pop(group_id, [])

                    scheduled.pop(group_id, None)

                    if not album_messages:

                        return None

                    if pass_input_media:

                        return await func(message, album_messages, to_input_media(album_messages))

                    return await func(message, album_messages)

                return None



            if singles and is_media_message(message):

                single_album = [message]

                if pass_input_media:

                    return await func(message, single_album, to_input_media(single_album))

                return await func(message, single_album)



            return await func(message)



        return wrapper



    return decorator

Пример использования(простейший эхо-бот):

import asyncio
from typing import List

from aiogram import Bot, Dispatcher, F
from aiogram.types import Message

from aiogram_media import InputMedia, album_handler
# Куда удобно себе закиньте код и оттуда импортируйте :)

async def main() -> None:
token = "" # <--- ТОКЕН СЮДА

bot = Bot(token)
dp = Dispatcher()

@dp.message(F.media_group_id | F.photo | F.video | F.document | F.audio | F.animation)
@album_handler(pass_input_media=True)
async def handle_media(message: Message, album: List[Message], input_media: List[InputMedia]) -> None:
await message.answer_media_group(input_media)

@dp.message(F.text)
async def handle_text(message: Message) -> None:
await message.answer("Бот запущен. Отправьте альбом или медиа.")

await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())


if __name__ == "__main__":
asyncio.run(main())

Python

import asyncio

from typing import List



from aiogram import Bot, Dispatcher, F

from aiogram.types import Message



from aiogram_media import InputMedia, album_handler

# Куда удобно себе закиньте код и оттуда импортируйте :)



async def main() -> None:

    token = "" # <--- ТОКЕН СЮДА



    bot = Bot(token)

    dp = Dispatcher()



    @dp.message(F.media_group_id | F.photo | F.video | F.document | F.audio | F.animation)

    @album_handler(pass_input_media=True)

    async def handle_media(message: Message, album: List[Message], input_media: List[InputMedia]) -> None:

        await message.answer_media_group(input_media)



    @dp.message(F.text)

    async def handle_text(message: Message) -> None:

        await message.answer("Бот запущен. Отправьте альбом или медиа.")



    await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())





if __name__ == "__main__":

    asyncio.run(main())

Спасибо за внимание