LWX/ai_teaching_system/backend/services/teacher_report_service.py

# -*- coding: utf-8 -*-
"""
教师端：班级学情报表聚合服务

口径说明（用于前端展示）：
- 知识点掌握：使用 `student_knowledge_mastery.mastery_score`，并把“缺失记录”视为 0（与前端学生端 outer join 后的默认 0 一致）。
- 高频错题知识点排行：使用 `student_knowledge_mastery.wrong_count` 在班级内求和。
- 知识点掌握分层：对“每个知识点在全班的平均掌握度(0-100)”进行分桶：<=40/<=60/<=80/>80。
- 实时进度：按 `daily_recommendations` 的 recommend_date 统计完成率（is_done=1）。
- 提升效果：用最近 7 天 vs 前 7 天的整体正确率(AnswerRecord 维度)与完成率(DailyRecommendation 维度)对比。
"""

from datetime import date, datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import case, func
from sqlalchemy.orm import Session

from backend.models import (
    AnswerRecord,
    DailyRecommendation,
    KnowledgePoint,
    StudentKnowledgeMastery,
    User,
)

# 复用已存在的知识点/题目初始化逻辑，避免 tables 为空导致全班统计为空
from backend.services.recommendation import _ensure_knowledge_points_from_questions, _sync_questions_from_chemistry_bank


def _get_teacher_class_students(db: Session, teacher_id: int) -> tuple[str, List[int]]:
    teacher = db.query(User).filter(User.id == teacher_id).first()
    if not teacher or not teacher.class_name:
        raise ValueError("teacher class_name is empty")

    student_rows = (
        db.query(User.id)
        .filter(User.role == "student")
        .filter(User.class_name == teacher.class_name)
        .all()
    )
    student_ids = [int(r[0]) for r in student_rows]
    return teacher.class_name, student_ids


def _bucket_tier(avg_score: float) -> Dict[str, Any]:
    # 与前端 Analysis.jsx masteryColor() 的分级保持一致：
    # <=40 红；<=60 橙；<=80 黄；>80 绿
    if avg_score <= 40:
        return {"tier": "薄弱", "min": 0, "max": 40}
    if avg_score <= 60:
        return {"tier": "待提升", "min": 41, "max": 60}
    if avg_score <= 80:
        return {"tier": "良好", "min": 61, "max": 80}
    return {"tier": "优秀", "min": 81, "max": 100}


