Управление состояниями и FSM

Проблема: Пользователь начал регистрацию: ввёл имя, но бот не знает, что делать с следующим сообщением. Это email? Возраст? Комментарий? Без контекста каждое сообщение обрабатывается изолированно.

Типичные ошибки:

Хранение состояния в переменной — теряется между сообщениями

Проверка "последней команды" в каждом обработчике — дублирование кода

Нет способа отменить многошаговый процесс

Решение: FSM (Finite State Machine) хранит состояние диалога между сообщениями. Бот "помнит", на каком этапе находится пользователь, и обрабатывает сообщения соответственно.

Зачем это нужно: Любая многошаговая форма (регистрация, заказ, опрос) требует FSM. Без неё код превращается в спагетти из проверок.

#Что такое FSM

FSM (Finite State Machine, конечный автомат) — модель, позволяющая боту "помнить" текущий этап диалога с пользователем.

#Проблема без FSM

# ❌ Так не делайте: хранение состояния в переменной
user_state = {}  # Потеряется при перезапуске бота!

@router.message(Command('register'))
async def cmd_register(msg: Message):
    user_state[msg.from_user.id] = 'name'
    await msg.answer('Введите имя:')

@router.message()
async def any_message(msg: Message):
    # ⚠️ Проблема: user_state может не существовать
    # ⚠️ Проблема: нет структуры для данных
    if user_state.get(msg.from_user.id) == 'name':
        name = msg.text
        # ...

Проблемы такого подхода:

Состояние теряется при перезапуске бота
Нет персистентного хранения данных формы
Невозможно масштабировать на несколько инстансов

#Решение с FSM

from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext


class Registration(StatesGroup):
    """Сценарий регистрации пользователя."""
    name = State()
    email = State()
    age = State()


@router.message(Command('register'))
async def cmd_register(msg: Message, state: FSMContext):
    """
    Начало регистрации.
    
    💡 Зачем state.set_state(): Бот "запоминает", что пользователь
    начал регистрацию и ждёт ввода имени.
    """
    await state.set_state(Registration.name)
    await msg.answer('Введите имя:')


@router.message(Registration.name)
async def process_name(msg: Message, state: FSMContext):
    """
    Обработка имени.
    
    💡 Зачем: Фильтр Registration.name гарантирует, что
    обработчик сработает только для пользователей в состоянии name.
    """
    # Сохраняем данные в состоянии
    await state.update_data(name=msg.text)
    
    # Переходим к следующему шагу
    await state.set_state(Registration.email)
    await msg.answer('Введите email:')

💡 Зачем StatesGroup: Группировка состояний логически связывает этапы одного сценария.

#StatesGroup и State

StatesGroup — группа состояний для одного сценария. State — отдельное состояние внутри группы.

#Объявление состояний

from aiogram.fsm.state import State, StatesGroup


class Registration(StatesGroup):
    """
    Сценарий регистрации пользователя.
    
    💡 Зачем:
    - name → email → age → confirm — линейный сценарий
    - Можно вернуться на шаг назад
    - Можно отменить в любой момент
    """
    name = State()
    email = State()
    age = State()
    confirm = State()


class OrderForm(StatesGroup):
    """
    Сценарий оформления заказа.
    
    💡 Зачем: Отдельная группа для другого сценария.
    Состояния не пересекаются с Registration.
    """
    product = State()
    quantity = State()
    address = State()
    payment = State()


class FeedbackForm(StatesGroup):
    """Сценарий обратной связи."""
    rating = State()
    comment = State()
    contact = State()

⚠️ Антипаттерн: Не используйте одно состояние для разных сценариев. name в Registration и name в ProfileForm — это разные состояния.

#Вложенные группы

class Profile(StatesGroup):
    """
    Вложенные группы для сложных сценариев.
    
    💡 Зачем: Логическая группировка внутри сценария.
    """
    class Personal(StatesGroup):
        name = State()
        birthday = State()

    class Contact(StatesGroup):
        email = State()
        phone = State()


# Использование
@router.message(Profile.Personal.name)
async def process_name(msg: Message, state: FSMContext):
    ...

💡 Совет: Избегайте вложенности глубже 2 уровней. Это усложняет отладку.

