diff --git a/bot/utils/state_utils.py b/bot/utils/state_utils.py new file mode 100644 index 0000000..3c87674 --- /dev/null +++ b/bot/utils/state_utils.py @@ -0,0 +1,650 @@ +""" +Утилиты для работы с FSM состояниями и обновлениями +""" +from typing import Optional, Any, Set, Union +from contextlib import suppress + +from aiogram.fsm.context import FSMContext +from aiogram.fsm.state import State +from aiogram.types import CallbackQuery, Message, ReplyKeyboardRemove +from aiogram.exceptions import TelegramBadRequest + +from middleware.loggers import logger + +__all__ = ( + 'clear_state', + 'answer_callback', + 'safe_answer_callback', + 'safe_delete_message', + 'safe_edit_message', + 'clear_state_keep_data', + 'get_state_data', + 'set_state_data', + 'update_state_data', + 'is_state_active', + 'inline_clear', + 'status_clear', + 'delete_messages', + 'set_state_with_data', + 'get_or_create_data', + 'increment_state_value', + 'append_to_state_list', + 'remove_from_state_list', + 'toggle_state_flag', + 'debug_state' +) + + +# ================= РАБОТА С FSM СОСТОЯНИЯМИ ================= + +async def clear_state( + state: FSMContext, + log: bool = True +) -> None: + """ + Очищает FSM состояние. + + Args: + state: Контекст FSM + log: Логировать очистку + + Example: + >> await clear_state(state) + """ + current_state = await state.get_state() + + if log and current_state: + logger.debug( + f"Очистка FSM состояния: {current_state}", + log_type='FSM' + ) + + await state.clear() + + +async def clear_state_keep_data( + state: FSMContext, + keep_keys: Optional[Set[str]] = None +) -> None: + """ + Очищает FSM состояние, но сохраняет определенные данные. + + Args: + state: Контекст FSM + keep_keys: Множество ключей для сохранения + + Example: + >> # Очищаем состояние, но сохраняем user_id и language + >> await clear_state_keep_data(state, keep_keys={'user_id', 'language'}) + """ + if keep_keys: + # Получаем текущие данные + current_data = await state.get_data() + + # Сохраняем только нужные ключи + saved_data = { + key: value for key, value in current_data.items() + if key in keep_keys + } + + # Очищаем состояние + await state.clear() + + # Восстанавливаем сохраненные данные + if saved_data: + await state.update_data(**saved_data) + + logger.debug( + f"FSM очищен, сохранены ключи: {', '.join(keep_keys)}", + log_type='FSM' + ) + else: + await state.clear() + + +async def get_state_data( + state: FSMContext, + key: Optional[str] = None, + default: Any = None +) -> Any: + """ + Получает данные из FSM состояния. + + Args: + state: Контекст FSM + key: Ключ для получения (если None, возвращает все данные) + default: Значение по умолчанию + + Returns: + Any: Данные из состояния + + Example: + >> # Получить все данные + >> data = await get_state_data(state) + + >> # Получить конкретный ключ + >> user_id = await get_state_data(state, 'user_id') + + >> # С значением по умолчанию + >> lang = await get_state_data(state, 'language', default='ru') + """ + data = await state.get_data() + + if key is None: + return data + + return data.get(key, default) + + +async def set_state_data( + state: FSMContext, + key: str, + value: Any +) -> None: + """ + Устанавливает данные в FSM состояние. + + Args: + state: Контекст FSM + key: Ключ + value: Значение + + Example: + >> await set_state_data(state, 'user_id', 123456789) + """ + await state.update_data(**{key: value}) + + +async def update_state_data( + state: FSMContext, + **kwargs +) -> None: + """ + Обновляет несколько полей в FSM состоянии. + + Args: + state: Контекст FSM + **kwargs: Пары ключ-значение для обновления + + Example: + >> await update_state_data( + ... state, + ... user_id=123456789, + ... language='ru', + ... step=1 + ... ) + """ + await state.update_data(**kwargs) + + +async def is_state_active(state: FSMContext) -> bool: + """ + Проверяет, активно ли какое-либо состояние. + + Args: + state: Контекст FSM + + Returns: + bool: True если есть активное состояние + + Example: + >> if await is_state_active(state): + ... await message.answer("У вас есть незавершенное действие") + """ + current_state = await state.get_state() + return current_state is not None + + +# ================= РАБОТА С CALLBACK QUERIES ================= + +async def answer_callback( + callback: CallbackQuery, + text: Optional[str] = None, + show_alert: bool = False, + cache_time: int = 0 +) -> bool: + """ + Отвечает на callback query. + + Args: + callback: Callback query + text: Текст уведомления + show_alert: Показать как alert + cache_time: Время кэширования + + Returns: + bool: True если успешно + + Example: + >> await answer_callback(callback, "✅ Готово!") + >> await answer_callback(callback, "⚠️ Ошибка", show_alert=True) + """ + try: + await callback.answer(text=text, show_alert=show_alert, cache_time=cache_time) + return True + except TelegramBadRequest as e: + logger.warning( + f"Не удалось ответить на callback: {e}", + log_type='CALLBACK' + ) + return False + + +async def safe_answer_callback( + callback: CallbackQuery, + text: Optional[str] = None, + show_alert: bool = False +) -> None: + """ + Безопасно отвечает на callback query (подавляет ошибки). + + Args: + callback: Callback query + text: Текст уведомления + show_alert: Показать как alert + + Example: + >> await safe_answer_callback(callback, "✅ Готово!") + """ + with suppress(TelegramBadRequest): + await callback.answer(text=text, show_alert=show_alert) + + +# ================= РАБОТА С СООБЩЕНИЯМИ ================= + +async def safe_delete_message( + message: Message, + log: bool = False +) -> bool: + """ + Безопасно удаляет сообщение. + + Args: + message: Сообщение для удаления + log: Логировать попытку удаления + + Returns: + bool: True если успешно удалено + + Example: + >> await safe_delete_message(message) + """ + try: + await message.delete() + + if log: + logger.debug( + f"Сообщение удалено: {message.message_id}", + log_type='MESSAGE' + ) + + return True + except TelegramBadRequest as e: + if log: + logger.warning( + f"Не удалось удалить сообщение: {e}", + log_type='MESSAGE' + ) + return False + + +async def safe_edit_message( + message: Message, + text: str, + **kwargs +) -> bool: + """ + Безопасно редактирует сообщение. + + Args: + message: Сообщение для редактирования + text: Новый текст + **kwargs: Дополнительные параметры (reply_markup, parse_mode, и т.д.) + + Returns: + bool: True если успешно отредактировано + + Example: + >> await safe_edit_message( + ... message, + ... "Новый текст", + ... parse_mode="HTML" + ... ) + """ + try: + await message.edit_text(text, **kwargs) + return True + except TelegramBadRequest as e: + logger.warning( + f"Не удалось отредактировать сообщение: {e}", + log_type='MESSAGE' + ) + return False + + +async def delete_messages( + chat_id: int, + message_ids: list[int], + bot +) -> int: + """ + Удаляет несколько сообщений. + + Args: + chat_id: ID чата + message_ids: Список ID сообщений + bot: Экземпляр бота + + Returns: + int: Количество успешно удаленных сообщений + + Example: + >> deleted = await delete_messages( + ... chat_id=message.chat.id, + ... message_ids=[123, 124, 125], + ... bot=bot + ... ) + >> print(f"Удалено {deleted} сообщений") + """ + deleted_count = 0 + + for message_id in message_ids: + try: + await bot.delete_message(chat_id=chat_id, message_id=message_id) + deleted_count += 1 + except TelegramBadRequest: + pass + + return deleted_count + + +# ================= КОМБИНИРОВАННЫЕ ФУНКЦИИ ================= + +async def inline_clear(update: Union[Message, CallbackQuery]) -> None: + """ + Очищает все инлайн взаимодействия (отвечает на callback). + + Args: + update: Объект обновления (Message или CallbackQuery) + + Example: + >> await inline_clear(callback) + """ + if isinstance(update, CallbackQuery): + await safe_answer_callback(update) + + +async def status_clear( + update: Union[Message, CallbackQuery], + state: FSMContext, + keep_data: Optional[Set[str]] = None, + remove_keyboard: bool = False +) -> None: + """ + Полная очистка: состояние FSM + ответ на callback + удаление клавиатуры. + + Args: + update: Объект обновления + state: Контекст FSM + keep_data: Данные для сохранения + remove_keyboard: Удалить клавиатуру (только для Message) + + Example: + >> # Полная очистка + >> await status_clear(message, state) + + >> # С сохранением данных + >> await status_clear( + ... callback, + ... state, + ... keep_data={'user_id', 'language'} + ... ) + + >> # С удалением клавиатуры + >> await status_clear(message, state, remove_keyboard=True) + """ + # Очищаем состояние + if keep_data: + await clear_state_keep_data(state, keep_keys=keep_data) + else: + await clear_state(state, log=True) + + # Отвечаем на callback + await inline_clear(update) + + # Удаляем клавиатуру если нужно + if remove_keyboard and isinstance(update, Message): + with suppress(TelegramBadRequest): + await update.answer( + "Отменено", + reply_markup=ReplyKeyboardRemove() + ) + + +# ================= УТИЛИТЫ ДЛЯ РАБОТЫ С СОСТОЯНИЯМИ ================= + +async def set_state_with_data( + state: FSMContext, + new_state: State, + **data +) -> None: + """ + Устанавливает новое состояние и данные одновременно. + + Args: + state: Контекст FSM + new_state: Новое состояние + **data: Данные для сохранения + + Example: + >> await set_state_with_data( + ... state, + ... FormStates.waiting_name, + ... user_id=123456789, + ... step=1 + ... ) + """ + await state.set_state(new_state) + if data: + await state.update_data(**data) + + logger.debug( + f"Установлено состояние: {new_state.state}", + log_type='FSM' + ) + + +async def get_or_create_data( + state: FSMContext, + key: str, + factory: Any +) -> Any: + """ + Получает данные из состояния или создает их если их нет. + + Args: + state: Контекст FSM + key: Ключ данных + factory: Значение по умолчанию или функция для создания + + Returns: + Any: Данные из состояния или созданные + + Example: + >> # С простым значением + >> items = await get_or_create_data(state, 'items', []) + + >> # С функцией + >> data = await get_or_create_data(state, 'data', lambda: {'count': 0}) + """ + data = await state.get_data() + + if key not in data: + # Создаем значение + if callable(factory): + value = factory() + else: + value = factory + + await state.update_data(**{key: value}) + return value + + return data[key] + + +async def increment_state_value( + state: FSMContext, + key: str, + amount: int = 1 +) -> int: + """ + Инкрементирует числовое значение в состоянии. + + Args: + state: Контекст FSM + key: Ключ значения + amount: Величина инкремента + + Returns: + int: Новое значение + + Example: + >> # Увеличиваем счетчик + >> new_count = await increment_state_value(state, 'attempts') + >> if new_count >= 3: + ... await message.answer("Слишком много попыток!") + """ + data = await state.get_data() + current = data.get(key, 0) + new_value = current + amount + + await state.update_data(**{key: new_value}) + return new_value + + +async def append_to_state_list( + state: FSMContext, + key: str, + value: Any +) -> list: + """ + Добавляет значение в список в состоянии. + + Args: + state: Контекст FSM + key: Ключ списка + value: Значение для добавления + + Returns: + list: Обновленный список + + Example: + >> # Добавляем товар в корзину + >> cart = await append_to_state_list(state, 'cart', product_id) + >> await message.answer(f"В корзине {len(cart)} товаров") + """ + data = await state.get_data() + current_list = data.get(key, []) + + if not isinstance(current_list, list): + current_list = [] + + current_list.append(value) + await state.update_data(**{key: current_list}) + + return current_list + + +async def remove_from_state_list( + state: FSMContext, + key: str, + value: Any +) -> list: + """ + Удаляет значение из списка в состоянии. + + Args: + state: Контекст FSM + key: Ключ списка + value: Значение для удаления + + Returns: + list: Обновленный список + + Example: + >> # Удаляем товар из корзины + >> cart = await remove_from_state_list(state, 'cart', product_id) + """ + data = await state.get_data() + current_list = data.get(key, []) + + if isinstance(current_list, list) and value in current_list: + current_list.remove(value) + await state.update_data(**{key: current_list}) + + return current_list + + +async def toggle_state_flag( + state: FSMContext, + key: str +) -> bool: + """ + Переключает boolean флаг в состоянии. + + Args: + state: Контекст FSM + key: Ключ флага + + Returns: + bool: Новое значение флага + + Example: + >> # Переключаем режим + >> is_active = await toggle_state_flag(state, 'notifications') + >> await message.answer( + ... f"Уведомления: {'включены' if is_active else 'выключены'}" + ... ) + """ + data = await state.get_data() + current = data.get(key, False) + new_value = not current + + await state.update_data(**{key: new_value}) + return new_value + + +# ================= ОТЛАДКА ================= + +async def debug_state(state: FSMContext) -> str: + """ + Возвращает отладочную информацию о состоянии. + + Args: + state: Контекст FSM + + Returns: + str: Форматированная информация о состоянии + + Example: + >> debug_info = await debug_state(state) + >> print(debug_info) + """ + current_state = await state.get_state() + data = await state.get_data() + + lines = [ + "🔍 Debug FSM:\n", + f"📊 Состояние: {current_state or 'None'}\n", + f"📦 Данных: {len(data)}\n" + ] + + if data: + lines.append("\nДанные:") + for key, value in data.items(): + value_str = str(value) + if len(value_str) > 50: + value_str = value_str[:50] + "..." + lines.append(f"• {key}: {value_str}") + + return "\n".join(lines)