def get_teacher_class_report(
    db: Session,
    teacher_id: int,
    days: int = 14,
    weak_top: int = 5,
    wrong_top: int = 10,
) -> Dict[str, Any]:
    """
    返回教师端班级学情报表（用于前端可视化）。
    """
    class_name, student_ids = _get_teacher_class_students(db, teacher_id)

    # 若 teacher 班级没有学生，则返回空壳结构（前端可提示“暂无数据”）
    students_count = len(student_ids)
    if students_count == 0:
        return {
            "classInfo": {"className": class_name, "studentsCount": 0},
            "overall": {"avgMasteryScore": 0, "masteredKnowledgePointRate": 0, "totalKnowledgePoints": 0},
            "wrongKnowledgePointRanking": [],
            "weakKnowledgePoints": [],
            "tierDistribution": {"tiers": []},
            "progress": {
                "completionTrend": [],
                "accuracyTrend": [],
                "improvementSummary": {},
            },
        }

    # 确保 questions / knowledge_points 至少有基础数据可用于聚合
    _sync_questions_from_chemistry_bank(db)
    _ensure_knowledge_points_from_questions(db)

    knowledge_points = db.query(KnowledgePoint.id, KnowledgePoint.name, KnowledgePoint.chapter).all()
    total_kp = len(knowledge_points)
    if total_kp == 0:
        return {
            "classInfo": {"className": class_name, "studentsCount": students_count},
            "overall": {"avgMasteryScore": 0, "masteredKnowledgePointRate": 0, "totalKnowledgePoints": 0},
            "wrongKnowledgePointRanking": [],
            "weakKnowledgePoints": [],
            "tierDistribution": {"tiers": []},
            "progress": {
                "completionTrend": [],
                "accuracyTrend": [],
                "improvementSummary": {},
            },
        }

    # 一次性拉取：全班学生在每个知识点的掌握/错题聚合（缺失按 0 处理）
    mastery_rows = (
        db.query(
            StudentKnowledgeMastery.knowledge_point_id,
            StudentKnowledgeMastery.mastery_score,
            StudentKnowledgeMastery.wrong_count,
        )
        .filter(StudentKnowledgeMastery.user_id.in_(student_ids))
        .all()
    )

    sum_mastery_by_kp: Dict[int, int] = {int(kp_id): 0 for kp_id, _, _ in knowledge_points}
    sum_wrong_by_kp: Dict[int, int] = {int(kp_id): 0 for kp_id, _, _ in knowledge_points}
    participant_count_by_kp: Dict[int, int] = {int(kp_id): 0 for kp_id, _, _ in knowledge_points}

    for kp_id, mastery_score, wrong_count in mastery_rows:
        kp_id = int(kp_id)
        sum_mastery_by_kp[kp_id] = sum_mastery_by_kp.get(kp_id, 0) + int(mastery_score or 0)
        sum_wrong_by_kp[kp_id] = sum_wrong_by_kp.get(kp_id, 0) + int(wrong_count or 0)
        participant_count_by_kp[kp_id] = participant_count_by_kp.get(kp_id, 0) + 1

    # 每个知识点的“全班平均掌握度”
    avg_mastery_by_kp: Dict[int, float] = {}
    for kp_id in sum_mastery_by_kp.keys():
        avg_mastery_by_kp[kp_id] = (sum_mastery_by_kp[kp_id] / students_count) if students_count else 0.0

    # 业务口径优化：
    # - 旧口径：全量知识点（未覆盖知识点按 0），容易被大量未练习知识点稀释为 0
    # - 新口径：优先使用“班级有记录的活跃知识点”（student_knowledge_mastery 中出现过）
    #   若活跃知识点为空，再回退到全量知识点，避免无数据时报错。
    active_kp_ids = {
        int(getattr(r, "knowledge_point_id", r[0]))
        for r in mastery_rows
        if int(getattr(r, "knowledge_point_id", r[0])) in avg_mastery_by_kp
    }
    metric_kp_ids = sorted(active_kp_ids) if active_kp_ids else sorted(avg_mastery_by_kp.keys())
    metric_kp_count = len(metric_kp_ids)

    # 活跃口径下，按“参与该知识点练习的学生”计算知识点均值，避免被未覆盖学生二次稀释。
    metric_scores = []
    for kp_id in metric_kp_ids:
        participants = int(participant_count_by_kp.get(kp_id, 0))
        if participants > 0:
            metric_scores.append(float(sum_mastery_by_kp.get(kp_id, 0)) / participants)
        else:
            metric_scores.append(float(avg_mastery_by_kp.get(kp_id, 0.0)))
    overall_avg_mastery = int(round(sum(metric_scores) / metric_kp_count)) if metric_kp_count else 0
    mastered_kp_count = sum(1 for v in metric_scores if float(v) >= 80.0)
    mastered_knowledge_point_rate = int(round(mastered_kp_count / metric_kp_count * 100)) if metric_kp_count else 0

    # 知识点薄弱列表：按平均掌握度升序（错题数降序作为 tie-breaker）
    kp_name_map = {int(kp_id): name for kp_id, name, _ in knowledge_points}

    weak_items = [
        {
            "knowledgePointId": kp_id,
            "name": kp_name_map.get(kp_id, ""),
            "avgMasteryScore": int(round(avg_mastery_by_kp.get(kp_id, 0.0))),
            "wrongCountTotal": int(sum_wrong_by_kp.get(kp_id, 0)),
        }
        for kp_id in avg_mastery_by_kp.keys()
    ]
    weak_items.sort(key=lambda x: (x["avgMasteryScore"], -x["wrongCountTotal"]))
    weak_knowledge_points = weak_items[: max(1, int(weak_top))]

    # 高频错题知识点排行：按错题数降序
    wrong_items = [
        {
            "knowledgePointId": kp_id,
            "name": kp_name_map.get(kp_id, ""),
            "wrongCountTotal": int(sum_wrong_by_kp.get(kp_id, 0)),
            "avgMasteryScore": int(round(avg_mastery_by_kp.get(kp_id, 0.0))),
        }
        for kp_id in avg_mastery_by_kp.keys()
    ]
    wrong_items.sort(key=lambda x: x["wrongCountTotal"], reverse=True)
    wrong_items = [x for x in wrong_items if x["wrongCountTotal"] > 0]
    wrong_knowledge_point_ranking = wrong_items[: max(1, int(wrong_top))]

    # 知识点掌握分层：对“每个知识点平均掌握度”分桶
    tier_buckets = [
        {"tier": "薄弱", "min": 0, "max": 40},
        {"tier": "待提升", "min": 41, "max": 60},
        {"tier": "良好", "min": 61, "max": 80},
        {"tier": "优秀", "min": 81, "max": 100},
    ]
    tier_counts = {b["tier"]: 0 for b in tier_buckets}
    for kp_id in avg_mastery_by_kp.keys():
        b = _bucket_tier(avg_mastery_by_kp.get(kp_id, 0.0))
        tier_counts[b["tier"]] = tier_counts.get(b["tier"], 0) + 1

    tier_distribution = []
    for b in tier_buckets:
        cnt = int(tier_counts.get(b["tier"], 0))
        pct = float(cnt) / total_kp * 100 if total_kp else 0.0
        tier_distribution.append(
            {
                "tier": b["tier"],
                "min": b["min"],
                "max": b["max"],
                "count": cnt,
                "percent": round(pct, 2),
            }
        )

    # 进度追踪：completionTrend（每日一练完成率）+ accuracyTrend（答题正确率）
    end_day = date.today()
    start_day = end_day - timedelta(days=max(1, int(days)) - 1)
    day_list = [start_day + timedelta(days=i) for i in range((end_day - start_day).days + 1)]

    # 1) daily_recommendations completionTrend
    completion_map: Dict[date, Dict[str, Any]] = {d: {"done": 0, "total": 0} for d in day_list}
    completion_rows = (
        db.query(
            DailyRecommendation.recommend_date,
            func.sum(case((DailyRecommendation.is_done == 1, 1), else_=0)).label("done_cnt"),
            func.count(DailyRecommendation.id).label("total_cnt"),
        )
        .filter(DailyRecommendation.user_id.in_(student_ids))
        .filter(DailyRecommendation.recommend_date >= start_day)
        .filter(DailyRecommendation.recommend_date <= end_day)
        .group_by(DailyRecommendation.recommend_date)
        .all()
    )
    for rec_date, done_cnt, total_cnt in completion_rows:
        rec_date = rec_date if isinstance(rec_date, date) else rec_date.date()
        completion_map[rec_date] = {"done": int(done_cnt or 0), "total": int(total_cnt or 0)}

    completion_trend = []
    for d in day_list:
        info = completion_map.get(d, {"done": 0, "total": 0})
        done = int(info.get("done", 0))
        total = int(info.get("total", 0))
        completion_rate = (done / total * 100.0) if total else None
        completion_trend.append(
            {
                "date": d.isoformat(),
                "done": done,
                "total": total,
                "completionRate": round(completion_rate, 2) if completion_rate is not None else None,
            }
        )

    # 2) answer_records accuracyTrend
    start_dt = datetime.combine(start_day, datetime.min.time())
    accuracy_map: Dict[date, Dict[str, Any]] = {d: {"correct": 0, "total": 0} for d in day_list}
    accuracy_rows = (
        db.query(
            func.date(AnswerRecord.answered_at).label("day"),
            func.sum(case((AnswerRecord.is_correct == True, 1), else_=0)).label("correct_cnt"),
            func.count(AnswerRecord.id).label("total_cnt"),
        )
        .filter(AnswerRecord.user_id.in_(student_ids))
        .filter(AnswerRecord.answered_at >= start_dt)
        .filter(AnswerRecord.answered_at <= datetime.combine(end_day, datetime.max.time()))
        .group_by(func.date(AnswerRecord.answered_at))
        .all()
    )
    for day_val, correct_cnt, total_cnt in accuracy_rows:
        # day_val 通常是 datetime/date 的一类
        day_obj = day_val if isinstance(day_val, date) else day_val.date()
        accuracy_map[day_obj] = {"correct": int(correct_cnt or 0), "total": int(total_cnt or 0)}

    accuracy_trend = []
    for d in day_list:
        info = accuracy_map.get(d, {"correct": 0, "total": 0})
        correct = int(info.get("correct", 0))
        total = int(info.get("total", 0))
        accuracy_rate = (correct / total * 100.0) if total else None
        accuracy_trend.append(
            {
                "date": d.isoformat(),
                "correct": correct,
                "total": total,
                "accuracyRate": round(accuracy_rate, 2) if accuracy_rate is not None else None,
            }
        )

    # 3) improvementSummary：last 7 days vs prev 7 days（加权平均）
    def _range_rates(range_start: date, range_end: date) -> Tuple[Optional[float], Optional[float]]:
        # accuracy
        acc = (
            db.query(
                func.sum(case((AnswerRecord.is_correct == True, 1), else_=0)).label("correct_cnt"),
                func.count(AnswerRecord.id).label("total_cnt"),
            )
            .filter(AnswerRecord.user_id.in_(student_ids))
            .filter(AnswerRecord.answered_at >= datetime.combine(range_start, datetime.min.time()))
            .filter(AnswerRecord.answered_at <= datetime.combine(range_end, datetime.max.time()))
            .one()
        )
        correct_cnt, total_cnt = int(acc.correct_cnt or 0), int(acc.total_cnt or 0)
        accuracy_rate = (correct_cnt / total_cnt * 100.0) if total_cnt else None

        # completion
        comp = (
            db.query(
                func.sum(case((DailyRecommendation.is_done == 1, 1), else_=0)).label("done_cnt"),
                func.count(DailyRecommendation.id).label("total_cnt"),
            )
            .filter(DailyRecommendation.user_id.in_(student_ids))
            .filter(DailyRecommendation.recommend_date >= range_start)
            .filter(DailyRecommendation.recommend_date <= range_end)
            .one()
        )
        done_cnt, comp_total_cnt = int(comp.done_cnt or 0), int(comp.total_cnt or 0)
        completion_rate = (done_cnt / comp_total_cnt * 100.0) if comp_total_cnt else None

        return accuracy_rate, completion_rate

    last_start = end_day - timedelta(days=6)
    prev_start = end_day - timedelta(days=13)
    prev_end = end_day - timedelta(days=7)

    last_acc_rate, last_comp_rate = _range_rates(last_start, end_day)
    prev_acc_rate, prev_comp_rate = _range_rates(prev_start, prev_end)

    def _delta(a: Optional[float], b: Optional[float]) -> Optional[float]:
        if a is None or b is None:
            return None
        return round(a - b, 2)

    improvement_summary = {
        "last7AvgAccuracyRate": round(last_acc_rate, 2) if last_acc_rate is not None else None,
        "prev7AvgAccuracyRate": round(prev_acc_rate, 2) if prev_acc_rate is not None else None,
        "deltaAccuracyRate": _delta(last_acc_rate, prev_acc_rate),
        "last7AvgCompletionRate": round(last_comp_rate, 2) if last_comp_rate is not None else None,
        "prev7AvgCompletionRate": round(prev_comp_rate, 2) if prev_comp_rate is not None else None,
        "deltaCompletionRate": _delta(last_comp_rate, prev_comp_rate),
    }

    return {
        "classInfo": {"className": class_name, "studentsCount": students_count},
        "overall": {
            "avgMasteryScore": overall_avg_mastery,
            "masteredKnowledgePointRate": mastered_knowledge_point_rate,
            "totalKnowledgePoints": total_kp,
            "activeKnowledgePoints": metric_kp_count,
        },
        "wrongKnowledgePointRanking": wrong_knowledge_point_ranking,
        "weakKnowledgePoints": weak_knowledge_points,
        "tierDistribution": {"tiers": tier_distribution},
        "progress": {
            "completionTrend": completion_trend,
            "accuracyTrend": accuracy_trend,
            "improvementSummary": improvement_summary,
        },
    }