#FSMContext

FSMContext — объект для управления состоянием пользователя.

#Основные методы

from aiogram.fsm.context import FSMContext


@router.message(Command('start'))
async def cmd_start(msg: Message, state: FSMContext):
    """
    Управление состоянием.
    
    💡 Методы:
    - set_state(): установить состояние
    - get_state(): получить текущее
    - update_data(): сохранить данные
    - get_data(): получить все данные
    - clear(): очистить состояние и данные
    """
    # Установить состояние
    await state.set_state(Registration.name)

    # Получить текущее состояние
    current_state = await state.get_state()
    # 'Registration:name'

    # Очистить состояние (завершить сценарий)
    await state.set_state(None)

    # Очистить все данные
    await state.clear()


@router.message(Registration.name)
async def process_name(msg: Message, state: FSMContext):
    # Сохранить данные
    await state.update_data(name=msg.text)

    # Получить все данные
    data = await state.get_data()
    # {'name': 'Иван'}

    # Получить конкретное поле
    name = data.get('name')

    # Изменить данные
    await state.update_data(name='Алексей')

    # Удалить поле (установить None)
    await state.update_data(name=None)

⚠️ Важно: update_data() не перезаписывает все данные, а обновляет указанные поля.

#Пример: полная регистрация

from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command, StateFilter
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup

from bot.keyboards.inline import build_confirm_keyboard

router = Router()


class Registration(StatesGroup):
    name = State()
    email = State()
    age = State()
    confirm = State()


@router.message(Command('register'))
async def cmd_register(msg: Message, state: FSMContext):
    """Начало регистрации."""
    await state.set_state(Registration.name)
    await msg.answer('📝 Введите ваше имя:')


@router.message(Registration.name, F.text.func(lambda x: len(x) >= 2))
async def process_name(msg: Message, state: FSMContext):
    """
    Обработка имени.
    
    ⚠️ Важно: Фильтр длины текста отсеивает невалидный ввод.
    Обработчик не сработает для имени из 1 символа.
    """
    await state.update_data(name=msg.text)
    await state.set_state(Registration.email)
    await msg.answer('📧 Введите email:')


@router.message(Registration.name)  # Слишком короткое имя
async def invalid_name(msg: Message):
    """
    Обработчик для невалидного имени.
    
    💡 Зачем: Второй обработчик срабатывает, если первый не прошёл фильтр.
    """
    await msg.answer('Имя должно быть не менее 2 символов. Попробуйте снова:')


@router.message(Registration.email, F.text.contains('@'))
async def process_email(msg: Message, state: FSMContext):
    """Обработка email."""
    await state.update_data(email=msg.text)
    await state.set_state(Registration.age)
    await msg.answer('🎂 Введите возраст:')


@router.message(Registration.email)  # Неверный email
async def invalid_email(msg: Message):
    await msg.answer('Email должен содержать @. Попробуйте снова:')


@router.message(Registration.age, F.text.func(lambda x: x.isdigit()))
async def process_age(msg: Message, state: FSMContext):
    """Обработка возраста."""
    age = int(msg.text)

    if age < 1 or age > 120:
        await msg.answer('Возраст должен быть от 1 до 120. Попробуйте снова:')
        return

    await state.update_data(age=age)
    await state.set_state(Registration.confirm)

    # Показываем подтверждение
    data = await state.get_data()
    text = f"""
Проверьте данные:

Имя: {data['name']}
Email: {data['email']}
Возраст: {data['age']}
"""
    await msg.answer(text, reply_markup=build_confirm_keyboard())


@router.message(Registration.confirm, F.text == 'Подтвердить')
async def confirm_registration(msg: Message, state: FSMContext, user_repo):
    """
    Подтверждение регистрации.
    
    💡 Зачем state.clear(): Завершаем сценарий, очищаем память.
    """
    data = await state.get_data()

    # Создаём пользователя в БД
    await user_repo.create(
        telegram_id=msg.from_user.id,
        username=msg.from_user.username,
        name=data['name'],
        email=data['email'],
        age=data['age']
    )

    await state.clear()
    await msg.answer('✅ Регистрация завершена!')


