Как реализуешь очередность вопросов в Telegram боте на Python?

Реализация очередности вопросов в Telegram боте на Python

Это реальная задача, которая часто встречается при разработке интерактивных ботов. Есть несколько подходов — от простых (линейные вопросы) до сложных (с ветвлением и условиями).

1. Простой подход: FSM (Finite State Machine) в aiogram

FSM (конечный автомат) — это стандартный способ управления состояниями в диалоге.

from aiogram import F, Router, types
from aiogram.filters.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
from aiogram.fsm.storage.memory import MemoryStorage

router = Router()

# Определить состояния (очередь вопросов)
class QuestionStates(StatesGroup):
    waiting_for_name = State()      # Вопрос 1: Имя
    waiting_for_age = State()       # Вопрос 2: Возраст
    waiting_for_email = State()     # Вопрос 3: Email
    waiting_for_confirmation = State()  # Вопрос 4: Подтверждение

# Команда начать
@router.message(commands=['start'])
async def start_command(message: types.Message, state: FSMContext):
    await state.set_state(QuestionStates.waiting_for_name)
    await message.answer("Привет! Как тебя зовут?")

# Обработка ответа на первый вопрос
@router.message(QuestionStates.waiting_for_name)
async def process_name(message: types.Message, state: FSMContext):
    await state.update_data(name=message.text)  # Сохранить ответ
    await state.set_state(QuestionStates.waiting_for_age)  # Перейти к следующему
    await message.answer("Сколько тебе лет?")

# Обработка ответа на второй вопрос
@router.message(QuestionStates.waiting_for_age)
async def process_age(message: types.Message, state: FSMContext):
    try:
        age = int(message.text)
        if age < 0 or age > 120:
            await message.answer("Пожалуйста, введи корректный возраст")
            return
        
        await state.update_data(age=age)
        await state.set_state(QuestionStates.waiting_for_email)
        await message.answer("Какой твой email?")
    except ValueError:
        await message.answer("Пожалуйста, введи число")

# Обработка ответа на третий вопрос
@router.message(QuestionStates.waiting_for_email)
async def process_email(message: types.Message, state: FSMContext):
    if '@' not in message.text:
        await message.answer("Пожалуйста, введи корректный email")
        return
    
    await state.update_data(email=message.text)
    
    # Показать все собранные данные
    data = await state.get_data()
    summary = f"""
    Спасибо за информацию!
    
    Имя: {data['name']}
    Возраст: {data['age']}
    Email: {data['email']}
    
    Это правильно?
    """
    
    await state.set_state(QuestionStates.waiting_for_confirmation)
    await message.answer(summary, reply_markup=get_confirmation_keyboard())

# Подтверждение
@router.message(QuestionStates.waiting_for_confirmation)
async def process_confirmation(message: types.Message, state: FSMContext):
    if message.text.lower() == 'да':
        await message.answer("Спасибо! Данные сохранены.")
        await state.clear()  # Завершить диалог
    else:
        # Вернуться к первому вопросу
        await state.set_state(QuestionStates.waiting_for_name)
        await message.answer("Давай начнём заново. Как тебя зовут?")

def get_confirmation_keyboard():
    """Клавиатура для подтверждения"""
    from aiogram.types import ReplyKeyboardMarkup, KeyboardButton
    return ReplyKeyboardMarkup(
        keyboard=[
            [KeyboardButton(text="Да"), KeyboardButton(text="Нет")]
        ],
        resize_keyboard=True
    )

2. Продвинутый подход: Контроллер для управления очередью

Для более сложных сценариев создай специальный класс:

from typing import Callable, Optional
from dataclasses import dataclass
from aiogram import F, types
from aiogram.fsm.context import FSMContext

@dataclass
class Question:
    """Один вопрос в очереди"""
    key: str  # Ключ для сохранения ответа
    text: str  # Текст вопроса
    validator: Optional[Callable] = None  # Функция валидации
    on_answer: Optional[Callable] = None  # Обработчик ответа

class QuestionnaireController:
    """Контроллер для управления очередью вопросов"""
    
    def __init__(self):
        self.questions: list[Question] = []
        self.current_index: dict[int, int] = {}  # user_id -> index
    
    def add_question(self, key: str, text: str, validator=None, on_answer=None):
        """Добавить вопрос в очередь"""
        self.questions.append(Question(key, text, validator, on_answer))
        return self
    
    async def start(self, user_id: int, state: FSMContext):
        """Начать опрос"""
        self.current_index[user_id] = 0
        await state.update_data(questionnaire_data={})
        await self._ask_current_question(user_id, state)
    
    async def _ask_current_question(self, user_id: int, state: FSMContext, message: types.Message = None):
        """Задать текущий вопрос"""
        index = self.current_index[user_id]
        
        if index >= len(self.questions):
            # Все вопросы закончились
            data = await state.get_data()
            await message.answer(f"Спасибо! Ваши ответы сохранены: {data['questionnaire_data']}")
            await state.clear()
            return
        
        question = self.questions[index]
        await message.answer(question.text)
        await state.update_data(current_question_key=question.key)
    
    async def process_answer(self, user_id: int, answer: str, state: FSMContext, message: types.Message):
        """Обработать ответ и перейти к следующему вопросу"""
        index = self.current_index[user_id]
        question = self.questions[index]
        
        # Валидация
        if question.validator and not question.validator(answer):
            await message.answer(f"Некорректный ответ. Пожалуйста, попробуй снова.")
            return
        
        # Сохранить ответ
        data = await state.get_data()
        data['questionnaire_data'][question.key] = answer
        await state.update_data(questionnaire_data=data['questionnaire_data'])
        
        # Пользовательский обработчик
        if question.on_answer:
            await question.on_answer(answer, state)
        
        # Перейти к следующему вопросу
        self.current_index[user_id] += 1
        await self._ask_current_question(user_id, state, message)

