mamonov.ep
69c651c031
Исправлено: при выполнении задач использовалась глобальная модель вместо привязанной к пользователю. Обновлены все handlers и services для передачи user_id в AI методы. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
317 lines
12 KiB
Python
317 lines
12 KiB
Python
from aiogram import Router, F
|
||
from aiogram.filters import Command
|
||
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery
|
||
from aiogram.fsm.context import FSMContext
|
||
from aiogram.fsm.state import State, StatesGroup
|
||
|
||
from database.db import async_session_maker
|
||
from services.user_service import UserService
|
||
from services.ai_service import ai_service
|
||
from utils.i18n import t, get_user_lang
|
||
from utils.levels import get_level_system, get_available_levels, CEFR_LEVELS, JLPT_LEVELS
|
||
|
||
router = Router()
|
||
|
||
|
||
class LevelTestStates(StatesGroup):
|
||
"""Состояния для прохождения теста уровня"""
|
||
taking_test = State()
|
||
|
||
|
||
@router.message(Command("level_test"))
|
||
async def cmd_level_test(message: Message, state: FSMContext):
|
||
"""Обработчик команды /level_test"""
|
||
await start_level_test(message, state)
|
||
|
||
|
||
async def start_level_test(message: Message, state: FSMContext, telegram_id: int = None):
|
||
"""Начать тест определения уровня"""
|
||
# Определяем ID пользователя (telegram_id передаётся при вызове из callback)
|
||
user_telegram_id = telegram_id or message.from_user.id
|
||
|
||
# Показываем описание теста
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_telegram_id(session, user_telegram_id)
|
||
lang = (user.language_interface if user else 'ru') or 'ru'
|
||
await message.answer(t(lang, 'level_test.intro'))
|
||
|
||
keyboard = InlineKeyboardMarkup(inline_keyboard=[
|
||
[InlineKeyboardButton(text=t(lang, 'level_test.start_btn'), callback_data="start_test")],
|
||
[InlineKeyboardButton(text=t(lang, 'level_test.cancel_btn'), callback_data="cancel_test")]
|
||
])
|
||
|
||
await message.answer(t(lang, 'level_test.press_button'), reply_markup=keyboard)
|
||
|
||
|
||
@router.callback_query(F.data == "cancel_test")
|
||
async def cancel_test(callback: CallbackQuery, state: FSMContext):
|
||
"""Отменить тест"""
|
||
await state.clear()
|
||
await callback.message.delete()
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_telegram_id(session, callback.from_user.id)
|
||
lang = (user.language_interface if user else 'ru') or 'ru'
|
||
await callback.message.answer(t(lang, 'level_test.cancelled'))
|
||
await callback.answer()
|
||
|
||
|
||
@router.callback_query(F.data == "start_test")
|
||
async def begin_test(callback: CallbackQuery, state: FSMContext):
|
||
"""Начать прохождение теста"""
|
||
# Сразу отвечаем на callback, чтобы избежать истечения таймаута
|
||
await callback.answer()
|
||
await callback.message.delete()
|
||
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_telegram_id(session, callback.from_user.id)
|
||
lang = get_user_lang(user)
|
||
learning_lang = getattr(user, 'learning_language', 'en') or 'en'
|
||
|
||
# Показываем индикатор загрузки
|
||
loading_msg = await callback.message.answer(t(lang, 'level_test_extra.generating'))
|
||
|
||
# Генерируем тест через AI с учётом языка изучения
|
||
questions = await ai_service.generate_level_test(learning_lang, user_id=user.id)
|
||
|
||
await loading_msg.delete()
|
||
|
||
if not questions:
|
||
await callback.message.answer(t(lang, 'level_test_extra.generate_failed'))
|
||
await state.clear()
|
||
return
|
||
|
||
# Сохраняем данные в состоянии (включая язык для определения системы уровней)
|
||
await state.update_data(
|
||
questions=questions,
|
||
current_question=0,
|
||
correct_answers=0,
|
||
answers=[], # Для отслеживания ответов по уровням
|
||
learning_language=learning_lang,
|
||
user_id=user.id
|
||
)
|
||
await state.set_state(LevelTestStates.taking_test)
|
||
|
||
# Показываем первый вопрос
|
||
await show_question(callback.message, state)
|
||
|
||
|
||
async def show_question(message: Message, state: FSMContext):
|
||
"""Показать текущий вопрос"""
|
||
data = await state.get_data()
|
||
questions = data.get('questions', [])
|
||
current_idx = data.get('current_question', 0)
|
||
user_id = data.get('user_id')
|
||
|
||
if current_idx >= len(questions):
|
||
# Тест завершён
|
||
await finish_test(message, state)
|
||
return
|
||
|
||
question = questions[current_idx]
|
||
|
||
# Формируем текст вопроса
|
||
# Язык интерфейса (берём user_id из state, т.к. message может быть от бота)
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_id(session, user_id)
|
||
lang = (user.language_interface if user else 'ru') or 'ru'
|
||
|
||
text = (
|
||
t(lang, 'level_test.q_header', i=current_idx + 1, n=len(questions)) + "\n\n" +
|
||
f"<b>{question['question']}</b>\n"
|
||
)
|
||
|
||
# Создаем кнопки с вариантами ответа
|
||
keyboard = []
|
||
letters = ['A', 'B', 'C', 'D']
|
||
for idx, option in enumerate(question['options']):
|
||
keyboard.append([
|
||
InlineKeyboardButton(
|
||
text=f"{letters[idx]}) {option}",
|
||
callback_data=f"answer_{idx}"
|
||
)
|
||
])
|
||
|
||
# Кнопка для показа перевода вопроса (локализованная)
|
||
keyboard.append([
|
||
InlineKeyboardButton(text=t(lang, 'level_test.show_translation_btn'), callback_data=f"show_qtr_{current_idx}")
|
||
])
|
||
|
||
reply_markup = InlineKeyboardMarkup(inline_keyboard=keyboard)
|
||
await message.answer(text, reply_markup=reply_markup)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("show_qtr_"), LevelTestStates.taking_test)
|
||
async def show_question_translation(callback: CallbackQuery, state: FSMContext):
|
||
"""Показать перевод текущего вопроса"""
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_telegram_id(session, callback.from_user.id)
|
||
lang = get_user_lang(user)
|
||
|
||
try:
|
||
idx = int(callback.data.split("_")[-1])
|
||
except Exception:
|
||
await callback.answer(t(lang, 'level_test_extra.translation_unavailable'), show_alert=True)
|
||
return
|
||
|
||
data = await state.get_data()
|
||
questions = data.get('questions', [])
|
||
if not (0 <= idx < len(questions)):
|
||
await callback.answer(t(lang, 'level_test_extra.translation_unavailable'), show_alert=True)
|
||
return
|
||
|
||
ru = questions[idx].get('question_ru') or t(lang, 'level_test_extra.translation_unavailable')
|
||
|
||
# Вставляем перевод в текущий текст сообщения
|
||
orig = callback.message.text or ""
|
||
marker = t(lang, 'level_test_extra.translation_marker')
|
||
if marker in orig:
|
||
await callback.answer(t(lang, 'level_test_extra.translation_already'))
|
||
return
|
||
|
||
new_text = f"{orig}\n{marker} <i>{ru}</i>"
|
||
try:
|
||
await callback.message.edit_text(new_text, reply_markup=callback.message.reply_markup)
|
||
except Exception:
|
||
await callback.message.answer(f"{marker} <i>{ru}</i>")
|
||
await callback.answer()
|
||
|
||
|
||
@router.callback_query(F.data.startswith("answer_"), LevelTestStates.taking_test)
|
||
async def process_answer(callback: CallbackQuery, state: FSMContext):
|
||
"""Обработать ответ на вопрос"""
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_telegram_id(session, callback.from_user.id)
|
||
lang = get_user_lang(user)
|
||
|
||
answer_idx = int(callback.data.split("_")[1])
|
||
|
||
data = await state.get_data()
|
||
questions = data.get('questions', [])
|
||
current_idx = data.get('current_question', 0)
|
||
correct_answers = data.get('correct_answers', 0)
|
||
answers = data.get('answers', [])
|
||
|
||
question = questions[current_idx]
|
||
is_correct = (answer_idx == question['correct'])
|
||
|
||
# Сохраняем результат
|
||
if is_correct:
|
||
correct_answers += 1
|
||
|
||
# Сохраняем ответ с уровнем вопроса
|
||
answers.append({
|
||
'level': question['level'],
|
||
'correct': is_correct
|
||
})
|
||
|
||
# Показываем результат
|
||
if is_correct:
|
||
result_text = t(lang, 'level_test_extra.correct')
|
||
else:
|
||
correct_option = question['options'][question['correct']]
|
||
result_text = t(lang, 'level_test_extra.incorrect') + "\n" + t(lang, 'level_test_extra.correct_answer', answer=correct_option)
|
||
|
||
await callback.message.edit_text(
|
||
t(lang, 'level_test.q_header', i=current_idx + 1, n=len(questions)) + "\n\n" +
|
||
result_text
|
||
)
|
||
|
||
# Переходим к следующему вопросу
|
||
await state.update_data(
|
||
current_question=current_idx + 1,
|
||
correct_answers=correct_answers,
|
||
answers=answers
|
||
)
|
||
|
||
# Небольшая пауза перед следующим вопросом
|
||
import asyncio
|
||
await asyncio.sleep(1.5)
|
||
|
||
await show_question(callback.message, state)
|
||
await callback.answer()
|
||
|
||
|
||
async def finish_test(message: Message, state: FSMContext):
|
||
"""Завершить тест и определить уровень"""
|
||
data = await state.get_data()
|
||
questions = data.get('questions', [])
|
||
correct_answers = data.get('correct_answers', 0)
|
||
answers = data.get('answers', [])
|
||
learning_lang = data.get('learning_language', 'en')
|
||
user_id = data.get('user_id')
|
||
|
||
total = len(questions)
|
||
accuracy = int((correct_answers / total) * 100) if total > 0 else 0
|
||
|
||
# Определяем уровень на основе правильных ответов по уровням
|
||
level = determine_level(answers, learning_lang)
|
||
|
||
# Сохраняем уровень в базе данных (берём user_id из state, т.к. message может быть от бота)
|
||
async with async_session_maker() as session:
|
||
user = await UserService.get_user_by_id(session, user_id)
|
||
if user:
|
||
await UserService.update_user_level(session, user.id, level, learning_lang)
|
||
|
||
lang = get_user_lang(user)
|
||
level_desc = t(lang, f'level_test_extra.level_desc.{level}')
|
||
|
||
await state.clear()
|
||
|
||
result_text = (
|
||
t(lang, 'level_test_extra.result_title') +
|
||
t(lang, 'level_test_extra.results_header') +
|
||
t(lang, 'level_test_extra.correct_count', correct=correct_answers, total=total) +
|
||
t(lang, 'level_test_extra.accuracy', accuracy=accuracy) +
|
||
t(lang, 'level_test_extra.your_level', level=level) +
|
||
f"<i>{level_desc}</i>\n\n" +
|
||
t(lang, 'level_test_extra.level_set_hint')
|
||
)
|
||
|
||
await message.answer(result_text)
|
||
|
||
|
||
def determine_level(answers: list, learning_language: str = "en") -> str:
|
||
"""
|
||
Определить уровень на основе ответов
|
||
|
||
Args:
|
||
answers: Список ответов с уровнями
|
||
learning_language: Язык изучения для выбора системы уровней
|
||
|
||
Returns:
|
||
Определённый уровень (строка: A1-C2 или N5-N1)
|
||
"""
|
||
# Выбираем систему уровней
|
||
level_system = get_level_system(learning_language)
|
||
|
||
if level_system == "jlpt":
|
||
levels_order = JLPT_LEVELS # ["N5", "N4", "N3", "N2", "N1"]
|
||
default_level = "N5"
|
||
else:
|
||
levels_order = CEFR_LEVELS # ["A1", "A2", "B1", "B2", "C1", "C2"]
|
||
default_level = "A1"
|
||
|
||
# Подсчитываем правильные ответы по уровням
|
||
level_stats = {level: {'correct': 0, 'total': 0} for level in levels_order}
|
||
|
||
for answer in answers:
|
||
level = answer['level']
|
||
if level in level_stats:
|
||
level_stats[level]['total'] += 1
|
||
if answer['correct']:
|
||
level_stats[level]['correct'] += 1
|
||
|
||
# Определяем уровень: ищем последний уровень, где правильно >= 50%
|
||
determined_level = default_level
|
||
|
||
for level in levels_order:
|
||
if level_stats[level]['total'] > 0:
|
||
accuracy = level_stats[level]['correct'] / level_stats[level]['total']
|
||
if accuracy >= 0.5: # 50% и выше
|
||
determined_level = level
|
||
else:
|
||
# Если не прошёл этот уровень, останавливаемся
|
||
break
|
||
|
||
return determined_level
|