Создание опросников на Python с использованием aiogram 2.x

Введение

В мире разработки чат-ботов на платформе Telegram создание интерактивных опросников может быть задачей нетривиальной. В этом посте я поделюсь системой, которую разработал на основе библиотеки aiogram 2.x. Она позволяет легко создавать и обрабатывать опросники с текстовыми ответами и вариантами выбора, а также управлять состояниями бота. В статье мы разберем ключевые аспекты реализации, включая обработку состояний, сохранение ответов и управление сообщениями.

Основная идея

Система основана на использовании словаря, где каждый ключ представляет собой шаг опросника. В зависимости от типа вопроса (варианты ответов или текстовое поле) обработчик будет корректно реагировать на взаимодействие с пользователем. Для управления состояниями используется aiogram.dispatcher.FSMContext.

Сама идея родилась когда мне прилетела задача перелопатить опросники в легаси-боте, в которых каждый шаг был определен явно, что привело бы к долгим часам сверки состояний и замен step_13 на step_12.

Структура опросника

Опросник представлен в виде словаря, где ключами являются номера вопросов, а значениями — словари, описывающие текст вопроса, варианты ответов и тип вопроса. Пример структуры:

test_survey_payload = {
    1: {
        'text': 'Вопрос 1 с вариантами',
        'options': custom_kb({
            'вариант 1': 'test_survey 1:1',
            'вариант 2': 'test_survey 1:2',
            'вариант 3': 'test_survey 1:3',
        }),
        'type': 'vars',
    },
    2: {
        'text': 'Вопрос 2 с текстом',
        'options': custom_kb({
            'назад': 'test_survey back'
        }),
        'type': 'text',
    },
    # ...
}

Здесь custom_kb — функция для создания кастомной клавиатуры (inline keyboard), а ключи type определяют, как бот должен обрабатывать ответы.

def custom_kb(button_data: dict) -> InlineKeyboardMarkup:
    """
    Creates custom keyboard from dict {'button_text': 'button_callback'}
    """
    returned_keyboard = InlineKeyboardMarkup(row_width=1)
    for key, value in button_data.items():
        button = InlineKeyboardButton(text=key, callback_data=value)
        returned_keyboard.add(button)
    return returned_keyboard

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

Для управления состояниями мы используем StatesGroup из aiogram. Это позволяет определить шаги опросника, которые бот будет проходить последовательно:

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

class TestSurvey(StatesGroup):
    step_1 = State()
    step_2 = State()
    step_3 = State()
    step_4 = State()
    step_5 = State()
    finish = State()

Обработчик опросника

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

@dp.message_handler(state=TestSurvey)
@dp.callback_query_handler(lambda c: c.data.startswith('test_survey'), state=TestSurvey)
async def test_survey(event: types.CallbackQuery | types.Message, state: FSMContext):
    current_state = await state.get_state()
    step = get_step(current_state, test_survey_payload)

    if not is_answer_correct_type(event, step, test_survey_payload):
        return

    await delete_prew_messages(state, event)

    if isinstance(event, types.CallbackQuery) and event.data == 'test_survey back':
        await TestSurvey.previous()
        current_state = await state.get_state()
        step = get_step(current_state, test_survey_payload)
        await process_new_question(step - 1, state, test_survey_payload, event)
        return

    if await process_save_answer(state, current_state, step, test_survey_payload, event) == 'finish':
        return

    await process_new_question(step, state, test_survey_payload, event)
    await TestSurvey.next()

Основные функции

process_save_answer

Эта функция сохраняет ответы пользователя и завершает опросник, если достигнут финальный шаг.