@router.message(Registration.confirm, F.text == 'Изменить')
async def edit_registration(msg: Message, state: FSMContext):
    """Возврат к редактированию."""
    await state.set_state(Registration.name)
    await msg.answer('✏️ Введите новое имя:')


@router.message(Registration.confirm, F.text == 'Отмена')
async def cancel_registration(msg: Message, state: FSMContext):
    """Отмена регистрации."""
    await state.clear()
    await msg.answer('❌ Регистрация отменена')

⚠️ Антипаттерн: Не забывайте state.clear() после завершения сценария. Иначе пользователь останется в состоянии и следующие сообщения будут обрабатываться неправильно.

#Хранение состояний

#MemoryStorage (для тестов)

from aiogram.fsm.storage.memory import MemoryStorage

storage = MemoryStorage()
dp = Dispatcher(storage=storage)

Плюсы	Минусы
Не требует внешних зависимостей	Данные теряются при перезапуске
Быстрое	Не работает при нескольких инстансах бота
Удобно для локальной разработки	Не для продакшена

💡 Когда использовать: Локальная разработка, тесты, прототипы.

#RedisStorage (для продакшена)

from redis.asyncio import Redis
from aiogram.fsm.storage.redis import RedisStorage

redis = Redis.from_url('redis://localhost:6379/0')
storage = RedisStorage(redis=redis, key_prefix='fsm')
dp = Dispatcher(storage=storage)

Плюсы	Минусы
Персистентное хранение	Требует Redis
Работает с несколькими инстансами	Дополнительная инфраструктура
Быстрый доступ

💡 Зачем key_prefix: Изоляция ключей в Redis для разных ботов/окружений.

#Кастомное хранилище

from aiogram.fsm.storage.base import BaseStorage, StateType, StorageKey
from typing import Dict, Any, Optional
from sqlalchemy.ext.asyncio import AsyncSession


class DatabaseStorage(BaseStorage):
    """
    Хранение состояний в БД.
    
    💡 Зачем:
    - Не нужен Redis
    - Данные в той же БД, что и пользователи
    - Бэкапы автоматически
    
    ⚠️ Антипаттерн: Не используйте для высоконагруженных ботов.
    Redis быстрее для частых операций записи/чтения.
    """

    def __init__(self, session_factory):
        self.session_factory = session_factory

    async def set_state(self, key: StorageKey, state: StateType = None) -> None:
        async with self.session_factory() as session:
            # Сохраняем состояние в БД
            ...

    async def get_state(self, key: StorageKey) -> Optional[str]:
        async with self.session_factory() as session:
            # Получаем состояние из БД
            ...

    async def set_data(self, key: StorageKey, data: Dict[str, Any]) -> None:
        async with self.session_factory() as session:
            # Сохраняем данные в БД
            ...

    async def get_data(self, key: StorageKey) -> Dict[str, Any]:
        async with self.session_factory() as session:
            # Получаем данные из БД
            ...

#Сценарии с ветвлением

Проблема: В зависимости от ответа пользователя нужно идти по разным веткам сценария.

Решение: Условная логика в обработчиках для перехода к разным состояниям.

from aiogram.fsm.state import State, StatesGroup
from aiogram.types import CallbackQuery


class Survey(StatesGroup):
    """
    Опрос с ветвлением.
    
    💡 Зачем: Разные вопросы для разных возрастных групп.
    """
    age_group = State()
    experience = State()
    skills = State()
    goals = State()


@router.callback_query(F.data == 'start_survey')
async def start_survey(cb: CallbackQuery, state: FSMContext):
    await state.set_state(Survey.age_group)
    await cb.message.edit_text(
        'Выберите возраст:',
        reply_markup=build_age_keyboard()
    )


@router.callback_query(Survey.age_group, F.data.startswith('age_'))
async def process_age(cb: CallbackQuery, state: FSMContext):
    """
    Обработка возраста с ветвлением.
    
    💡 Зачем: Молодым пользователям не задаём вопрос об опыте.
    """
    age_group = cb.data.split('_')[1]
    await state.update_data(age_group=age_group)

    # Ветвление: разные вопросы для разных возрастов
    if age_group in ['under_18', '18_25']:
        # Молодые → сразу к навыкам
        await state.set_state(Survey.skills)
        await cb.message.edit_text(
            'Какие навыки вы хотите развить?',
            reply_markup=build_skills_keyboard()
        )
    else:
        # Старшие → вопрос об опыте
        await state.set_state(Survey.experience)
        await cb.message.edit_text(
            'Ваш опыт работы?',
            reply_markup=build_experience_keyboard()
        )


