initial commit

app/middlewares/admin_middleware.py

@@ -0,0 +1,18 @@
+from typing import Callable, Dict, Any, Awaitable
+
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject, User
+
+from app.settings import secrets
+
+
+class AdminMiddleware(BaseMiddleware):
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: Dict[str, Any],
+    ) -> Any:
+        user: User = data.get("event_from_user")
+        if user.id == secrets.admin_id:
+            return await handler(event, data)

app/middlewares/album_middleware.py

@@ -0,0 +1,33 @@
+import asyncio
+from typing import Dict, List, Union, Callable, Any, Awaitable
+
+from aiogram import BaseMiddleware
+from aiogram.types import Message, TelegramObject
+
+DEFAULT_DELAY = 0.6
+
+
+class MediaGroupMiddleware(BaseMiddleware):
+    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)