MZC/KnowMathBot/backend/app/routers/learning_analytics.py

"""
学情分析路由：个人/全班统计、按年级/类别/知识点 breakdown、错题本查询。
"""
from collections import defaultdict
from uuid import UUID

from fastapi import APIRouter, Depends
from sqlmodel import Session, select

from app.db import get_session
from app.deps import get_current_user, require_role
from app.models import Exercise, ExerciseAttempt, User, UserRole, WrongBookItem
from app.utils.knowledge_codes import split_knowledge_codes
from app.schemas import (
    CategoryBreakdownItem,
    GradeBreakdownItem,
    LearningBreakdownResponse,
    LearningStatsResponse,
    StudentLearningStatsItem,
    TopicBreakdownItem,
    WrongBookItemPublic,
    WrongBookSummaryItem,
)


router = APIRouter(prefix="/learning-analytics", tags=["learning-analytics"])


def _exercise_map_by_id(exercises: list[Exercise]) -> dict[str, Exercise]:
    """按习题 id 字符串建索引，避免 UUID 与驱动类型不一致导致查找失败。"""
    m: dict[str, Exercise] = {}
    for ex in exercises:
        m[str(ex.id)] = ex
    return m


def _resolve_exercise(session: Session, ex_map: dict[str, Exercise], exercise_id: UUID) -> Exercise | None:
    ex = ex_map.get(str(exercise_id))
    if ex is not None:
        return ex
    ex2 = session.get(Exercise, exercise_id)
    return ex2


# 更具体的路径须先于 /me 注册，避免被通配路由遮蔽
@router.get("/me/knowledge-breakdown", response_model=LearningBreakdownResponse)
def my_knowledge_breakdown(session: Session = Depends(get_session), current_user: User = Depends(get_current_user)):
    attempts = list(session.exec(select(ExerciseAttempt).where(ExerciseAttempt.user_id == current_user.id)).all())
    if not attempts:
        return LearningBreakdownResponse(by_grade=[], by_category=[], by_topic=[])

    ex_ids = list({a.exercise_id for a in attempts})
    exercises = list(session.exec(select(Exercise).where(Exercise.id.in_(ex_ids))).all())
    ex_map = _exercise_map_by_id(exercises)

    grade_acc: dict[str, dict[str, int]] = defaultdict(lambda: {"attempts": 0, "correct": 0, "wrong": 0})
    cat_acc: dict[tuple[str, str], dict[str, int]] = defaultdict(lambda: {"attempts": 0, "correct": 0, "wrong": 0})
    topic_acc: dict[str, dict[str, int]] = defaultdict(lambda: {"attempts": 0, "correct": 0, "wrong": 0})
    topic_meta: dict[str, dict[str, str | None]] = {}

    for a in attempts:
        ex = _resolve_exercise(session, ex_map, a.exercise_id)
        g = (ex.knowledge_grade if ex else None) or "未标注"
        c = (ex.knowledge_category if ex else None) or "未标注"
        ic = a.is_correct is True
        iw = a.is_correct is False

        grade_acc[g]["attempts"] += 1
        if ic:
            grade_acc[g]["correct"] += 1
        elif iw:
            grade_acc[g]["wrong"] += 1

        cat_acc[(g, c)]["attempts"] += 1
        if ic:
            cat_acc[(g, c)]["correct"] += 1
        elif iw:
            cat_acc[(g, c)]["wrong"] += 1

        codes = split_knowledge_codes(ex.knowledge_code) if ex else []
        code = codes[0] if codes else "未标注"
        if code not in topic_meta:
            topic_meta[code] = {
                "name": (ex.knowledge_point_name if ex else None) or None,
                "grade": g,
                "category": c,
            }
        topic_acc[code]["attempts"] += 1
        if ic:
            topic_acc[code]["correct"] += 1
        elif iw:
            topic_acc[code]["wrong"] += 1

    def acc_float(row: dict[str, int]) -> float:
        n = row["attempts"]
        if not n:
            return 0.0
        return row["correct"] / n

    by_grade = [
        GradeBreakdownItem(
            grade=k,
            attempts=v["attempts"],
            correct=v["correct"],
            wrong=v["wrong"],
            accuracy=acc_float(v),
        )
        for k, v in sorted(grade_acc.items(), key=lambda x: (-x[1]["attempts"], x[0]))
    ]
    by_category = [
        CategoryBreakdownItem(
            grade=g,
            category=c,
            attempts=v["attempts"],
            correct=v["correct"],
            wrong=v["wrong"],
            accuracy=acc_float(v),
        )
        for (g, c), v in sorted(cat_acc.items(), key=lambda x: (-x[1]["attempts"], x[0][0], x[0][1]))
    ]
    by_topic = [
        TopicBreakdownItem(
            knowledge_code=code,
            knowledge_point_name=(topic_meta.get(code) or {}).get("name"),
            knowledge_grade=(topic_meta.get(code) or {}).get("grade"),
            knowledge_category=(topic_meta.get(code) or {}).get("category"),
            attempts=v["attempts"],
            correct=v["correct"],
            wrong=v["wrong"],
            accuracy=acc_float(v),
        )
        for code, v in sorted(topic_acc.items(), key=lambda x: (-x[1]["attempts"], x[0]))
    ]

    return LearningBreakdownResponse(by_grade=by_grade, by_category=by_category, by_topic=by_topic)


