| from __future__ import annotations | |
| from typing import TYPE_CHECKING, Any, Callable | |
| from aiogram import BaseMiddleware | |
| from loguru import logger | |
| if TYPE_CHECKING: | |
| from collections.abc import Awaitable | |
| from aiogram.types import CallbackQuery, ChatMemberUpdated, InlineQuery, Message, PreCheckoutQuery, TelegramObject | |
| class LoggingMiddleware(BaseMiddleware): | |
| def __init__(self) -> None: | |
| self.logger = logger | |
| super().__init__() | |
| def process_message(self, message: Message) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = {"chat_type": message.chat.type} | |
| if message.from_user: | |
| print_attrs["user_id"] = message.from_user.id | |
| if message.text: | |
| print_attrs["text"] = message.text | |
| if message.video: | |
| print_attrs["caption"] = message.caption | |
| print_attrs["caption_entities"] = message.caption_entities | |
| print_attrs["video_id"] = message.video.file_id | |
| print_attrs["video_unique_id"] = message.video.file_unique_id | |
| if message.audio: | |
| print_attrs["duration"] = message.audio.duration | |
| print_attrs["file_size"] = message.audio.file_size | |
| if message.photo: | |
| print_attrs["caption"] = message.caption | |
| print_attrs["caption_entities"] = message.caption_entities | |
| print_attrs["photo_id"] = message.photo[-1].file_id | |
| print_attrs["photo_unique_id"] = message.photo[-1].file_unique_id | |
| return print_attrs | |
| def process_callback_query(self, callback_query: CallbackQuery) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = { | |
| "query_id": callback_query.id, | |
| "data": callback_query.data, | |
| "user_id": callback_query.from_user.id, | |
| "inline_message_id": callback_query.inline_message_id, | |
| } | |
| if callback_query.message: | |
| print_attrs["message_id"] = callback_query.message.message_id | |
| print_attrs["chat_type"] = callback_query.message.chat.type | |
| print_attrs["chat_id"] = callback_query.message.chat.id | |
| return print_attrs | |
| def process_inline_query(self, inline_query: InlineQuery) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = { | |
| "query_id": inline_query.id, | |
| "user_id": inline_query.from_user.id, | |
| "query": inline_query.query, | |
| "offset": inline_query.offset, | |
| "chat_type": inline_query.chat_type, | |
| "location": inline_query.location, | |
| } | |
| return print_attrs | |
| def process_pre_checkout_query(self, pre_checkout_query: PreCheckoutQuery) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = { | |
| "query_id": pre_checkout_query.id, | |
| "user_id": pre_checkout_query.from_user.id, | |
| "currency": pre_checkout_query.currency, | |
| "amount": pre_checkout_query.total_amount, | |
| "payload": pre_checkout_query.invoice_payload, | |
| "option": pre_checkout_query.shipping_option_id, | |
| } | |
| return print_attrs | |
| def process_my_chat_member(self, my_chat_member: ChatMemberUpdated) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = { | |
| "user_id": my_chat_member.from_user.id, | |
| "chat_id": my_chat_member.chat.id, | |
| } | |
| return print_attrs | |
| def process_chat_member(self, chat_member: ChatMemberUpdated) -> dict[str, Any]: | |
| print_attrs: dict[str, Any] = { | |
| "user_id": chat_member.from_user.id, | |
| "chat_id": chat_member.chat.id, | |
| "old_state": chat_member.old_chat_member, | |
| "new_state": chat_member.new_chat_member, | |
| } | |
| return print_attrs | |
| async def __call__( | |
| self, | |
| handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]], | |
| event: TelegramObject, | |
| data: dict[str, Any], | |
| ) -> Any: | |
| print_attrs: dict[str, Any] = {} | |
| message: Message | None = getattr(event, "message", None) | |
| callback_query: CallbackQuery | None = getattr(event, "callback_query", None) | |
| inline_query: InlineQuery | None = getattr(event, "inline_query", None) | |
| pre_checkout_query: PreCheckoutQuery | None = getattr(event, "pre_checkout_query", None) | |
| my_chat_member: ChatMemberUpdated | None = getattr(event, "my_chat_member", None) | |
| chat_member: ChatMemberUpdated | None = getattr(event, "chat_member", None) | |
| if message: | |
| print_attrs = self.process_message(message) | |
| logger_msg = ( | |
| "received message | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| elif callback_query: | |
| print_attrs = self.process_callback_query(callback_query) | |
| logger_msg = ( | |
| "received callback query | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| elif inline_query: | |
| print_attrs = self.process_inline_query(inline_query) | |
| logger_msg = ( | |
| "received inline query | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| elif pre_checkout_query: | |
| print_attrs = self.process_pre_checkout_query(pre_checkout_query) | |
| logger_msg = ( | |
| "received pre-checkout query | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| elif my_chat_member: | |
| print_attrs = self.process_my_chat_member(my_chat_member) | |
| logger_msg = ( | |
| "received my chat member update | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| elif chat_member: | |
| print_attrs = self.process_chat_member(chat_member) | |
| logger_msg = ( | |
| "received chat member update | " | |
| + " | ".join(f"{key}: {value}" for key, value in print_attrs.items() if value is not None), | |
| ) | |
| self.logger.info(*logger_msg) | |
| return await handler(event, data) | |