import html from datetime import datetime from typing import Any, Dict from aiogram import BaseMiddleware, types from aiogram.types import CallbackQuery, Message, TelegramObject from helper_bot.utils.base_dependency_factory import get_global_instance from logs.custom_logger import logger bdf = get_global_instance() BotDB = bdf.get_db() class BlacklistMiddleware(BaseMiddleware): async def __call__( self, handler, event: TelegramObject, data: Dict[str, Any] ) -> Any: # Проверяем тип события и получаем пользователя user = None if isinstance(event, Message): user = event.from_user elif isinstance(event, CallbackQuery): user = event.from_user # Если это не сообщение или callback, пропускаем проверку if not user: return await handler(event, data) logger.info(f"Вызов BlacklistMiddleware для пользователя {user.username}") # Используем асинхронную версию для предотвращения блокировки if await BotDB.check_user_in_blacklist(user.id): logger.info( f"BlacklistMiddleware результат для пользователя: {user.username} заблокирован!" ) user_info = await BotDB.get_blacklist_users_by_id(user.id) # Экранируем потенциально проблемные символы reason = ( html.escape(str(user_info[1])) if user_info and user_info[1] else "Не указана" ) # Преобразуем timestamp в человекочитаемый формат if user_info and user_info[2]: try: timestamp = int(user_info[2]) date_unban = datetime.fromtimestamp(timestamp).strftime( "%d-%m-%Y %H:%M" ) except (ValueError, TypeError): date_unban = "Не указана" else: date_unban = "Не указана" # Отправляем сообщение в зависимости от типа события if isinstance(event, Message): await event.answer( f"Ты заблокирован.\nПричина блокировки: {reason}\nДата разбана: {date_unban}" ) elif isinstance(event, CallbackQuery): await event.answer( f"Ты заблокирован.\nПричина блокировки: {reason}\nДата разбана: {date_unban}", show_alert=True, ) return False logger.info( f"BlacklistMiddleware результат для пользователя: {user.username} доступ разрешен" ) return await handler(event, data)