@router.callback_query(Survey.experience, F.data.startswith('exp_'))
async def process_experience(cb: CallbackQuery, state: FSMContext):
    """Продолжение после опыта."""
    await state.update_data(experience=cb.data.split('_')[1])
    await state.set_state(Survey.skills)
    await cb.message.edit_text(
        'Какие навыки вы хотите развить?',
        reply_markup=build_skills_keyboard()
    )

💡 Совет: Визуализируйте сценарий на бумаге перед реализацией. Это поможет выявить сложные ветвления.

#Таймауты состояний

Проблема: Пользователь начал регистрацию и исчез. Состояние остаётся в памяти бесконечно.

Решение: Таймаут сбрасывает состояние после периода неактивности.

import asyncio
from aiogram.fsm.context import FSMContext


async def state_timeout(state: FSMContext, delay: int):
    """
    Сбрасывает состояние через delay секунд.
    
    ⚠️ Антипаттерн: asyncio.create_task() без обработки ошибок.
    Задача может упасть незаметно.
    """
    await asyncio.sleep(delay)
    current_state = await state.get_state()
    if current_state:
        await state.clear()
        # Уведомить пользователя
        ...


@router.message(Command('order'))
async def cmd_order(msg: Message, state: FSMContext):
    await state.set_state(OrderForm.product)
    await msg.answer('Выберите товар (у вас 5 минут):')

    # Запускаем таймаут
    # ⚠️ Важно: В продакшене используйте Celery (см. "Масштабирование")
    asyncio.create_task(state_timeout(state, 300))

⚠️ Антипаттерн: Не полагайтесь на asyncio.create_task() в продакшене. Используйте Celery beat или Redis key expiry для надёжности.

#Машина состояний для оплаты

class Payment(StatesGroup):
    """
    Сценарий оплаты.
    
    💡 Зачем:
    - method → card_number → confirm — линейный процесс
    - Разные пути для карты и криптовалюты
    - Возможность отмены на любом этапе
    """
    method = State()
    card_number = State()
    confirm = State()


@router.callback_query(F.data == 'pay')
async def start_payment(cb: CallbackQuery, state: FSMContext):
    await state.set_state(Payment.method)
    await cb.message.edit_text(
        'Выберите способ оплаты:',
        reply_markup=build_payment_keyboard()
    )


@router.callback_query(Payment.method, F.data == 'card')
async def choose_card(cb: CallbackQuery, state: FSMContext):
    await state.set_state(Payment.card_number)
    await cb.message.edit_text(
        'Введите номер карты (или нажмите /cancel):'
    )


@router.callback_query(Payment.method, F.data == 'crypto')
async def choose_crypto(cb: CallbackQuery, state: FSMContext):
    """
    Альтернативный путь для криптовалюты.
    
    💡 Зачем: Пропускаем ввод карты, сразу к подтверждению.
    """
    await state.update_data(method='crypto')
    await state.set_state(Payment.confirm)
    await cb.message.edit_text(
        'Отправляем на криптокошелёк...\n'
        'Адрес: 0x1234567890abcdef\n\n'
        'Подтвердить оплату?',
        reply_markup=build_confirm_keyboard()
    )


@router.message(Payment.card_number, F.text.func(lambda x: len(x) == 16 and x.isdigit()))
async def process_card(msg: Message, state: FSMContext):
    """Валидация номера карты."""
    await state.update_data(card_number=msg.text)
    await state.set_state(Payment.confirm)
    await msg.answer(
        f'Оплата картой {msg.text[-4:]}.\nПодтвердить?',
        reply_markup=build_confirm_keyboard()
    )


@router.message(Payment.card_number)
async def invalid_card(msg: Message):
    """Неверный номер карты."""
    await msg.answer('Неверный номер карты. Попробуйте снова:')


