""" Современный Telegram бот, работающий только через API БЕЗ прямого доступа к базе данных """ import asyncio import io from datetime import datetime from aiogram import Bot, Dispatcher, types from aiogram.filters import Command from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.types import KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove, BufferedInputFile from aiogram.filters.state import StateFilter import logging from bot_config import config from api_client import api_client # Настройка логирования logging.basicConfig( level=getattr(logging, config.LOG_LEVEL), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Инициализация бота bot = Bot(token=config.TELEGRAM_BOT_TOKEN) dp = Dispatcher(storage=MemoryStorage()) # ===== FSM States ===== class Form(StatesGroup): consent = State() fio = State() dob = State() phone = State() city = State() test_in_progress = State() # ===== /start command ===== @dp.message(Command("start")) async def start(message: types.Message, state: FSMContext): """Обработка команды /start""" try: # Создаем или получаем пользователя через API user_data = { "telegram_id": message.from_user.id, "username": message.from_user.username, "first_name": message.from_user.first_name, "last_name": message.from_user.last_name, "language_code": message.from_user.language_code or 'ru' } async with api_client: # Попробуем получить пользователя response = await api_client.get_user_by_telegram_id(message.from_user.id) if response.success: user = response.data logger.info(f"Найден существующий пользователь: {user['telegram_id']}") else: # Создаем нового пользователя create_response = await api_client.create_user(user_data) if create_response.success: user = create_response.data logger.info(f"Создан новый пользователь: {user['telegram_id']}") else: logger.error(f"Ошибка создания пользователя: {create_response.error}") await message.answer("Произошла ошибка при регистрации. Попробуйте позже.") return # Сохраняем данные пользователя в состояние await state.update_data(user_id=user['id'], telegram_id=user['telegram_id']) welcome_text = f"Добро пожаловать, {user.get('first_name', 'пользователь')}!\n\n" welcome_text += "Этот бот поможет вам пройти тестирование.\n" welcome_text += "Для продолжения необходимо дать согласие на обработку персональных данных.\n\n" welcome_text += "Даете ли вы согласие на обработку персональных данных?" keyboard = ReplyKeyboardMarkup( keyboard=[ [KeyboardButton(text="Да, согласен")], [KeyboardButton(text="Нет, не согласен")] ], resize_keyboard=True, one_time_keyboard=True ) await message.answer(welcome_text, reply_markup=keyboard) await state.set_state(Form.consent) except Exception as e: logger.error(f"Ошибка в start handler: {e}") await message.answer("Произошла ошибка. Попробуйте позже.") # ===== Согласие на обработку данных ===== @dp.message(Form.consent) async def process_consent(message: types.Message, state: FSMContext): """Обработка согласия на обработку данных""" if message.text == "Да, согласен": await message.answer( "Спасибо! Теперь введите ваше ФИО:", reply_markup=ReplyKeyboardRemove() ) await state.set_state(Form.fio) elif message.text == "Нет, не согласен": await message.answer( "К сожалению, без согласия на обработку персональных данных использование бота невозможно.", reply_markup=ReplyKeyboardRemove() ) await state.clear() else: await message.answer("Пожалуйста, выберите один из предложенных вариантов.") # ===== Сбор ФИО ===== @dp.message(Form.fio) async def process_fio(message: types.Message, state: FSMContext): """Обработка ввода ФИО""" if len(message.text.strip()) < 5: await message.answer("Пожалуйста, введите полное ФИО (минимум 5 символов).") return await state.update_data(fio=message.text.strip()) await message.answer("Введите вашу дату рождения в формате ДД.ММ.ГГГГ:") await state.set_state(Form.dob) # ===== Дата рождения ===== @dp.message(Form.dob) async def process_dob(message: types.Message, state: FSMContext): """Обработка ввода даты рождения""" try: # Проверяем формат даты dob = datetime.strptime(message.text.strip(), "%d.%m.%Y") # Проверяем возраст (должен быть от 16 до 80 лет) age = (datetime.now() - dob).days // 365 if age < 16 or age > 80: await message.answer("Возраст должен быть от 16 до 80 лет. Проверьте правильность даты.") return await state.update_data(date_of_birth=message.text.strip()) await message.answer("Введите ваш номер телефона:") await state.set_state(Form.phone) except ValueError: await message.answer("Неверный формат даты. Используйте формат ДД.ММ.ГГГГ (например, 15.05.1990):") # ===== Телефон ===== @dp.message(Form.phone) async def process_phone(message: types.Message, state: FSMContext): """Обработка ввода телефона""" phone = message.text.strip() # Простая проверка телефона if len(phone) < 10 or not any(char.isdigit() for char in phone): await message.answer("Пожалуйста, введите корректный номер телефона.") return await state.update_data(phone=phone) await message.answer("Введите ваш город:") await state.set_state(Form.city) # ===== Город ===== @dp.message(Form.city) async def process_city(message: types.Message, state: FSMContext): """Обработка ввода города""" city = message.text.strip() if len(city) < 2: await message.answer("Пожалуйста, введите название города (минимум 2 символа).") return # Получаем все собранные данные data = await state.get_data() telegram_id = data['telegram_id'] # Обновляем данные пользователя через API update_data = { "first_name": data.get('fio', '').split()[0] if data.get('fio') else None, "last_name": ' '.join(data.get('fio', '').split()[1:]) if len(data.get('fio', '').split()) > 1 else None, # Можно добавить дополнительные поля в модель User для хранения ФИО, телефона, города и даты рождения } async with api_client: response = await api_client.update_user(telegram_id, update_data) if not response.success: logger.error(f"Ошибка обновления пользователя: {response.error}") await message.answer( f"Спасибо! Ваши данные сохранены:\n" f"ФИО: {data['fio']}\n" f"Дата рождения: {data['date_of_birth']}\n" f"Телефон: {data['phone']}\n" f"Город: {city}\n\n" f"Теперь вы можете приступить к тестированию!" ) # Показываем доступные тесты await show_available_tests(message, state) # ===== Показать доступные тесты ===== async def show_available_tests(message: types.Message, state: FSMContext): """Показать список доступных тестов""" try: async with api_client: response = await api_client.get_tests(is_active=True) if not response.success: logger.error(f"Ошибка получения тестов: {response.error}") await message.answer("Ошибка загрузки тестов. Попробуйте позже.") return tests = response.data if not tests: await message.answer("На данный момент нет доступных тестов.") return # Создаем клавиатуру с тестами keyboard_buttons = [] for test in tests: keyboard_buttons.append([KeyboardButton(text=test['title'])]) keyboard = ReplyKeyboardMarkup( keyboard=keyboard_buttons, resize_keyboard=True, one_time_keyboard=True ) await message.answer( "Выберите тест для прохождения:", reply_markup=keyboard ) await state.clear() # Очищаем предыдущее состояние except Exception as e: logger.error(f"Ошибка в show_available_tests: {e}") await message.answer("Произошла ошибка при загрузке тестов.") # ===== Выбор теста ===== @dp.message() async def handle_test_selection(message: types.Message, state: FSMContext): """Обработка выбора теста""" try: async with api_client: # Получаем все тесты и ищем по названию response = await api_client.get_tests(is_active=True) if not response.success: await message.answer("Ошибка загрузки тестов.") return selected_test = None for test in response.data: if test['title'] == message.text: selected_test = test break if not selected_test: await message.answer( "Тест не найден. Используйте /start для просмотра доступных тестов.", reply_markup=ReplyKeyboardRemove() ) return # Получаем вопросы теста questions_response = await api_client.get_questions_by_test(selected_test['id']) if not questions_response.success: await message.answer("Ошибка загрузки вопросов теста.") return questions = questions_response.data if not questions: await message.answer("В этом тесте пока нет вопросов.") return # Загружаем варианты ответов для всех вопросов for question in questions: options_response = await api_client.get_answer_options_by_question(question['id']) if options_response.success: question['options'] = options_response.data else: question['options'] = [] # Сохраняем данные теста в состояние await state.update_data( test=selected_test, questions=questions, current_question=0, answers=[] ) await message.answer( f"Вы выбрали тест: {selected_test['title']}\n" f"Описание: {selected_test.get('description', 'Описание отсутствует')}\n" f"Количество вопросов: {len(questions)}\n\n" f"Готовы начать?", reply_markup=ReplyKeyboardMarkup( keyboard=[[KeyboardButton(text="Начать тест")]], resize_keyboard=True, one_time_keyboard=True ) ) await state.set_state(Form.test_in_progress) except Exception as e: logger.error(f"Ошибка в handle_test_selection: {e}") await message.answer("Произошла ошибка при выборе теста.") # ===== Прохождение теста ===== @dp.message(Form.test_in_progress) async def handle_test_progress(message: types.Message, state: FSMContext): """Обработка прохождения теста""" try: data = await state.get_data() if message.text == "Начать тест": # Показываем первый вопрос await show_question(message, state, 0) return # Обрабатываем ответ на вопрос questions = data.get('questions', []) current_question_idx = data.get('current_question', 0) answers = data.get('answers', []) if current_question_idx >= len(questions): await message.answer("Тест уже завершен.") return current_question = questions[current_question_idx] # Находим выбранный вариант ответа selected_option = None for option in current_question.get('options', []): if option['option_text'] == message.text: selected_option = option break if not selected_option: await message.answer("Пожалуйста, выберите один из предложенных вариантов ответа.") return # Сохраняем ответ answers.append({ 'question_id': current_question['id'], 'option_id': selected_option['id'], 'is_correct': selected_option.get('is_correct', False) }) # Переходим к следующему вопросу next_question_idx = current_question_idx + 1 await state.update_data( answers=answers, current_question=next_question_idx ) if next_question_idx >= len(questions): # Тест завершен await finish_test(message, state) else: # Показываем следующий вопрос await show_question(message, state, next_question_idx) except Exception as e: logger.error(f"Ошибка в handle_test_progress: {e}") await message.answer("Произошла ошибка при обработке ответа.") async def show_question(message: types.Message, state: FSMContext, question_idx: int): """Показать вопрос""" try: data = await state.get_data() questions = data.get('questions', []) if question_idx >= len(questions): return question = questions[question_idx] # Формируем текст вопроса question_text = f"Вопрос {question_idx + 1} из {len(questions)}:\n\n" question_text += question['question_text'] # Создаем клавиатуру с вариантами ответов keyboard_buttons = [] for option in question.get('options', []): keyboard_buttons.append([KeyboardButton(text=option['option_text'])]) keyboard = ReplyKeyboardMarkup( keyboard=keyboard_buttons, resize_keyboard=True, one_time_keyboard=True ) await message.answer(question_text, reply_markup=keyboard) except Exception as e: logger.error(f"Ошибка в show_question: {e}") await message.answer("Ошибка при показе вопроса.") async def finish_test(message: types.Message, state: FSMContext): """Завершить тест и показать результаты""" try: data = await state.get_data() answers = data.get('answers', []) test = data.get('test', {}) # Подсчитываем результаты correct_answers = sum(1 for answer in answers if answer.get('is_correct', False)) total_questions = len(answers) score_percentage = (correct_answers / total_questions * 100) if total_questions > 0 else 0 result_text = f"🎉 Тест завершен!\n\n" result_text += f"Тест: {test.get('title', 'Неизвестный тест')}\n" result_text += f"Правильных ответов: {correct_answers} из {total_questions}\n" result_text += f"Результат: {score_percentage:.1f}%\n\n" if score_percentage >= 80: result_text += "Отличный результат! 🏆" elif score_percentage >= 60: result_text += "Хороший результат! 👍" elif score_percentage >= 40: result_text += "Удовлетворительный результат. 👌" else: result_text += "Рекомендуем повторить материал. 📚" await message.answer( result_text, reply_markup=ReplyKeyboardMarkup( keyboard=[[KeyboardButton(text="/start")]], resize_keyboard=True, one_time_keyboard=True ) ) await state.clear() except Exception as e: logger.error(f"Ошибка в finish_test: {e}") await message.answer("Ошибка при завершении теста.") # ===== Основная функция ===== async def main(): """Основная функция запуска бота""" logger.info("Запуск Telegram бота...") try: # Проверяем доступность API async with api_client: if not await api_client.health_check(): logger.error("API недоступен! Убедитесь, что API сервер запущен.") return logger.info("API доступен, бот готов к работе") # Запускаем бота await dp.start_polling(bot) except Exception as e: logger.error(f"Ошибка при запуске бота: {e}") finally: await bot.session.close() if __name__ == "__main__": asyncio.run(main())