LWX/ai_teaching_system/backend/services/mastery_service.py

# -*- coding: utf-8 -*-
"""
弱点建模与掌握度更新服务
测评 → 诊断(弱点更新) → 每日一练(推送) → 训练/辅导 → 再测评 闭环核心逻辑
"""
from datetime import datetime, date
from typing import Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from backend.models import (
    Question,
    AnswerRecord,
    KnowledgePoint,
    QuestionKnowledgePoint,
    StudentKnowledgeMastery,
    StudentProfile,
    DailyRecommendation,
)


# 弱点更新参数
ALPHA = 1  # 答错时 weakness_weight 增加系数
BETA = 5   # 答错时 mastery_score 降低系数
GAMMA = 1  # 答对时 weakness_weight 衰减系数
MASTERY_GAIN = 3  # 答对时 mastery_score 提升量
MASTERY_THRESHOLD = 80  # 推荐时排除已掌握的知识点（mastery_score > 此值）


def _get_knowledge_points_for_question(
    db: Session, question_id: int, question: Question = None
) -> List[Tuple[int, int]]:
    """
    获取题目关联的知识点 (knowledge_point_id, weight) 列表。
    优先从 question_knowledge_points 读取；若无则从 Question.knowledge_point 字符串解析并懒建映射。
    """
    qkps = (
        db.query(QuestionKnowledgePoint)
        .filter(QuestionKnowledgePoint.question_id == question_id)
        .all()
    )
    if qkps:
        return [(qkp.knowledge_point_id, qkp.weight or 1) for qkp in qkps]

    # 回退：从 Question.knowledge_point 字符串解析
    if question is None:
        question = db.query(Question).filter(Question.id == question_id).first()
    if not question or not (question.knowledge_point or "").strip():
        return []

    kp_name = (question.knowledge_point or "").strip()
    kp = db.query(KnowledgePoint).filter(KnowledgePoint.name == kp_name).first()
    if not kp:
        kp = KnowledgePoint(name=kp_name)
        db.add(kp)
        db.flush()  # 获取 id

    # 懒建 question_knowledge_points
    qkp = QuestionKnowledgePoint(
        question_id=question_id,
        knowledge_point_id=kp.id,
        weight=1,
    )
    db.add(qkp)
    db.flush()
    return [(kp.id, 1)]


def _get_or_create_profile(db: Session, user_id: int) -> StudentProfile:
    """获取或创建学生学习档案"""
    profile = db.query(StudentProfile).filter(StudentProfile.user_id == user_id).first()
    if not profile:
        profile = StudentProfile(user_id=user_id)
        db.add(profile)
        db.flush()
    return profile


def _get_or_create_mastery(
    db: Session, user_id: int, knowledge_point_id: int
) -> StudentKnowledgeMastery:
    """获取或创建学生-知识点掌握度记录"""
    row = (
        db.query(StudentKnowledgeMastery)
        .filter(
            StudentKnowledgeMastery.user_id == user_id,
            StudentKnowledgeMastery.knowledge_point_id == knowledge_point_id,
        )
        .first()
    )
    if not row:
        row = StudentKnowledgeMastery(
            user_id=user_id,
            knowledge_point_id=knowledge_point_id,
        )
        db.add(row)
        db.flush()
    return row


def update_mastery_from_answers(
    db: Session,
    user_id: int,
    answers: List[dict],
    session_kind: str = "practice",
    include_diagnostic_in_mastery: bool = False,
) -> None:
    """
    根据一次测评的答题记录更新 student_knowledge_mastery 与 student_profiles。
    answers: [{"question_id": int, "selected_option": str}, ...]
    内部会查询 Question 判断 is_correct。
    """
    # 诊断/再测评用于弱点建模与推荐题选择，不直接影响“学情分析展示口径”
    # （掌握度颜色/雷达图/章节均值/最近练习），避免用户感知到“被重置”。
    is_diagnostic = session_kind in {"diagnostic", "re_evaluate"}
    update_mastery_score = not is_diagnostic
    update_last_practiced_at = not is_diagnostic
    if is_diagnostic and include_diagnostic_in_mastery:
        # 兼容式开关：仅在显式开启时，诊断/再测评也计入掌握度分值
        update_mastery_score = True
        update_last_practiced_at = True

    profile = _get_or_create_profile(db, user_id)
    profile.total_questions_answered += len(answers)
    profile.last_active_at = datetime.utcnow()

    for item in answers:
        qid = item.get("question_id")
        selected = (item.get("selected_option") or "").strip()
        if not qid or not selected:
            continue

        question = db.query(Question).filter(Question.id == qid).first()
        if not question:
            continue

        is_correct = selected == question.correct_option
        kp_list = _get_knowledge_points_for_question(db, qid, question)

        for kp_id, weight in kp_list:
            mastery = _get_or_create_mastery(db, user_id, kp_id)
            if is_correct:
                mastery.correct_count = (mastery.correct_count or 0) + 1
                mastery.weakness_weight = max(
                    0, (mastery.weakness_weight or 0) - GAMMA * weight
                )
                if update_mastery_score:
                    mastery.mastery_score = min(
                        100, (mastery.mastery_score or 0) + MASTERY_GAIN
                    )
            else:
                mastery.wrong_count = (mastery.wrong_count or 0) + 1
                mastery.weakness_weight = (mastery.weakness_weight or 0) + ALPHA * weight
                if update_mastery_score:
                    mastery.mastery_score = max(
                        0, (mastery.mastery_score or 0) - BETA * weight
                    )
            if update_last_practiced_at:
                mastery.last_practiced_at = datetime.utcnow()


def mark_daily_recommendations_done(
    db: Session,
    user_id: int,
    question_ids: List[int],
    today: date = None,
    answer_record_ids: Optional[Dict[int, int]] = None,
) -> None:
    """将今日推荐中已作答的题目标记为完成"""
    if today is None:
        today = date.today()
    if not question_ids:
        return

    if not answer_record_ids:
        db.query(DailyRecommendation).filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.recommend_date == today,
            DailyRecommendation.question_id.in_(question_ids),
        ).update({DailyRecommendation.is_done: 1}, synchronize_session=False)
        return

    # 精细化更新：为每道题写入对应的 answer_record_id（用于闭环追踪）
    for qid in set(question_ids):
        arid = answer_record_ids.get(qid)
        if not arid:
            # 没拿到 answer_record_id 就只标记完成，避免影响主流程
            continue
        db.query(DailyRecommendation).filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.recommend_date == today,
            DailyRecommendation.question_id == qid,
        ).update(
            {DailyRecommendation.is_done: 1, DailyRecommendation.answer_record_id: arid},
            synchronize_session=False,
        )

    # 兜底：对那些没命中 answer_record_ids 的题，至少标记完成
    missing_qids = [qid for qid in question_ids if qid not in answer_record_ids]
    if missing_qids:
        db.query(DailyRecommendation).filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.recommend_date == today,
            DailyRecommendation.question_id.in_(missing_qids),
        ).update({DailyRecommendation.is_done: 1}, synchronize_session=False)