src/auth/access.py - master - vladelo/startups

src/auth/access.py
114 строк5 KB
vladelo
Adding docstrings
18 авг 2025, 15:05
18 авг 2025, 15:05dada0d9
import logging
from aiogram import types, Dispatcher
from aiogram.dispatcher import FSMContext

from src.auth.auth_store import authorize_user, block_user
from src.scenarios.base_scenario import BaseScenario

logger = logging.getLogger(__name__)


class Access(BaseScenario):
    """
    Обработка доступа пользователей через кнопки AuthorizeKeyboard с JSON-хранилищем.

    Методы:
        process: Абстрактный метод для совместимости, не используется напрямую.
        _extract_user_id: Извлекает ID пользователя из текста сообщения администратора.
        _process_user: Универсальная обработка авторизации или блокировки пользователя.
        authorize_process: Обработка колбэка для авторизации пользователя.
        decline_process: Обработка колбэка для отклонения пользователя.
        register: Регистрирует обработчики колбэков в диспетчере.
    """

    def __init__(self, bot):
        """
        Инициализация класса Access.

        Args:
            bot: Экземпляр бота aiogram
        """
        super().__init__(bot)
        self.bot = bot

    async def process(self, message: types.Message, state: FSMContext, **kwargs):
        """
        Абстрактный метод для совместимости со схемой BaseScenario.
        Не используется, оставлен для интерфейса.

        Raises:
            NotImplementedError: Всегда
        """
        raise NotImplementedError("Используйте authorize_process и decline_process для обработки колбэков.")

    @staticmethod
    def _extract_user_id(text: str) -> int:
        """
        Извлекает ID пользователя из текста сообщения администратора.

        Args:
            text (str): Текст сообщения администратора, содержащий ID пользователя

        Returns:
            int: ID пользователя
        """
        return int(text.split('id: ')[1].split(')')[0])

    async def _process_user(self, callback_query: types.CallbackQuery, action: str) -> None:
        """
        Универсальная обработка авторизации или блокировки пользователя.

        Args:
            callback_query (types.CallbackQuery): CallbackQuery от администратора
            action (str): 'authorize' для авторизации, 'decline' для блокировки

        Returns:
            None
        """
        admin_id = callback_query.from_user.id
        try:
            user_id = self._extract_user_id(callback_query.message.text)
            if action == 'authorize':
                authorize_user(user_id)
                await self.bot.send_message(user_id, 'Доступ получен. Отправьте /start для работы с ботом.')
                await callback_query.message.edit_text(f'Пользователь {user_id} успешно авторизован.')
                logger.info(f'Администратор {admin_id} авторизовал пользователя {user_id}')
            else:
                block_user(user_id)
                await self.bot.send_message(user_id, 'Доступ запрещён администратором.')
                await callback_query.message.edit_text(f'Пользователь {user_id} отклонён и заблокирован.')
                logger.info(f'Администратор {admin_id} отклонил пользователя {user_id}')
        except Exception as e:
            error_msg = f'Ошибка при {action} пользователя: {e}'
            logger.error(error_msg, exc_info=True)
            await callback_query.message.reply(error_msg)

    async def authorize_process(self, callback_query: types.CallbackQuery, state: FSMContext, **kwargs) -> None:
        """
        Обработка колбэка для авторизации пользователя.

        Args:
            callback_query (types.CallbackQuery): CallbackQuery от администратора
            state (FSMContext): FSM контекст
        """
        await self._process_user(callback_query, action='authorize')

    async def decline_process(self, callback_query: types.CallbackQuery, state: FSMContext, **kwargs) -> None:
        """
        Обработка колбэка для отклонения пользователя.

        Args:
            callback_query (types.CallbackQuery): CallbackQuery от администратора
            state (FSMContext): FSM контекст
        """
        await self._process_user(callback_query, action='decline')

    def register(self, dp: Dispatcher) -> None:
        """
        Регистрирует обработчики колбэков в диспетчере.

        Args:
            dp (Dispatcher): Диспетчер aiogram
        """
        dp.register_callback_query_handler(self.authorize_process, lambda c: c.data == 'authorize_yes', state='*')
        dp.register_callback_query_handler(self.decline_process, lambda c: c.data == 'authorize_no', state='*')