Zakhar Izmaylov
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from aiogram import BaseMiddleware
|
|
from aiogram.types import CallbackQuery, Message, TelegramObject
|
|
|
|
from logger import logger
|
|
|
|
|
|
class LoggingMiddleware(BaseMiddleware):
|
|
async def __call__(
|
|
self,
|
|
handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
|
|
event: TelegramObject,
|
|
data: dict[str, Any],
|
|
) -> Any:
|
|
user_info = self._extract_user_info(event)
|
|
|
|
logger.info(
|
|
f"Активность пользователя - "
|
|
f"ID пользователя: {user_info['user_id']}, "
|
|
f"Имя пользователя: {user_info['username']}, "
|
|
f"Действие: {user_info['action']}"
|
|
)
|
|
return await handler(event, data)
|
|
|
|
def _extract_user_info(self, event: TelegramObject) -> dict[str, str | None]:
|
|
user_id = None
|
|
username = None
|
|
action = None
|
|
|
|
if isinstance(event, Message):
|
|
user = event.from_user
|
|
user_id = user.id
|
|
username = user.username
|
|
action = f"Сообщение: {event.text}"
|
|
elif isinstance(event, CallbackQuery):
|
|
user = event.from_user
|
|
user_id = user.id
|
|
username = user.username
|
|
action = f"Обратный вызов: {event.data}"
|
|
|
|
return {"user_id": user_id, "username": username, "action": action}
|