async def process_save_answer(state: FSMContext, current_state: str, step: int, test_survey_payload: dict,
                              event: types.CallbackQuery | types.Message):
    answer = event.data if isinstance(event, types.CallbackQuery) else event.text
    if current_state.split(':')[-1] != 'finish':
        step -= 1
    async with state.proxy() as data:
        question = test_survey_payload.get(step)
        data[step] = {
            'question': question.get('text') if question else 'Начало опроса',
            'answer': answer
        }
        if current_state.split(':')[-1] == 'finish':
            del data['message_to_delete_id']
            await state.finish()
            return 'finish', dict(data)
    return 'next', ''

process_new_question

Отправляет новый вопрос пользователю, обновляя прогресс и сохраняя ID сообщения для последующего удаления (для прогресс-бара я использую эмодзи):

async def process_new_question(step: int, state: FSMContext, test_survey_payload: dict,
                               event: types.CallbackQuery | types.Message = None, show_progres: bool = True):
    payload = test_survey_payload.get(step)
    message = event.message if isinstance(event, types.CallbackQuery) else event

    if show_progres:
        total_steps = len(test_survey_payload)
        current_step = f'{"⬜" * step}{"⬛" * (total_steps - step)}'
        answer_text = f'{current_step}nn{payload.get("text")}'
    else:
        answer_text = payload.get('text')

    if payload.get('options'):
        answer = await message.answer(answer_text, reply_markup=payload.get('options'))
    else:
        answer = await message.answer(answer_text)
    async with state.proxy() as data:
        data['message_to_delete_id'] = answer.message_id

Проверка корректности ответа

Для каждого шага опросника проверяется, является ли ответ корректным в зависимости от типа вопроса:

def is_answer_correct_type(event: types.CallbackQuery | types.Message, step: int, test_survey_payload: dict) -> bool:
    question = test_survey_payload.get(step - 1)
    if question:
        question_type = question.get('type')
        if question_type == 'vars' and isinstance(event, types.Message):
            return False
    return True

Удаление предыдущих сообщений

Удаляет предыдущие сообщения опросника, чтобы интерфейс оставался чистым:

async def delete_prew_messages(state: FSMContext, event: types.CallbackQuery | types.Message):
    message = event.message if isinstance(event, types.CallbackQuery) else event
    await message.delete()
    async with state.proxy() as data:
        if data.get('message_to_delete_id'):
            try:
                await bot.delete_message(chat_id=message.chat.id, message_id=data['message_to_delete_id'])
            except (MessageCantBeDeleted, MessageToDeleteNotFound):
                pass

Пример использования

Теперь рассмотрим, как можно использовать созданную систему. Предположим, что мы хотим создать опросник для сбора отзывов о продукте. У нас есть несколько вопросов, на которые пользователи могут ответить:

feedback_survey_payload = {
    1: {
        'text': 'Оцените качество нашего продукта:',
        'options': custom_kb({
            'Отлично': 'feedback_survey 1:1',
            'Хорошо': 'feedback_survey 1:2',
            'Удовлетворительно': 'feedback_survey 1:3',
            'Плохо': 'feedback_survey 1:4',
        }),
        'type': 'vars',
    },
    2: {
        'text': 'Что вам понравилось в нашем продукте?',
        'options': custom_kb({
            'назад': 'feedback_survey back'
        }),
        'type': 'text',
    },
    3: {
        'text': 'Как мы можем улучшить наш продукт?',
        'options': custom_kb({
            'назад': 'feedback_survey back'
        }),
        'type': 'text',
    },
    4: {
        'text': 'Спасибо за ваши ответы!',
        'type': 'text',
    }
}

Заключение

В данной статье мы рассмотрели, как можно легко и гибко реализовать систему опросников на основе библиотеки aiogram 2.x. Эта система может быть адаптирована под различные нужды, будь то сбор отзывов, проведение опросов или даже обучение через бот. Надеюсь, что этот подход поможет вам в разработке собственных проектов на Python!

Примечание: В статье я использовал библиотеку aiogram, поскольку она является широко используемой и хорошо документированной для создания ботов на Python. Версия 2.x выбрана по принуждению, так как приходится работать с легаси.