Finally fixed media groups

This commit is contained in:
2024-02-13 23:08:29 +03:00
parent 02bd67e34a
commit d181e5ac33
9 changed files with 57 additions and 28 deletions

View File

@@ -0,0 +1,35 @@
import asyncio
from typing import Any, Awaitable, Callable, Dict, List, Union
from aiogram import BaseMiddleware, types
from aiogram.types import Message, TelegramObject
DEFAULT_DELAY = .6
class MediaGroupMiddleware(BaseMiddleware):
"""Middleware for handling telegram media groups
Big thanks to https://github.com/WhiteMemory99 for this
"""
ALBUM_DATA: Dict[str, List[Message]] = {}
def __init__(self, delay: Union[int, float] = DEFAULT_DELAY):
self.delay = delay
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: Message,
data: Dict[str, Any],
) -> Any:
if not event.media_group_id:
return await handler(event, data)
try:
self.ALBUM_DATA[event.media_group_id].append(event)
return # Don't propagate the event
except KeyError:
self.ALBUM_DATA[event.media_group_id] = [event]
await asyncio.sleep(self.delay)
data["album"] = self.ALBUM_DATA.pop(event.media_group_id)
return await handler(event, data)