diff --git a/.drone.yml b/.drone.yml index 272bc12..5c9dc16 100644 --- a/.drone.yml +++ b/.drone.yml @@ -12,10 +12,7 @@ trigger: - push - pull_request -# Глобальные переменные -environment: - IMAGE_NAME: quiz-bot - REGISTRY: localhost:5000 # Локальный registry или замените на ваш +# Примечание: Глобальные переменные определяются в шагах steps: # 1. Клонирование и подготовка diff --git a/Dockerfile b/Dockerfile index 6c189f7..6942707 100644 --- a/Dockerfile +++ b/Dockerfile @@ -18,10 +18,10 @@ RUN pip install --no-cache-dir --upgrade pip && \ # Production stage FROM python:3.12-slim -# Создаем пользователя для безопасности +# Создание пользователя и группы для безопасности RUN groupadd -r quizbot && useradd -r -g quizbot quizbot -# Устанавливаем системные зависимости +# Установка sqlite3 для работы с базой данных RUN apt-get update && apt-get install -y \ sqlite3 \ && rm -rf /var/lib/apt/lists/* \ @@ -34,9 +34,10 @@ ENV PATH="/opt/venv/bin:$PATH" # Создаем рабочую директорию WORKDIR /app -# Создаем необходимые директории +# Создание директорий с правильными правами доступа RUN mkdir -p /app/data /app/logs && \ - chown -R quizbot:quizbot /app + chown -R quizbot:quizbot /app && \ + chmod -R 775 /app/data /app/logs # Копируем код приложения COPY --chown=quizbot:quizbot . . diff --git a/config/config.py b/config/config.py index 1e29ed9..d9c8aca 100644 --- a/config/config.py +++ b/config/config.py @@ -1,29 +1,33 @@ import os from dataclasses import dataclass, field from typing import List + from dotenv import load_dotenv load_dotenv() + def get_admin_ids() -> List[int]: admin_str = os.getenv("ADMIN_IDS", "") if admin_str: return [int(x) for x in admin_str.split(",") if x.strip()] return [] + @dataclass class Config: bot_token: str = os.getenv("BOT_TOKEN", "") admin_ids: List[int] = field(default_factory=get_admin_ids) database_path: str = os.getenv("DATABASE_PATH", "data/quiz_bot.db") csv_data_path: str = os.getenv("CSV_DATA_PATH", "data/") - + # Настройки викторины questions_per_quiz: int = int(os.getenv("QUESTIONS_PER_QUIZ", "10")) time_per_question: int = int(os.getenv("TIME_PER_QUESTION", "30")) - + # Режимы работы guest_mode_enabled: bool = os.getenv("GUEST_MODE_ENABLED", "true").lower() == "true" test_mode_enabled: bool = os.getenv("TEST_MODE_ENABLED", "true").lower() == "true" + config = Config() diff --git a/data/korean_level_1.csv b/data/korean_level_1.csv old mode 100644 new mode 100755 diff --git a/data/korean_level_2.csv b/data/korean_level_2.csv old mode 100644 new mode 100755 diff --git a/data/korean_level_3.csv b/data/korean_level_3.csv old mode 100644 new mode 100755 diff --git a/data/korean_level_4.csv b/data/korean_level_4.csv old mode 100644 new mode 100755 diff --git a/data/korean_level_5.csv b/data/korean_level_5.csv old mode 100644 new mode 100755 diff --git a/data/quiz_bot.db b/data/quiz_bot.db old mode 100644 new mode 100755 index 2b408b7..ff82acc Binary files a/data/quiz_bot.db and b/data/quiz_bot.db differ diff --git a/docker-compose.yml b/docker-compose.yml index c98c439..fb7e065 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,15 +7,15 @@ services: dockerfile: Dockerfile container_name: quiz-bot restart: unless-stopped + user: "0:0" environment: - BOT_TOKEN=${BOT_TOKEN} - DATABASE_PATH=data/quiz_bot.db - CSV_DATA_PATH=data/ - LOG_LEVEL=INFO volumes: - # Персистентное хранение данных - - ./data:/app/data - - ./logs:/app/logs + - "./data:/app/data" + - "./logs:/app/logs" networks: - quiz-bot-network healthcheck: @@ -24,7 +24,12 @@ services: timeout: 10s retries: 3 start_period: 60s - # Ограничения ресурсов + command: > + sh -c " + chown -R quizbot:quizbot /app/data /app/logs && + chmod -R 775 /app/data /app/logs && + python -m src.bot + " deploy: resources: limits: @@ -34,7 +39,7 @@ services: cpus: '0.1' memory: 128M - # Опциональный сервис для мониторинга логов + # Опциональный сервис для мониторинга логов log-viewer: image: goharbor/harbor-log:v2.5.0 container_name: quiz-bot-logs diff --git a/requirements.txt b/requirements.txt index 2ad8dbb..8f61e45 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,3 +5,14 @@ pandas==2.1.4 python-dotenv==1.0.0 asyncio-mqtt==0.16.1 loguru==0.7.2 +pytest==7.4.0 +pytest-asyncio==0.22.0 +black +isort +flake8 +mypy +pytest +pytest-asyncio +pytest-cov +safety +bandit \ No newline at end of file diff --git a/src/bot.py b/src/bot.py index f19b5f7..8225dda 100644 --- a/src/bot.py +++ b/src/bot.py @@ -5,16 +5,18 @@ import asyncio import logging +import os import random +import sys + from aiogram import Bot, Dispatcher, F from aiogram.filters import Command, StateFilter from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage -from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton, InaccessibleMessage +from aiogram.types import (CallbackQuery, InaccessibleMessage, + InlineKeyboardButton, InlineKeyboardMarkup, Message) from aiogram.utils.keyboard import InlineKeyboardBuilder -import sys -import os # Добавляем путь к проекту project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -27,31 +29,33 @@ from src.services.csv_service import CSVQuizLoader, QuizGenerator # Настройка логирования logging.basicConfig(level=logging.INFO) + class QuizStates(StatesGroup): choosing_mode = State() choosing_category = State() choosing_level = State() in_quiz = State() + class QuizBot: def __init__(self): self.bot = Bot(token=config.bot_token) self.dp = Dispatcher(storage=MemoryStorage()) self.db = DatabaseManager(config.database_path) self.csv_loader = CSVQuizLoader(config.csv_data_path) - + # Регистрируем обработчики self.setup_handlers() - + def setup_handlers(self): """Регистрация всех обработчиков""" # Команды self.dp.message(Command("start"))(self.start_command) - self.dp.message(Command("help"))(self.help_command) + self.dp.message(Command("help"))(self.help_command) self.dp.message(Command("stats"))(self.stats_command) self.dp.message(Command("stop"))(self.stop_command) - - # Callback обработчики + + # Callback обработчики self.dp.callback_query(F.data == "guest_mode")(self.guest_mode_handler) self.dp.callback_query(F.data == "test_mode")(self.test_mode_handler) self.dp.callback_query(F.data.startswith("category_"))(self.category_handler) @@ -64,33 +68,43 @@ class QuizBot: async def start_command(self, message: Message, state: FSMContext): """Обработка команды /start""" user = message.from_user - + # Регистрируем пользователя await self.db.register_user( user_id=user.id, username=user.username, first_name=user.first_name, last_name=user.last_name, - language_code=user.language_code or 'ru' + language_code=user.language_code or "ru", ) - + await state.set_state(QuizStates.choosing_mode) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🎯 Гостевой режим (QUIZ)", callback_data="guest_mode")], - [InlineKeyboardButton(text="📚 Тестирование по материалам", callback_data="test_mode")], - [InlineKeyboardButton(text="📊 Моя статистика", callback_data="stats")], - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🎯 Гостевой режим (QUIZ)", callback_data="guest_mode" + ) + ], + [ + InlineKeyboardButton( + text="📚 Тестирование по материалам", callback_data="test_mode" + ) + ], + [InlineKeyboardButton(text="📊 Моя статистика", callback_data="stats")], + ] + ) + await message.answer( f"👋 Добро пожаловать в Quiz Bot, {user.first_name}!\n\n" "🎯 Гостевой режим - быстрая викторина для развлечения\n" "📚 Тестирование - серьезное изучение материалов с результатами\n\n" "Выберите режим работы:", reply_markup=keyboard, - parse_mode='HTML' + parse_mode="HTML", ) - + async def help_command(self, message: Message): """Обработка команды /help""" help_text = """🤖 Команды бота: @@ -115,18 +129,22 @@ class QuizBot: 📊 Доступные категории: • Корейский язык (уровни 1-5) • Более 120 уникальных вопросов""" - await message.answer(help_text, parse_mode='HTML') - + await message.answer(help_text, parse_mode="HTML") + async def stats_command(self, message: Message): """Обработка команды /stats""" user_stats = await self.db.get_user_stats(message.from_user.id) - - if not user_stats or user_stats['total_questions'] == 0: + + if not user_stats or user_stats["total_questions"] == 0: await message.answer("📊 У вас пока нет статистики. Пройдите первый тест!") return - - accuracy = (user_stats['correct_answers'] / user_stats['total_questions']) * 100 if user_stats['total_questions'] > 0 else 0 - + + accuracy = ( + (user_stats["correct_answers"] / user_stats["total_questions"]) * 100 + if user_stats["total_questions"] > 0 + else 0 + ) + stats_text = f"""📊 Ваша статистика: ❓ Всего вопросов: {user_stats['total_questions']} @@ -135,12 +153,18 @@ class QuizBot: 🎯 Завершенных сессий: {user_stats['sessions_completed'] or 0} 🏆 Лучший результат: {user_stats['best_score'] or 0:.1f}% 📊 Средний балл: {user_stats['average_score'] or 0:.1f}%""" - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🏠 Главное меню", callback_data="back_to_menu")] - ]) - - await message.answer(stats_text, reply_markup=keyboard, parse_mode='HTML') + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🏠 Главное меню", callback_data="back_to_menu" + ) + ] + ] + ) + + await message.answer(stats_text, reply_markup=keyboard, parse_mode="HTML") async def stop_command(self, message: Message): """Остановка текущего теста""" @@ -150,43 +174,61 @@ class QuizBot: await message.answer("❌ Текущий тест остановлен.") else: await message.answer("❌ У вас нет активного теста.") - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🏠 Главное меню", callback_data="back_to_menu")] - ]) + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🏠 Главное меню", callback_data="back_to_menu" + ) + ] + ] + ) await message.answer("Что бы вы хотели сделать дальше?", reply_markup=keyboard) async def guest_mode_handler(self, callback: CallbackQuery, state: FSMContext): """Обработка выбора гостевого режима""" - await state.update_data(mode='guest') + await state.update_data(mode="guest") await state.set_state(QuizStates.choosing_category) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🇰🇷 Корейский язык", callback_data="category_korean")], - [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")] - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🇰🇷 Корейский язык", callback_data="category_korean" + ) + ], + [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")], + ] + ) + await callback.message.edit_text( "🎯 Гостевой режим\n\nВыберите категорию для викторины:", reply_markup=keyboard, - parse_mode='HTML' + parse_mode="HTML", ) await callback.answer() async def test_mode_handler(self, callback: CallbackQuery, state: FSMContext): """Обработка выбора режима тестирования""" - await state.update_data(mode='test') + await state.update_data(mode="test") await state.set_state(QuizStates.choosing_category) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🇰🇷 Корейский язык", callback_data="category_korean")], - [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")] - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🇰🇷 Корейский язык", callback_data="category_korean" + ) + ], + [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")], + ] + ) + await callback.message.edit_text( "📚 Режим тестирования\n\nВыберите категорию для серьезного изучения:", reply_markup=keyboard, - parse_mode='HTML' + parse_mode="HTML", ) await callback.answer() @@ -194,20 +236,42 @@ class QuizBot: """Обработка выбора категории""" category = callback.data.split("_")[1] await state.update_data(category=category) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🥉 Уровень 1 (начальный)", callback_data="level_1")], - [InlineKeyboardButton(text="🥈 Уровень 2 (базовый)", callback_data="level_2")], - [InlineKeyboardButton(text="🥇 Уровень 3 (средний)", callback_data="level_3")], - [InlineKeyboardButton(text="🏆 Уровень 4 (продвинутый)", callback_data="level_4")], - [InlineKeyboardButton(text="💎 Уровень 5 (эксперт)", callback_data="level_5")], - [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")] - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🥉 Уровень 1 (начальный)", callback_data="level_1" + ) + ], + [ + InlineKeyboardButton( + text="🥈 Уровень 2 (базовый)", callback_data="level_2" + ) + ], + [ + InlineKeyboardButton( + text="🥇 Уровень 3 (средний)", callback_data="level_3" + ) + ], + [ + InlineKeyboardButton( + text="🏆 Уровень 4 (продвинутый)", callback_data="level_4" + ) + ], + [ + InlineKeyboardButton( + text="💎 Уровень 5 (эксперт)", callback_data="level_5" + ) + ], + [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")], + ] + ) + await callback.message.edit_text( f"🇰🇷 Корейский язык\n\nВыберите уровень сложности:", reply_markup=keyboard, - parse_mode='HTML' + parse_mode="HTML", ) await callback.answer() @@ -215,152 +279,187 @@ class QuizBot: """Обработка выбора уровня""" level = int(callback.data.split("_")[1]) data = await state.get_data() - + # Загружаем вопросы filename = f"{data['category']}_level_{level}.csv" questions = await self.csv_loader.load_questions_from_csv(filename) - + if not questions: await callback.message.edit_text( "❌ Вопросы для этого уровня пока недоступны.", - reply_markup=InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_menu")] - ]) + reply_markup=InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🔙 Назад", callback_data="back_to_menu" + ) + ] + ] + ), ) await callback.answer() return - + # Определяем количество вопросов - questions_count = 5 if data['mode'] == 'guest' else 10 - + questions_count = 5 if data["mode"] == "guest" else 10 + # Берем случайные вопросы - selected_questions = random.sample(questions, min(questions_count, len(questions))) - + selected_questions = random.sample( + questions, min(questions_count, len(questions)) + ) + # Создаем тестовую запись в БД test_id = await self.db.add_test( name=f"{data['category'].title()} Level {level}", description=f"Тест по {data['category']} языку, уровень {level}", level=level, - category=data['category'], - csv_file=filename + category=data["category"], + csv_file=filename, ) - + # Начинаем сессию await self.db.start_session( user_id=callback.from_user.id, test_id=test_id or 1, questions=selected_questions, - mode=data['mode'] + mode=data["mode"], ) - + await state.set_state(QuizStates.in_quiz) await self.show_question_safe(callback, callback.from_user.id, 0) await callback.answer() - + def shuffle_answers(self, question_data: dict) -> dict: """Перемешивает варианты ответов и обновляет правильный ответ""" options = [ - question_data['option1'], - question_data['option2'], - question_data['option3'], - question_data['option4'] + question_data["option1"], + question_data["option2"], + question_data["option3"], + question_data["option4"], ] - - correct_answer_text = options[question_data['correct_answer'] - 1] - + + correct_answer_text = options[question_data["correct_answer"] - 1] + # Перемешиваем варианты random.shuffle(options) - + # Находим новую позицию правильного ответа new_correct_position = options.index(correct_answer_text) + 1 - + # Обновляем данные вопроса shuffled_question = question_data.copy() - shuffled_question['option1'] = options[0] - shuffled_question['option2'] = options[1] - shuffled_question['option3'] = options[2] - shuffled_question['option4'] = options[3] - shuffled_question['correct_answer'] = new_correct_position - + shuffled_question["option1"] = options[0] + shuffled_question["option2"] = options[1] + shuffled_question["option3"] = options[2] + shuffled_question["option4"] = options[3] + shuffled_question["correct_answer"] = new_correct_position + return shuffled_question - async def show_question_safe(self, callback: CallbackQuery, user_id: int, question_index: int): + async def show_question_safe( + self, callback: CallbackQuery, user_id: int, question_index: int + ): """Безопасный показ вопроса через callback""" session = await self.db.get_active_session(user_id) - if not session or question_index >= len(session['questions_data']): + if not session or question_index >= len(session["questions_data"]): return - question = session['questions_data'][question_index] - + question = session["questions_data"][question_index] + # Перемешиваем варианты ответов только в тестовом режиме - if session['mode'] == 'test': + if session["mode"] == "test": question = self.shuffle_answers(question) - session['questions_data'][question_index] = question - await self.db.update_session_questions(user_id, session['questions_data']) - - total_questions = len(session['questions_data']) + session["questions_data"][question_index] = question + await self.db.update_session_questions(user_id, session["questions_data"]) + + total_questions = len(session["questions_data"]) # Создаем клавиатуру с ответами keyboard_builder = InlineKeyboardBuilder() for i in range(1, 5): - keyboard_builder.add(InlineKeyboardButton( - text=f"{i}. {question[f'option{i}']}", - callback_data=f"answer_{i}" - )) - + keyboard_builder.add( + InlineKeyboardButton( + text=f"{i}. {question[f'option{i}']}", callback_data=f"answer_{i}" + ) + ) + keyboard_builder.adjust(1) - + question_text = ( f"❓ Вопрос {question_index + 1}/{total_questions}\n\n" f"{question['question']}" ) - + # Безопасная отправка сообщения if callback.message and not isinstance(callback.message, InaccessibleMessage): try: - await callback.message.edit_text(question_text, reply_markup=keyboard_builder.as_markup(), parse_mode='HTML') + await callback.message.edit_text( + question_text, + reply_markup=keyboard_builder.as_markup(), + parse_mode="HTML", + ) except Exception: - await self.bot.send_message(callback.from_user.id, question_text, reply_markup=keyboard_builder.as_markup(), parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + question_text, + reply_markup=keyboard_builder.as_markup(), + parse_mode="HTML", + ) else: - await self.bot.send_message(callback.from_user.id, question_text, reply_markup=keyboard_builder.as_markup(), parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + question_text, + reply_markup=keyboard_builder.as_markup(), + parse_mode="HTML", + ) async def answer_handler(self, callback: CallbackQuery, state: FSMContext): """Обработка ответа на вопрос""" answer = int(callback.data.split("_")[1]) user_id = callback.from_user.id - + session = await self.db.get_active_session(user_id) if not session: await callback.answer("❌ Сессия не найдена") return - - current_q_index = session['current_question'] - question = session['questions_data'][current_q_index] - is_correct = answer == question['correct_answer'] - mode = session['mode'] - + + current_q_index = session["current_question"] + question = session["questions_data"][current_q_index] + is_correct = answer == question["correct_answer"] + mode = session["mode"] + # Обновляем счетчик правильных ответов if is_correct: - session['correct_count'] += 1 - + session["correct_count"] += 1 + # Обновляем прогресс в базе await self.db.update_session_progress( - user_id, current_q_index + 1, session['correct_count'] + user_id, current_q_index + 1, session["correct_count"] ) - + # Проверяем, есть ли еще вопросы - if current_q_index + 1 >= len(session['questions_data']): + if current_q_index + 1 >= len(session["questions_data"]): # Тест завершен - score = (session['correct_count'] / len(session['questions_data'])) * 100 + score = (session["correct_count"] / len(session["questions_data"])) * 100 await self.db.finish_session(user_id, score) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🏠 Главное меню", callback_data="back_to_menu")], - [InlineKeyboardButton(text="📊 Моя статистика", callback_data="stats")] - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🏠 Главное меню", callback_data="back_to_menu" + ) + ], + [ + InlineKeyboardButton( + text="📊 Моя статистика", callback_data="stats" + ) + ], + ] + ) + # Разный текст для разных режимов - if mode == 'test': + if mode == "test": final_text = ( f"🎉 Тест завершен!\n\n" f"📊 Результат: {session['correct_count']}/{len(session['questions_data'])}\n" @@ -369,7 +468,11 @@ class QuizBot: f"💡 Результат сохранен в вашей статистике" ) else: - result_text = "✅ Правильно!" if is_correct else f"❌ Неправильно. Правильный ответ: {question['correct_answer']}" + result_text = ( + "✅ Правильно!" + if is_correct + else f"❌ Неправильно. Правильный ответ: {question['correct_answer']}" + ) final_text = ( f"{result_text}\n\n" f"🎉 Викторина завершена!\n\n" @@ -377,62 +480,102 @@ class QuizBot: f"📈 Точность: {score:.1f}%\n" f"🏆 Оценка: {self.get_grade(score)}" ) - + # Безопасная отправка сообщения - if callback.message and not isinstance(callback.message, InaccessibleMessage): + if callback.message and not isinstance( + callback.message, InaccessibleMessage + ): try: - await callback.message.edit_text(final_text, reply_markup=keyboard, parse_mode='HTML') + await callback.message.edit_text( + final_text, reply_markup=keyboard, parse_mode="HTML" + ) except Exception: - await self.bot.send_message(callback.from_user.id, final_text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + final_text, + reply_markup=keyboard, + parse_mode="HTML", + ) else: - await self.bot.send_message(callback.from_user.id, final_text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + final_text, + reply_markup=keyboard, + parse_mode="HTML", + ) else: # Есть еще вопросы - if mode == 'test': + if mode == "test": # В тестовом режиме сразу переходим к следующему вопросу - await self.show_question_safe(callback, callback.from_user.id, current_q_index + 1) + await self.show_question_safe( + callback, callback.from_user.id, current_q_index + 1 + ) else: # В гостевом режиме показываем результат и кнопку "Следующий" - result_text = "✅ Правильно!" if is_correct else f"❌ Неправильно. Правильный ответ: {question['correct_answer']}" - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="➡️ Следующий вопрос", callback_data="next_question")] - ]) - + result_text = ( + "✅ Правильно!" + if is_correct + else f"❌ Неправильно. Правильный ответ: {question['correct_answer']}" + ) + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="➡️ Следующий вопрос", callback_data="next_question" + ) + ] + ] + ) + # Безопасная отправка сообщения - if callback.message and not isinstance(callback.message, InaccessibleMessage): + if callback.message and not isinstance( + callback.message, InaccessibleMessage + ): try: - await callback.message.edit_text(result_text, reply_markup=keyboard) + await callback.message.edit_text( + result_text, reply_markup=keyboard + ) except Exception: - await self.bot.send_message(callback.from_user.id, result_text, reply_markup=keyboard) + await self.bot.send_message( + callback.from_user.id, result_text, reply_markup=keyboard + ) else: - await self.bot.send_message(callback.from_user.id, result_text, reply_markup=keyboard) - + await self.bot.send_message( + callback.from_user.id, result_text, reply_markup=keyboard + ) + await callback.answer() - + async def next_question(self, callback: CallbackQuery): """Переход к следующему вопросу""" session = await self.db.get_active_session(callback.from_user.id) if session: - await self.show_question_safe(callback, callback.from_user.id, session['current_question']) + await self.show_question_safe( + callback, callback.from_user.id, session["current_question"] + ) await callback.answer() - + async def stats_callback_handler(self, callback: CallbackQuery): """Обработчик кнопки статистики через callback""" user_stats = await self.db.get_user_stats(callback.from_user.id) - - if not user_stats or user_stats['total_questions'] == 0: + + if not user_stats or user_stats["total_questions"] == 0: stats_text = "📊 У вас пока нет статистики. Пройдите первый тест!" else: - accuracy = (user_stats['correct_answers'] / user_stats['total_questions']) * 100 if user_stats['total_questions'] > 0 else 0 - + accuracy = ( + (user_stats["correct_answers"] / user_stats["total_questions"]) * 100 + if user_stats["total_questions"] > 0 + else 0 + ) + # Получаем дополнительную статистику recent_results = await self.db.get_recent_results(callback.from_user.id, 3) category_stats = await self.db.get_category_stats(callback.from_user.id) - - best_score = user_stats['best_score'] or 0 - avg_score = user_stats['average_score'] or 0 - + + best_score = user_stats["best_score"] or 0 + avg_score = user_stats["average_score"] or 0 + stats_text = f"""📊 Ваша статистика: 📈 Общие показатели: @@ -451,68 +594,116 @@ class QuizBot: if category_stats: stats_text += "\n\n🏷️ По категориям:" for cat_stat in category_stats[:2]: - cat_accuracy = (cat_stat['correct_answers'] / cat_stat['total_questions']) * 100 if cat_stat['total_questions'] > 0 else 0 + cat_accuracy = ( + (cat_stat["correct_answers"] / cat_stat["total_questions"]) + * 100 + if cat_stat["total_questions"] > 0 + else 0 + ) stats_text += f"\n📖 {cat_stat['category']}: {cat_stat['attempts']} попыток, {cat_accuracy:.1f}% точность" - + # Добавляем последние результаты if recent_results: stats_text += "\n\n📈 Последние результаты:" for result in recent_results: - mode_emoji = "🎯" if result['mode'] == 'guest' else "📚" + mode_emoji = "🎯" if result["mode"] == "guest" else "📚" stats_text += f"\n{mode_emoji} {result['score']:.1f}% ({result['correct_answers']}/{result['questions_asked']})" - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🏠 Главное меню", callback_data="back_to_menu")], - [InlineKeyboardButton(text="🔄 Обновить статистику", callback_data="stats")] - ]) - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🏠 Главное меню", callback_data="back_to_menu" + ) + ], + [ + InlineKeyboardButton( + text="🔄 Обновить статистику", callback_data="stats" + ) + ], + ] + ) + # Безопасная отправка сообщения if callback.message and not isinstance(callback.message, InaccessibleMessage): try: - await callback.message.edit_text(stats_text, reply_markup=keyboard, parse_mode='HTML') + await callback.message.edit_text( + stats_text, reply_markup=keyboard, parse_mode="HTML" + ) except Exception: - await self.bot.send_message(callback.from_user.id, stats_text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + stats_text, + reply_markup=keyboard, + parse_mode="HTML", + ) else: - await self.bot.send_message(callback.from_user.id, stats_text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + stats_text, + reply_markup=keyboard, + parse_mode="HTML", + ) await callback.answer() - + async def back_to_menu(self, callback: CallbackQuery, state: FSMContext): """Возврат в главное меню""" await state.clear() - + user = callback.from_user - + # Регистрируем пользователя (если еще не зарегистрирован) await self.db.register_user( user_id=user.id, username=user.username, first_name=user.first_name, last_name=user.last_name, - language_code=user.language_code or 'ru' + language_code=user.language_code or "ru", ) - + await state.set_state(QuizStates.choosing_mode) - - keyboard = InlineKeyboardMarkup(inline_keyboard=[ - [InlineKeyboardButton(text="🎯 Гостевой режим (QUIZ)", callback_data="guest_mode")], - [InlineKeyboardButton(text="📚 Тестирование по материалам", callback_data="test_mode")], - [InlineKeyboardButton(text="📊 Моя статистика", callback_data="stats")], - ]) - - text = (f"👋 Добро пожаловать в Quiz Bot, {user.first_name}!\n\n" - "🎯 Гостевой режим - быстрая викторина для развлечения\n" - "📚 Тестирование - серьезное изучение материалов с результатами\n\n" - "Выберите режим работы:") - + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="🎯 Гостевой режим (QUIZ)", callback_data="guest_mode" + ) + ], + [ + InlineKeyboardButton( + text="📚 Тестирование по материалам", callback_data="test_mode" + ) + ], + [InlineKeyboardButton(text="📊 Моя статистика", callback_data="stats")], + ] + ) + + text = ( + f"👋 Добро пожаловать в Quiz Bot, {user.first_name}!\n\n" + "🎯 Гостевой режим - быстрая викторина для развлечения\n" + "📚 Тестирование - серьезное изучение материалов с результатами\n\n" + "Выберите режим работы:" + ) + if callback.message and not isinstance(callback.message, InaccessibleMessage): try: - await callback.message.edit_text(text, reply_markup=keyboard, parse_mode='HTML') + await callback.message.edit_text( + text, reply_markup=keyboard, parse_mode="HTML" + ) except Exception: - await self.bot.send_message(callback.from_user.id, text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, + text, + reply_markup=keyboard, + parse_mode="HTML", + ) else: - await self.bot.send_message(callback.from_user.id, text, reply_markup=keyboard, parse_mode='HTML') + await self.bot.send_message( + callback.from_user.id, text, reply_markup=keyboard, parse_mode="HTML" + ) await callback.answer() - + def get_grade(self, score: float) -> str: """Получение оценки по проценту правильных ответов""" if score >= 90: @@ -523,31 +714,36 @@ class QuizBot: return "Удовлетворительно 📚" else: return "Нужно подтянуть знания 📖" - + async def start(self): """Запуск бота""" # Проверяем токен - if not config.bot_token or config.bot_token in ['your_bot_token_here', 'test_token_for_demo_purposes']: + if not config.bot_token or config.bot_token in [ + "your_bot_token_here", + "test_token_for_demo_purposes", + ]: print("❌ Ошибка: не настроен BOT_TOKEN в файле .env") return False - + # Инициализируем базу данных await self.db.init_database() - + print("✅ Bot starting...") print(f"🗄️ Database: {config.database_path}") print(f"📁 CSV files: {config.csv_data_path}") - + try: await self.dp.start_polling(self.bot) except Exception as e: logging.error(f"Error starting bot: {e}") return False + async def main(): """Главная функция""" bot = QuizBot() await bot.start() + if __name__ == "__main__": asyncio.run(main()) diff --git a/src/database/database.py b/src/database/database.py index b6059c7..33d9ff1 100644 --- a/src/database/database.py +++ b/src/database/database.py @@ -1,17 +1,20 @@ -import aiosqlite -import logging -from typing import List, Dict, Optional, Tuple, Union import json +import logging +from typing import Dict, List, Optional, Tuple, Union + +import aiosqlite + class DatabaseManager: def __init__(self, db_path: str): self.db_path = db_path - + async def init_database(self): """Инициализация базы данных и создание таблиц""" async with aiosqlite.connect(self.db_path) as db: # Таблица пользователей - await db.execute(""" + await db.execute( + """ CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY, username TEXT, @@ -23,10 +26,12 @@ class DatabaseManager: total_questions INTEGER DEFAULT 0, correct_answers INTEGER DEFAULT 0 ) - """) - + """ + ) + # Таблица тестов - await db.execute(""" + await db.execute( + """ CREATE TABLE IF NOT EXISTS tests ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, @@ -37,10 +42,12 @@ class DatabaseManager: created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_active BOOLEAN DEFAULT TRUE ) - """) - + """ + ) + # Таблица вопросов - await db.execute(""" + await db.execute( + """ CREATE TABLE IF NOT EXISTS questions ( id INTEGER PRIMARY KEY AUTOINCREMENT, test_id INTEGER, @@ -52,10 +59,12 @@ class DatabaseManager: correct_answer INTEGER NOT NULL, FOREIGN KEY (test_id) REFERENCES tests (id) ) - """) - + """ + ) + # Таблица результатов - await db.execute(""" + await db.execute( + """ CREATE TABLE IF NOT EXISTS results ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, @@ -70,10 +79,12 @@ class DatabaseManager: FOREIGN KEY (user_id) REFERENCES users (user_id), FOREIGN KEY (test_id) REFERENCES tests (id) ) - """) - + """ + ) + # Таблица активных сессий - await db.execute(""" + await db.execute( + """ CREATE TABLE IF NOT EXISTS active_sessions ( user_id INTEGER PRIMARY KEY, test_id INTEGER, @@ -85,28 +96,38 @@ class DatabaseManager: FOREIGN KEY (user_id) REFERENCES users (user_id), FOREIGN KEY (test_id) REFERENCES tests (id) ) - """) - + """ + ) + await db.commit() logging.info("Database initialized successfully") - - async def register_user(self, user_id: int, username: Optional[str] = None, - first_name: Optional[str] = None, last_name: Optional[str] = None, - language_code: str = 'ru', is_guest: bool = True) -> bool: + + async def register_user( + self, + user_id: int, + username: Optional[str] = None, + first_name: Optional[str] = None, + last_name: Optional[str] = None, + language_code: str = "ru", + is_guest: bool = True, + ) -> bool: """Регистрация нового пользователя""" try: async with aiosqlite.connect(self.db_path) as db: - await db.execute(""" + await db.execute( + """ INSERT OR REPLACE INTO users (user_id, username, first_name, last_name, language_code, is_guest) VALUES (?, ?, ?, ?, ?, ?) - """, (user_id, username, first_name, last_name, language_code, is_guest)) + """, + (user_id, username, first_name, last_name, language_code, is_guest), + ) await db.commit() return True except Exception as e: logging.error(f"Error registering user {user_id}: {e}") return False - + async def get_user(self, user_id: int) -> Optional[Dict]: """Получение данных пользователя""" try: @@ -122,30 +143,34 @@ class DatabaseManager: except Exception as e: logging.error(f"Error getting user {user_id}: {e}") return None - - async def add_test(self, name: str, description: str, level: int, - category: str, csv_file: str) -> Optional[int]: + + async def add_test( + self, name: str, description: str, level: int, category: str, csv_file: str + ) -> Optional[int]: """Добавление нового теста""" try: async with aiosqlite.connect(self.db_path) as db: - cursor = await db.execute(""" + cursor = await db.execute( + """ INSERT INTO tests (name, description, level, category, csv_file) VALUES (?, ?, ?, ?, ?) - """, (name, description, level, category, csv_file)) + """, + (name, description, level, category, csv_file), + ) await db.commit() return cursor.lastrowid except Exception as e: logging.error(f"Error adding test: {e}") return None - + async def get_tests_by_category(self, category: Optional[str] = None) -> List[Dict]: """Получение тестов по категории""" try: async with aiosqlite.connect(self.db_path) as db: if category: cursor = await db.execute( - "SELECT * FROM tests WHERE category = ? AND is_active = TRUE ORDER BY level", - (category,) + "SELECT * FROM tests WHERE category = ? AND is_active = TRUE ORDER BY level", + (category,), ) else: cursor = await db.execute( @@ -157,56 +182,73 @@ class DatabaseManager: except Exception as e: logging.error(f"Error getting tests: {e}") return [] - + async def add_questions_to_test(self, test_id: int, questions: List[Dict]) -> bool: """Добавление вопросов к тесту""" try: async with aiosqlite.connect(self.db_path) as db: for q in questions: - await db.execute(""" + await db.execute( + """ INSERT INTO questions (test_id, question, option1, option2, option3, option4, correct_answer) VALUES (?, ?, ?, ?, ?, ?, ?) - """, (test_id, q['question'], q['option1'], q['option2'], - q['option3'], q['option4'], q['correct_answer'])) + """, + ( + test_id, + q["question"], + q["option1"], + q["option2"], + q["option3"], + q["option4"], + q["correct_answer"], + ), + ) await db.commit() return True except Exception as e: logging.error(f"Error adding questions to test {test_id}: {e}") return False - + async def get_random_questions(self, test_id: int, count: int = 10) -> List[Dict]: """Получение случайных вопросов из теста""" try: async with aiosqlite.connect(self.db_path) as db: - cursor = await db.execute(""" + cursor = await db.execute( + """ SELECT * FROM questions WHERE test_id = ? ORDER BY RANDOM() LIMIT ? - """, (test_id, count)) + """, + (test_id, count), + ) rows = await cursor.fetchall() columns = [description[0] for description in cursor.description] return [dict(zip(columns, row)) for row in rows] except Exception as e: logging.error(f"Error getting random questions: {e}") return [] - - async def start_session(self, user_id: int, test_id: int, - questions: List[Dict], mode: str) -> bool: + + async def start_session( + self, user_id: int, test_id: int, questions: List[Dict], mode: str + ) -> bool: """Начало новой сессии викторины""" try: async with aiosqlite.connect(self.db_path) as db: questions_json = json.dumps(questions) - await db.execute(""" + await db.execute( + """ INSERT OR REPLACE INTO active_sessions (user_id, test_id, questions_data, mode) VALUES (?, ?, ?, ?) - """, (user_id, test_id, questions_json, mode)) + """, + (user_id, test_id, questions_json, mode), + ) await db.commit() return True except Exception as e: logging.error(f"Error starting session: {e}") return False - + async def get_active_session(self, user_id: int) -> Optional[Dict]: """Получение активной сессии пользователя""" try: @@ -218,45 +260,54 @@ class DatabaseManager: if row: columns = [description[0] for description in cursor.description] session = dict(zip(columns, row)) - session['questions_data'] = json.loads(session['questions_data']) + session["questions_data"] = json.loads(session["questions_data"]) return session return None except Exception as e: logging.error(f"Error getting active session: {e}") return None - - async def update_session_progress(self, user_id: int, question_num: int, - correct_count: int) -> bool: + + async def update_session_progress( + self, user_id: int, question_num: int, correct_count: int + ) -> bool: """Обновление прогресса сессии""" try: async with aiosqlite.connect(self.db_path) as db: - await db.execute(""" + await db.execute( + """ UPDATE active_sessions SET current_question = ?, correct_count = ? WHERE user_id = ? - """, (question_num, correct_count, user_id)) + """, + (question_num, correct_count, user_id), + ) await db.commit() return True except Exception as e: logging.error(f"Error updating session progress: {e}") return False - - async def update_session_questions(self, user_id: int, questions_data: list) -> bool: + + async def update_session_questions( + self, user_id: int, questions_data: list + ) -> bool: """Обновление данных вопросов в сессии (например, после перемешивания)""" try: async with aiosqlite.connect(self.db_path) as db: questions_json = json.dumps(questions_data, ensure_ascii=False) - await db.execute(""" + await db.execute( + """ UPDATE active_sessions SET questions_data = ? WHERE user_id = ? - """, (questions_json, user_id)) + """, + (questions_json, user_id), + ) await db.commit() return True except Exception as e: logging.error(f"Error updating session questions: {e}") return False - + async def finish_session(self, user_id: int, score: float) -> bool: """Завершение сессии и сохранение результатов""" try: @@ -265,37 +316,52 @@ class DatabaseManager: session = await self.get_active_session(user_id) if not session: return False - + # Сохраняем результат - await db.execute(""" + await db.execute( + """ INSERT INTO results (user_id, test_id, mode, questions_asked, correct_answers, score) VALUES (?, ?, ?, ?, ?, ?) - """, (user_id, session['test_id'], session['mode'], - len(session['questions_data']), session['correct_count'], score)) - + """, + ( + user_id, + session["test_id"], + session["mode"], + len(session["questions_data"]), + session["correct_count"], + score, + ), + ) + # Обновляем статистику пользователя - await db.execute(""" + await db.execute( + """ UPDATE users SET total_questions = total_questions + ?, correct_answers = correct_answers + ? WHERE user_id = ? - """, (len(session['questions_data']), session['correct_count'], user_id)) - + """, + (len(session["questions_data"]), session["correct_count"], user_id), + ) + # Удаляем активную сессию - await db.execute("DELETE FROM active_sessions WHERE user_id = ?", (user_id,)) - + await db.execute( + "DELETE FROM active_sessions WHERE user_id = ?", (user_id,) + ) + await db.commit() return True except Exception as e: logging.error(f"Error finishing session: {e}") return False - + async def get_user_stats(self, user_id: int) -> Optional[Dict]: """Получение статистики пользователя""" try: async with aiosqlite.connect(self.db_path) as db: - cursor = await db.execute(""" + cursor = await db.execute( + """ SELECT u.total_questions, u.correct_answers, @@ -308,8 +374,10 @@ class DatabaseManager: LEFT JOIN results r ON u.user_id = r.user_id WHERE u.user_id = ? GROUP BY u.user_id - """, (user_id,)) - + """, + (user_id,), + ) + row = await cursor.fetchone() if row: columns = [description[0] for description in cursor.description] @@ -319,12 +387,13 @@ class DatabaseManager: except Exception as e: logging.error(f"Error getting user stats: {e}") return None - + async def get_recent_results(self, user_id: int, limit: int = 5) -> List[Dict]: """Получение последних результатов пользователя""" try: async with aiosqlite.connect(self.db_path) as db: - cursor = await db.execute(""" + cursor = await db.execute( + """ SELECT r.mode, r.questions_asked, @@ -338,20 +407,23 @@ class DatabaseManager: WHERE r.user_id = ? ORDER BY r.end_time DESC LIMIT ? - """, (user_id, limit)) - + """, + (user_id, limit), + ) + rows = await cursor.fetchall() columns = [description[0] for description in cursor.description] return [dict(zip(columns, row)) for row in rows] except Exception as e: logging.error(f"Error getting recent results: {e}") return [] - + async def get_category_stats(self, user_id: int) -> List[Dict]: """Получение статистики по категориям""" try: async with aiosqlite.connect(self.db_path) as db: - cursor = await db.execute(""" + cursor = await db.execute( + """ SELECT t.category, COUNT(r.id) as attempts, @@ -364,8 +436,10 @@ class DatabaseManager: WHERE r.user_id = ? GROUP BY t.category ORDER BY attempts DESC - """, (user_id,)) - + """, + (user_id,), + ) + rows = await cursor.fetchall() columns = [description[0] for description in cursor.description] return [dict(zip(columns, row)) for row in rows]