# Использование
questionnaire = QuestionnaireController()

# Добавить вопросы
questionnaire\
    .add_question(
        key='name',
        text='Как тебя зовут?',
        validator=lambda x: len(x) > 0
    )\
    .add_question(
        key='age',
        text='Сколько тебе лет?',
        validator=lambda x: x.isdigit() and 0 < int(x) < 120
    )\
    .add_question(
        key='city',
        text='Из какого города?',
    )

@router.message(commands=['start'])
async def start(message: types.Message, state: FSMContext):
    await questionnaire.start(message.from_user.id, state)

@router.message()
async def handle_answer(message: types.Message, state: FSMContext):
    await questionnaire.process_answer(
        message.from_user.id,
        message.text,
        state,
        message
    )

3. Вопросы с ветвлением (условные ветки)

Иногда следующий вопрос зависит от предыдущего ответа:

class ConditionalQuestionnaire:
    """Опрос с условными ветками"""
    
    async def ask_next_question(self, user_id: int, state: FSMContext, message: types.Message):
        """Выбрать следующий вопрос на основе данных"""
        data = await state.get_data()
        answers = data.get('answers', {})
        
        # Ветвление на основе предыдущего ответа
        if answers.get('age') and int(answers['age']) < 18:
            # Для несовершеннолетних
            await message.answer("Нужно согласие родителя. Введи имя родителя:")
        else:
            # Для взрослых
            await message.answer("Перейдём к следующему вопросу...")
        
        await state.set_state(QuestionStates.waiting_for_next_answer)

# Использование с инлайн клавиатурой
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton

async def ask_with_choice(message: types.Message, state: FSMContext):
    """Вопрос с выбором (сложнее чем текстовый ввод)"""
    keyboard = InlineKeyboardMarkup(
        inline_keyboard=[
            [InlineKeyboardButton(text="Да", callback_data="choice_yes")],
            [InlineKeyboardButton(text="Нет", callback_data="choice_no")],
        ]
    )
    
    await message.answer("Ты согласен с условиями?", reply_markup=keyboard)
    await state.set_state(QuestionStates.waiting_for_choice)

@router.callback_query(F.data == "choice_yes")
async def process_choice_yes(callback: types.CallbackQuery, state: FSMContext):
    await state.update_data(agreed=True)
    await callback.message.edit_text("Спасибо!")
    # Перейти к следующему вопросу

4. Сохранение и восстановление прогресса

import json
from datetime import datetime

class PersistentQuestionnaire:
    """Опрос с сохранением прогресса в БД"""
    
    async def save_progress(self, user_id: int, data: dict, db):
        """Сохранить прогресс пользователя"""
        await db.execute("""
            INSERT INTO questionnaire_progress (user_id, data, updated_at)
            VALUES (:user_id, :data, :updated_at)
            ON CONFLICT (user_id) DO UPDATE SET data = :data, updated_at = :updated_at
        """, {
            'user_id': user_id,
            'data': json.dumps(data),
            'updated_at': datetime.now()
        })
    
    async def load_progress(self, user_id: int, db):
        """Загрузить прогресс пользователя"""
        result = await db.fetch_one(
            "SELECT data FROM questionnaire_progress WHERE user_id = :user_id",
            {'user_id': user_id}
        )
        return json.loads(result['data']) if result else {}
    
    async def resume(self, user_id: int, state: FSMContext, db, message: types.Message):
        """Возобновить опрос с того же места"""
        data = await self.load_progress(user_id, db)
        if data:
            await state.update_data(questionnaire_data=data)
            await message.answer(f"Приветствую снова! Продолжим опрос. Следующий вопрос...")
            # Показать следующий вопрос
        else:
            await message.answer("Начнём с начала...")

Ключевые моменты

• FSM (State Machine) — основной инструмент для управления очередностью
• StatesGroup — объявить все состояния (вопросы)
• state.set_state() — переход к следующему состоянию (вопросу)
• state.update_data() — сохранение ответов
• Валидация — проверка корректности ответа перед сохранением
• Ветвления — разные вопросы на основе предыдущих ответов
• Сохранение прогресса — восстановление при перезагрузке бота
• Инлайн клавиатуры — для выбора вместо текстового ввода

Данный подход масштабируется от простых линейных опросов до сложных многоуровневых анкет с условной логикой.