Telegram Боты на Aiogram 3.x: Все про FSM простыми словами

Друзья, приветствую! Сегодня мы коснемся важной и не такой уж сложной темы — реализации конечных автоматов состояний (FSM) в телеграм-ботах на Aiogram 3.x.

Для лучшего понимания рекомендую ознакомиться с моими предыдущими публикациями на тему Aiogram 3.x:

• Telegram Боты на Aiogram 3.x: Первые Шаги

• Telegram Боты на Aiogram 3.x: Текстовая клавиатура и командное меню

• Telegram Боты на Aiogram 3.x: Инлайн кнопки и CallBack Дата

• Telegram Боты на Aiogram 3.x: Магия фильтров

• Работа с текстовыми сообщениями

• Работа с медиа

• Поднятие Redis сервера: Полное руководство (Redis будем использовать в качестве хранилища для FSM)

Также советую развернуть Redis сервер и использовать его в качестве хранилища данных для FSM. Но, если не хотите заморачиваться Redis, я покажу, как использовать его аналог (MemoryStorage) и расскажу почему этого лучше не делать.

Что мы будем делать сегодня?

Сегодня разберем FSM на конкретном примере: настроим анкету пользователя, захватим его логин, имя, возраст, фото и информацию о себе, а затем отобразим эти данные.

После изучения материала вы полностью овладеете навыком взаимодействия с FSM в Aiogram 3.x, и останется только научиться записывать эти данные в базу данных PostgreSQL (этим займемся в следующей статье).

Установка необходимых модулей

Для начала установим все необходимые модули:

pip install aiogram python-decouple redis

• redis — для взаимодействия с базой данных Redis.

• python-decouple — для работы с .env файлами.

• aiogram — библиотека для создания ботов.

Доступ к Redis

• Хост

• Порт

• Номер базы данных (по умолчанию от 0 до 15)

• (Опционально) Логин и пароль, если они заданы

Пример строки подключения к Redis без логина и пароля:

redis://HOST:PORT/NUM_DB

Пример строки подключения к Redis с логином и паролем:

redis://LOGIN:PASSWORD@HOST:PORT/NUM_DB

Настройка create_bot.py

В файле create_bot.py создадим объект storage для хранения данных FSM. Импортируем переменные из .env файла:

from decouple import config
redis_url = config('REDIS_URL')

Настройка Storage

Импортируем модуль RedisStorage:

from aiogram.fsm.storage.redis import RedisStorage

Это импортирует класс RedisStorage из библиотеки aiogram, который используется для хранения данных конечного автомата состояний (FSM) в Redis.

Создаем объект RedisStorage:

storage = RedisStorage.from_url(config('REDIS_URL'))

Здесь мы создаем объект RedisStorage, используя URL подключения к Redis, который берем из переменной окружения REDIS_URL, загруженной с помощью config из библиотеки python-decouple.

Инициализация Dispatcher с RedisStorage:

dp = Dispatcher(storage=storage)

Мы создаем объект Dispatcher, передавая ему наш storage для хранения состояния конечного автомата в Redis. Это позволяет боту сохранять и восстанавливать состояния пользователей, используя Redis как хранилище.

Если вы не хотите использовать Redis в качестве хранилища, можно использовать MemoryStorage:

from aiogram.fsm.storage.memory import MemoryStorage


dp = Dispatcher(storage=MemoryStorage())

MemoryStorage использует оперативную память, что приводит к полной потере данных при любом сбое на сервере или в боте.

Redis так-же использует оперативную память для хранения данных, но в отличие от MemoryStorage, поддерживает периодическую запись данных на диск и может работать в кластерной среде, обеспечивая масштабируемость и надежность системы.

Таким образом, использование RedisStorage в FSM для Telegram-ботов обеспечивает высокую производительность и надежность, что делает его предпочтительным выбором для любых телеграмм ботов, но, в учебных целях, можете использовать MemoryStorage.

Сейчас приведу 2 примера, которые явно продемонстрируют отличия.

1. Запускаем некий сценарий анкетирования (всего 5 вопросов)

2. После третего вопроса перезагружаем бота

Если использовалась MemoryStorage , то все данные будут потеряны и сценарий нужно будет начинать сначала. При использовании Redis - сценарий для каждого пользователя продолжится с того места где тот остановился.

Что такое FSM?

FSM, или конечный автомат состояний (Finite State Machine), — это простой способ управлять сложными взаимодействиями в вашем Telegram боте. Он помогает боту "запомнить", на каком шаге процесса находится пользователь и что делать дальше.