@router.callback_query(Payment.confirm, F.data == 'confirm_payment')
async def confirm_payment(cb: CallbackQuery, state: FSMContext, payment_service):
    """
    Подтверждение оплаты.
    
    ⚠️ Важно: state.clear() только после успешной оплаты.
    При ошибке сохраняем состояние для повторной попытки.
    """
    data = await state.get_data()

    try:
        await payment_service.process_payment(
            user_id=cb.from_user.id,
            method=data.get('method', 'card'),
            card_number=data.get('card_number')
        )

        await state.clear()
        await cb.message.edit_text('✅ Оплата прошла успешно!')
    except PaymentError as e:
        await cb.message.edit_text(f'❌ Ошибка оплаты: {e}')
        # Не очищаем состояние — пользователь может попробовать снова

#Отмена состояния

Проблема: Пользователь передумал заполнять форму. Как выйти из сценария?

Решение: Универсальный обработчик команды /cancel для любого состояния.

from aiogram.filters import Command


@router.message(Command('cancel'), ~StateFilter(None))
async def cancel_state(msg: Message, state: FSMContext):
    """
    Отмена текущего состояния.
    
    💡 Зачем ~StateFilter(None): Срабатывает только если есть активное состояние.
    """
    current_state = await state.get_state()

    if current_state:
        await state.clear()
        await msg.answer('❌ Действие отменено')


@router.callback_query(F.data == 'cancel', ~StateFilter(None))
async def cancel_callback(cb: CallbackQuery, state: FSMContext):
    """Отмена через callback."""
    await state.clear()
    await cb.message.edit_text('❌ Действие отменено')

💡 Совет: Добавьте кнопку "Отмена" в клавиатуру каждого шага многошаговой формы.

#Middleware для автосброса состояния

from aiogram import BaseMiddleware
from aiogram.types import Update
from aiogram.fsm.context import FSMContext


class StateTimeoutMiddleware(BaseMiddleware):
    """
    Сбрасывает состояние после 30 минут неактивности.
    
    💡 Зачем:
    - Очистка "зависших" состояний
    - Освобождение памяти
    
    ⚠️ Антипаттерн: Не используйте для критичных сценариев (оплата).
    Там нужен явный контроль таймаута.
    """

    async def __call__(
        self,
        handler,
        event: Update,
        data: dict
    ):
        state: FSMContext = data.get('state')
        if state:
            last_activity = await state.get_data().get('last_activity')
            if last_activity:
                import time
                if time.time() - last_activity > 1800:  # 30 минут
                    await state.clear()
                    await event.message.answer(
                        'Сессия истекла. Начните сначала.'
                    )
                    return

            await state.update_data(last_activity=time.time())

        return await handler(event, data)

#Итоги

#✅ Чек-лист: готово ли ваше FSM к продакшену?

StatesGroup для каждого сценария
RedisStorage для продакшена (не MemoryStorage)
state.clear() после завершения сценария
Обработчик /cancel для выхода из любого состояния
Валидация ввода на каждом шаге
Таймауты для длительных сценариев
Данные формы сохраняются через update_data()
Ветвления сценария документированы

#🔗 Связь с другими темами

Тема	Как связана
Обработчики	StateFilter для фильтрации по состоянию
Архитектура	Storage настраивается в dispatcher
Базы данных	Кастомное хранилище в БД
Масштабирование	Redis для shared state между инстансами
Inline-клавиатуры	Навигация в многошаговых формах

#📚 Что дальше?

→ Базы данных — сохранение данных формы в БД
→ Масштабирование — Celery для таймаутов и отложенных задач
→ Тестирование — тесты FSM-сценариев

Плюсы

Минусы

Не требует внешних зависимостей

Данные теряются при перезапуске

Быстрое

Не работает при нескольких инстансах бота

Удобно для локальной разработки

Не для продакшена

Плюсы

Минусы

Персистентное хранение

Требует Redis

Работает с несколькими инстансами

Дополнительная инфраструктура

Быстрый доступ

Тема

Как связана

Обработчики

StateFilter для фильтрации по состоянию

Архитектура

Storage настраивается в dispatcher

Базы данных

Кастомное хранилище в БД

Масштабирование

Redis для shared state между инстансами

Inline-клавиатуры

Навигация в многошаговых формах