Creating Surveys in Python Using aiogram 2.x
Introduction
In the world of chatbot development on the Telegram platform, creating interactive questionnaires can be a non-trivial task. In this post, I will share a system I developed based on the library aiogram 2.x. It allows you to easily create and process questionnaires with text answers and options, as well as manage bot states. In this article, we will look at key aspects of the implementation, including state handling, saving answers, and message management.
main idea
The system is based on the use of a dictionary, where each key represents a step of the questionnaire. Depending on the type of question (answer options or text field), the handler will correctly respond to user interaction. To manage states, a aiogram.dispatcher.FSMContext.
The idea itself was born when I was given the task of sifting through questionnaires in a legacy bot, in which each step was clearly defined, which would have led to long hours of checking states and replacing step_13 with step_12.
Structure of the questionnaire
The questionnaire is presented as a dictionary, where the keys are the numbers of questions, and the values are dictionaries describing the text of the question, the answer options, and the type of question. Example of the structure:
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',
},
# ...
} Here custom_kb — a function for creating a custom keyboard (inline keyboard), and the keys type determine how the bot should process responses.
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_keyboardDefinition of states
To manage states we use StatesGroup from aiogram. This allows you to define the steps of the questionnaire that the bot will go through sequentially:
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()Questionnaire Processor
The main handler is responsible for interaction with the user during the survey. It receives and processes both text messages and button clicks:
@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()Main functions
process_save_answer
This function saves the user's answers and completes the survey if the final step is reached.
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
Sends a new question to the user, updating the progress and saving the message ID for later deletion (I use emoji for the progress bar):
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}\n\n{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_idChecking the correctness of the answer
For each step of the questionnaire, it is checked whether the answer is correct depending on the type of question:
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 TrueDeleting previous messages
Removes previous survey messages to keep the interface clean:
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):
passExample of use
Now let's look at how we can use the system we've created. Let's say we want to create a questionnaire to collect feedback on a product. We have several questions that users can answer:
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',
}
}Conclusion
In this article, we looked at how you can easily and flexibly implement a library-based questionnaire system. aiogram 2.x. This system can be adapted to various needs, be it collecting feedback, conducting surveys, or even training via a bot. I hope this approach will help you in developing your own Python projects!
Note: In the article I used the library aiogramas it is widely used and well documented for building Python bots. Version 2.x was chosen out of necessity, as it is necessary to work with legacy.
