mirror of
https://github.com/MrSedan/neuro-reply-bot-reworked.git
synced 2026-01-14 21:49:42 +03:00
35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
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) |