К примеру создадим анкету, которая будет вести пользователя через следующие шаги:

1. Сначала спрашивает пол.

2. Затем возраст.

3. Потом имя и фамилию.

4. Далее логин.

5. Попросит отправить фото.

6. И наконец, попросит добавить описание о себе.

FSM помогает боту отслеживать, на каком из этих шагов находится пользователь, и что спросить дальше. Если пользователь отправил имя, бот "запоминает" это и знает, что следующий шаг — запросить логин.

Настройка первого скрипта FSM

Импортируем необходимые модули:

from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup

Определим состояния:

class Form(StatesGroup): 
    name = State()
    age = State()

Под каждое состояние напишем отдельные хендлеры, которые будут реагировать на ввод текста (имя и возраст):

import asyncio
from create_bot import bot
from aiogram import Router, F
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.types import Message
from aiogram.utils.chat_action import ChatActionSender
import re


def extract_number(text):
    match = re.search(r'\b(\d+)\b', text)
    if match:
        return int(match.group(1))
    else:
        return None

      
class Form(StatesGroup):
    name = State()
    age = State()

    
questionnaire_router = Router()


@questionnaire_router.message(Command('start_questionnaire'))
async def start_questionnaire_process(message: Message, state: FSMContext):
    async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
        await asyncio.sleep(2)
        await message.answer('Привет. Напиши как тебя зовут: ')
    await state.set_state(Form.name)

    
@questionnaire_router.message(F.text, Form.name)
async def capture_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
        await asyncio.sleep(2)
        await message.answer('Супер! А теперь напиши сколько тебе полных лет: ')
    await state.set_state(Form.age)

    
@questionnaire_router.message(F.text, Form.age)
async def capture_age(message: Message, state: FSMContext):
    check_age = extract_number(message.text)

    if not check_age or not (1 <= check_age <= 100):
        await message.reply('Пожалуйста, введите корректный возраст (число от 1 до 100).')
        return
    await state.update_data(age=check_age)

    data = await state.get_data()
    msg_text = (f'Вас зовут <b>{data.get("name")}</b> и вам <b>{data.get("age")}</b> лет. '
                f'Спасибо за то что ответили на мои вопросы.')
    await message.answer(msg_text)
    await state.clear()

Мы импортировали FSMContext из aiogram.dispatcher, а также State и StatesGroup из aiogram.dispatcher.filters.state для работы с конечным автоматом состояний (FSM) в нашем боте. Вот простое объяснение:

1. FSMContext:

   • Это специальный объект, который помогает нам управлять состояниями пользователя. Он хранит данные о текущем состоянии пользователя и позволяет изменять их, перемещая пользователя между различными состояниями.

   • Пример использования: С помощью FSMContext мы можем сохранить имя пользователя и затем перейти к следующему шагу, где спросим возраст.

2. State и StatesGroup:

   • State представляет собой конкретное состояние, в котором может находиться пользователь.

   • StatesGroup позволяет объединять несколько состояний в логическую группу.

   • Пример использования: Мы создаем класс Form, который наследуется от StatesGroup, и внутри него определяем состояния name и age. Это помогает нам структурировать и управлять последовательностью шагов анкеты.

С другими импортами вы уже знакомы, если читали мои прошлый статьи на тему aiogram

Функция extract_number извлекает число из текста. Полезно на случай если пользователь вместо "20" будет писать "мне 20 лет". Данная функция достанет 20 и сразу трансформирует это запись в int.

Теперь вы можете видеть, что у нас появился новый аргумент в функции — state. Он позволяет управлять состояниями пользователя, перемещать пользователя по состояниям и прочее.

Для аннотации типов используем FSMContext, который импортировали ранее.

Также под анкету я создал новый роутер. Он нам пригодится далее, когда мы будем делать «боевую» анкету.

Для красоты я добавил имитацию набора текста. И тут самое важное — это конструкция такого типа: await state.set_state(Form.name).

Эта запись указывает, что пользователь, когда дойдет до этого момента функции, окажется в состоянии Form.name, а значит, что его отправка данных (в нашем случае это ввод имени) окажется уже в этом состоянии.

Идем далее:

@questionnaire_router.message(F.text, Form.name)
async def start_questionnaire_process(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
        await asyncio.sleep(2)
        await message.answer('Супер! А теперь напиши сколько тебе полных лет:')
    await state.set_state(Form.age)

Здесь мы применили новый метод, а именно — сохранение данных от пользователя в переменную name при помощи state.update_data. После, как вы уже знаете, мы просто перенесли пользователя в новое состояние Form.age.

@questionnaire_router.message(F.text, Form.age)
async def start_questionnaire_process(message: Message, state: FSMContext):
    check_age = extract_number(message.text)

    if not check_age or not (1 <= int(message.text) <= 100):
        await message.reply("Пожалуйста, введите корректный возраст (число от 1 до 100).")
        await state.set_state(Form.age)
    else:
        await state.update_data(age=check_age)

        data = await state.get_data()
        msg_text = (f'Вас зовут <b>{data.get("name")}</b> и вам <b>{data.get("age")}</b> лет. '
                    f'Спасибо за то, что ответили на мои вопросы.')
        await message.answer(msg_text)
        await state.clear()

Здесь сначала мы проверили, есть ли в последнем сообщении пользователя возраст и находится ли он в диапазоне от 1 до 100. Если это не так, мы отправляем пользователя в состояние ввода возраста.

В случае, когда нужно просто оставить пользователя в том же состоянии при ошибке, можно использовать:

if not check_age or not (1 <= int(message.text) <= 100):
    await message.reply("Пожалуйста, введите корректный возраст (число от 1 до 100).")
    return

Тем самым мы укажем, что нужно остаться там же, но я в таких случаях обычно явно прописываю, в какое состояние нужно отправиться (когда через полгода, например, к своему проекту возвращаешься — это становится очень полезным).

Если возраст был введен корректно, мы записываем возраст в хранилище (этого также можно было не делать, так как мы все равно в этом хендлере завершаем сценарий, но я считаю, что явное лучше неявного).

Для того чтобы достать данные из хранилища, мы использовали запись:

data = await state.get_data()

В данном случае data — это обычный питоновский словарь, с которым можно делать все, что можно делать со словарями. Например, доставать значения по ключу.

Обратите особое внимание на:

await state.clear()

Это нужно делать обязательно, когда вы завершаете сценарий. Тут все просто. Если вы не завершите сценарий, бот не поймет, что возраст получен, и будет ждать его до бесконечности.

Сейчас рассмотрим самый частый случай у новичков.

Пользователь заполняет анкету, а после передумывает. Нажимает через командное меню /start, а у него ничего не происходит. Дело в том, что сценарий, который мы запустили, не завершился.

Бывает, что происходит. В таком случае вместо имени бот записывает имя «/start», после отправляет новый вопрос «Введи возраст». Пользователь снова жмет /start, и это все идет до момента, пока пользователь не удаляет бота и не считает, что его делали некомпетентные люди.

Чтобы эту проблему избежать, стоит в своей архитектуре закладывать возможность выхода из сценария анкетирования. Лично я всегда закладываю в команде /start и в прочих командах (в их хендлерах) сброс сценария. Для этого необходимо следующее:

async def cmd_start(message: Message, state: FSMContext):
    await state.clear()

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

Также советую добавлять возможность выхода по клику на кнопку клавиатуры. Это может быть текстовая кнопка с надписью «Отмена» или инлайн-кнопка с call_data = cancel, а далее просто обработчик, который будет закрывать (очищать) хранилище, тем самым выбивая пользователя из сценария.

На это акцентируйте особое внимание, так как проблем, связанных с «не выходом» из сценария, случается предостаточно.

Надеюсь, что к настоящему моменту вы уловили общие принципы взаимодействия с FSM, а значит что мы можем приступать к заполнению «боевой» анкеты. Бот пойдет по такому сценарию:

1. Сначала спрашивает пол (текстовая клавиатура с вариантами пола)

2. Затем возраст (удалим клавиатуру)

3. Потом имя и фамилию

4. Далее логин (инлайн клавиатура с возможностью выбрать свой логин с профиля телеграмм если он есть)

5. Попросит отправить фото (тут смысл в том чтоб захватить file_id фото)

6. И наконец, попросит добавить описание о себе.

Сейчас мы просто перепишем наш код, добавив новые состояния, а в результате выведем данные, введенные пользователем, с вопросом «Все верно?» и вариантами «Все верно» и «Заполнить сначала» (этот вариант будет сбрасывать анкету и запускать её с момента ввода имени).

Сначала давайте пропишем клавиатуры:

Текстовая клавиатура выбора пола:

def gender_kb():
    kb_list = [
        [KeyboardButton(text="