bot_api/qwen_api/services.py - master - servgri/interview

bot_api/qwen_api/services.py
173 строки7 KB
Seal
full project
27 ноя 2025, 14:24
27 ноя 2025, 14:24d1507ba
# tasks/services.py
import random
from datetime import datetime
from django.utils import timezone
from tasks.models import Task, UserProgress, UserBKTState, Topic
from bot_api.test_runner import run_task_tests
from django.contrib.auth import get_user_model


from .bkt_recommend import BKT
from .task_gen_analyzer import generate_task_with_llm, analyze_code_with_llm_and_pep8, get_hint_from_llm, SKILL_LIST

CRITICAL_SKILL_THRESHOLD = 0.2
HIGH_SKILL_THRESHOLD = 0.9
MAX_HINTS_PER_TASK = 2

def load_or_init_bkt_for_user(user):
    """Возвращает BKT объект для пользователя, загружая состояние из DB если есть."""
    try:
        ub = UserBKTState.objects.get(user=user)
        bkt = BKT()
        # предполагается, что у BKT есть метод load_state или прямое присваивание .state
        bkt.state = ub.state
        return bkt
    except UserBKTState.DoesNotExist:
        bkt = BKT()
        return bkt

def save_bkt_for_user(user, bkt):
    """Сохраняет сериализованное состояние BKT в БД."""
    obj, _ = UserBKTState.objects.get_or_create(user=user)
    obj.state = bkt.state
    obj.save()

def select_skill_for_task(bkt_model: BKT, available_skills: set) -> str:
    """Копия вашей логики выбора навыка, сокращённо."""
    recommended_skill = bkt_model.get_recommendation_skill()
    current_level = bkt_model.state.get(recommended_skill, bkt_model.pL0) if recommended_skill else bkt_model.pL0

    if recommended_skill and current_level < CRITICAL_SKILL_THRESHOLD:
        available = set(available_skills) - {recommended_skill}
        available = {s for s in available if bkt_model.state.get(s, bkt_model.pL0) <= HIGH_SKILL_THRESHOLD}
        if available:
            # выбираем наименее освоенный (но не критический)
            potentials = {s: bkt_model.state.get(s, bkt_model.pL0) for s in available}
            selected = min(potentials, key=potentials.get)
            return selected
        return recommended_skill

    if recommended_skill and current_level >= HIGH_SKILL_THRESHOLD:
        available = set(available_skills) - {recommended_skill}
        available = {s for s in available if bkt_model.state.get(s, bkt_model.pL0) < HIGH_SKILL_THRESHOLD}
        if available:
            potentials = {s: bkt_model.state.get(s, bkt_model.pL0) for s in available}
            selected = min(potentials, key=potentials.get)
            return selected
        return random.choice(list(available_skills)) if available_skills else recommended_skill

    return recommended_skill or random.choice(list(available_skills))

def determine_difficulty(skill_level: float, bkt_model: BKT, skill: str) -> str:
    min_diff, max_diff = bkt_model.get_recommended_difficulty_range(skill)
    if min_diff <= 0.33:
        return "easy"
    elif min_diff <= 0.66:
        return "medium"
    else:
        return "hard"

def format_task_for_db(raw_task: dict) -> dict:
    """
    Преобразование формата, который возвращает ваш LLM-генератор,
    в поля модели Task: title, difficulty, task_text, test_cases, ideal_solution, topic.
    Предполагается, что raw_task содержит эти ключи.
    """
    return {
        "title": raw_task.get("title", "Generated task"),
        "difficulty": raw_task.get("difficulty", "easy"),
        "task_text": raw_task.get("text", ""),
        "test_cases": raw_task.get("test_cases", []),
        "ideal_solution": raw_task.get("solution", ""),
        "topic_name": raw_task.get("topic", None),
    }

def insert_task_to_db(task_data: dict) -> Task:
    """Создаёт Task в БД, или возвращает существующую при совпадении текста/названия."""
    topic_name = task_data.get("topic_name") or "misc"
    topic_obj, _ = Topic.objects.get_or_create(name=topic_name)
    task = Task.objects.create(
        title=task_data["title"],
        difficulty=task_data["difficulty"],
        task_text=task_data["task_text"],
        test_cases=task_data["test_cases"],
        ideal_solution=task_data["ideal_solution"],
        topic=topic_obj,
        is_active=True,
    )
    return task

def get_task_from_db_for_skill(skill: str, exclude_ids: set = None):
    qs = Task.objects.filter(topic__name=skill, is_active=True)
    if exclude_ids:
        qs = qs.exclude(pk__in=list(exclude_ids))
    return qs.order_by("created_at").first()

def generate_new_task_for_skill(skill: str, llm_client) -> Task | None:
    difficulty = "medium"  # можно более тонко выбирать через BKT
    raw_task = generate_task_with_llm(skill, difficulty, llm_client)
    if not raw_task:
        return None
    data = format_task_for_db(raw_task)
    task_obj = insert_task_to_db(data)
    return task_obj

def get_or_generate_task_for_user(user, bkt_model: BKT, skill_to_focus: str, llm_client, used_task_ids: set):
    task = get_task_from_db_for_skill(skill_to_focus, exclude_ids=used_task_ids)
    if not task:
        task = generate_new_task_for_skill(skill_to_focus, llm_client)
        if not task:
            return None
    # избегаем повторов в сессии
    attempt = 0
    while task and task.id in used_task_ids and attempt < 5:
        task = generate_new_task_for_skill(skill_to_focus, llm_client)
        attempt += 1
    if not task:
        return None
    used_task_ids.add(task.id)
    return task

def process_submission(user, task: Task, user_code: str, llm_coder=None):
    """
    1) Прогон тестов (run_task_tests)
    2) Анализ (возможно LLM)
    3) Обновление UserProgress и BKT
    4) Сохранение и возвращаем результаты
    """
    passed, feedback = run_task_tests(task.id, user_code)

    # попытка сохранить прогресс
    progress, _ = UserProgress.objects.get_or_create(user=user, task=task)
    progress.user_answer = user_code
    progress.code = user_code
    progress.correct = passed
    progress.finished_at = timezone.now()
    progress.attempt_count = (progress.attempt_count or 0) + 1
    progress.save()

    # анализ кода через ваш LLM-анализатор (если есть)
    ai_result = None
    if llm_coder:
        try:
            ai_result = analyze_code_with_llm_and_pep8(user_code, task.task_text, llm_coder)
        except Exception as e:
            ai_result = {"error": str(e)}
    progress.ai_result = ai_result
    progress.save()

    # обновляем BKT
    bkt = load_or_init_bkt_for_user(user)
    skill = task.topic.name
    before = bkt.state.get(skill, bkt.pL0)
    bkt.update(skill, passed)
    after = bkt.state.get(skill, bkt.pL0)
    save_bkt_for_user(user, bkt)

    return {
        "passed": passed,
        "feedback": feedback,
        "ai_result": ai_result,
        "bkt_before": before,
        "bkt_after": after,
    }