@router.get("/me", response_model=LearningStatsResponse)
def my_stats(session: Session = Depends(get_session), current_user: User = Depends(get_current_user)):
    attempts = session.exec(select(ExerciseAttempt).where(ExerciseAttempt.user_id == current_user.id)).all()
    attempts_total = len(attempts)
    correct_total = sum(1 for a in attempts if a.is_correct is True)
    accuracy = (correct_total / attempts_total) if attempts_total else 0.0
    return LearningStatsResponse(
        user_id=current_user.id,
        attempts_total=attempts_total,
        correct_total=correct_total,
        accuracy=accuracy,
    )


@router.get("/students", response_model=list[StudentLearningStatsItem])
def stats_for_all_students(
    session: Session = Depends(get_session),
    _teacher=Depends(require_role(UserRole.teacher)),
):
    students = list(session.exec(select(User).where(User.role == UserRole.student)).all())
    if not students:
        return []

    attempts = list(session.exec(select(ExerciseAttempt)).all())
    by_user: dict[str, list[ExerciseAttempt]] = {}
    for a in attempts:
        by_user.setdefault(str(a.user_id), []).append(a)

    out: list[StudentLearningStatsItem] = []
    for s in students:
        s_attempts = by_user.get(str(s.id), [])
        attempts_total = len(s_attempts)
        correct_total = sum(1 for a in s_attempts if a.is_correct is True)
        accuracy = (correct_total / attempts_total) if attempts_total else 0.0
        out.append(
            StudentLearningStatsItem(
                user_id=s.id,
                username=s.username,
                attempts_total=attempts_total,
                correct_total=correct_total,
                accuracy=accuracy,
            )
        )

    out.sort(key=lambda x: (-x.attempts_total, x.username))
    return out


@router.get("/wrong-book/me", response_model=list[WrongBookItemPublic])
def my_wrong_book(session: Session = Depends(get_session), current_user: User = Depends(get_current_user)):
    items = list(
        session.exec(
            select(WrongBookItem)
            .where(WrongBookItem.user_id == current_user.id)
            .order_by(WrongBookItem.created_at.desc())
        ).all()
    )
    if not items:
        return []

    ex_ids = list({it.exercise_id for it in items})
    exercises = list(session.exec(select(Exercise).where(Exercise.id.in_(ex_ids))).all())
    ex_map = _exercise_map_by_id(exercises)

    out: list[WrongBookItemPublic] = []
    for it in items:
        ex = _resolve_exercise(session, ex_map, it.exercise_id)
        res = None
        if ex:
            res = (ex.resolution or "").strip() or "暂无"
        ca = (str(it.correct_answer).strip() if it.correct_answer else "") or None
        if not ca and ex and getattr(ex, "correct_answer", None):
            ca = str(ex.correct_answer).strip() or None
        out.append(
            WrongBookItemPublic(
                id=it.id,
                user_id=it.user_id,
                exercise_id=it.exercise_id,
                knowledge_code=it.knowledge_code,
                wrong_answer=it.wrong_answer,
                correct_answer=ca,
                created_at=it.created_at,
                title=ex.title if ex else None,
                body=ex.body if ex else None,
                question_type=ex.question_type if ex else None,
                resolution=res,
                knowledge_grade=ex.knowledge_grade if ex else None,
                knowledge_category=ex.knowledge_category if ex else None,
                knowledge_point_name=ex.knowledge_point_name if ex else None,
            )
        )
    return out


@router.get("/wrong-book/summary", response_model=list[WrongBookSummaryItem])
def wrong_book_summary(
    session: Session = Depends(get_session),
    _teacher=Depends(require_role(UserRole.teacher)),
):
    items = session.exec(select(WrongBookItem)).all()
    by_code: dict[str, int] = {}
    for it in items:
        by_code[it.knowledge_code] = by_code.get(it.knowledge_code, 0) + 1
    out = [WrongBookSummaryItem(knowledge_code=k, wrong_count=v) for k, v in by_code.items()]
    out.sort(key=lambda x: (-x.wrong_count, x.knowledge_code))
    return out