from typing import Dict, Any try: from typing import Annotated except ImportError: from typing_extensions import Annotated from aiogram import BaseMiddleware from aiogram.types import TelegramObject from helper_bot.utils.base_dependency_factory import get_global_instance from helper_bot.utils.helper_func import check_access from logs.custom_logger import logger class AdminAccessMiddleware(BaseMiddleware): """Middleware для проверки административного доступа""" async def __call__(self, handler, event: TelegramObject, data: Dict[str, Any]) -> Any: if hasattr(event, 'from_user'): user_id = event.from_user.id # Получаем bot_db из data (внедренного DependenciesMiddleware) bot_db = data.get('bot_db') if not bot_db: # Fallback: получаем напрямую если middleware не сработала bdf = get_global_instance() bot_db = bdf.get_db() if not check_access(user_id, bot_db): if hasattr(event, 'answer'): await event.answer('Доступ запрещен!') return try: # Вызываем хендлер с data return await handler(event, data) except TypeError as e: if "missing 1 required positional argument: 'data'" in str(e): logger.error(f"Ошибка в AdminAccessMiddleware: {e}. Хендлер не принимает параметр 'data'") # Пытаемся вызвать хендлер без data (для совместимости с MagicData) return await handler(event) else: logger.error(f"TypeError в AdminAccessMiddleware: {e}") raise except Exception as e: logger.error(f"Неожиданная ошибка в AdminAccessMiddleware: {e}") raise # Dependency providers def get_bot_db(): """Провайдер для получения экземпляра БД""" bdf = get_global_instance() return bdf.get_db() def get_settings(): """Провайдер для получения настроек""" bdf = get_global_instance() return bdf.settings # Type aliases for dependency injection BotDB = Annotated[object, get_bot_db()] Settings = Annotated[dict, get_settings()]