LWX/ai_teaching_system/backend/services/recommendation.py

# -*- coding: utf-8 -*-
"""个性化推荐模块：习题测评推荐算法、错题集、学情分析、每日一练"""
import json
from datetime import date, timedelta
from typing import List, Optional

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

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

QUESTION_MIN_ID = 385


def _difficulty_text_to_level(v: str) -> int:
    m = {"easy": 1, "medium": 2, "hard": 3}
    return m.get((v or "").lower(), 2)


def _sync_questions_from_chemistry_bank(db: Session, min_question_id: int = QUESTION_MIN_ID) -> None:
    """
    将 chemistry_question_bank 中可做选择题（question_id >= min_question_id）同步到 questions，
    以保持现有 answer_records / 推荐 / 弱点更新链路不变。
    """
    rows = db.execute(
        text(
            """
            SELECT question_id, difficulty, knowledge_tags, content, answer_schema
            FROM chemistry_question_bank
            WHERE question_id >= :min_id AND type = 'choice'
            """
        ),
        {"min_id": min_question_id},
    ).mappings().all()
    if not rows:
        return

    existing_ids = {
        r[0] for r in db.query(Question.id).filter(Question.id >= min_question_id).all()
    }

    inserted = 0
    for row in rows:
        qid = int(row["question_id"])
        if qid in existing_ids:
            continue

        answer_schema = row.get("answer_schema")
        if isinstance(answer_schema, str):
            try:
                answer_schema = json.loads(answer_schema)
            except Exception:
                answer_schema = {}
        elif answer_schema is None:
            answer_schema = {}

        options = answer_schema.get("options") or {}
        correct = answer_schema.get("correct")
        if isinstance(correct, list):
            correct = correct[0] if correct else None
        if not (options.get("A") and options.get("B") and options.get("C") and options.get("D") and correct):
            continue

        db.add(
            Question(
                id=qid,
                content=row.get("content") or "",
                option_a=str(options.get("A")),
                option_b=str(options.get("B")),
                option_c=str(options.get("C")),
                option_d=str(options.get("D")),
                correct_option=str(correct).strip()[:1],
                knowledge_point=(row.get("knowledge_tags") or None),
                difficulty=_difficulty_text_to_level(row.get("difficulty")),
            )
        )
        existing_ids.add(qid)
        inserted += 1

    if inserted > 0:
        db.commit()


def get_question_analysis(db: Session, question_id: int) -> str:
    """按 question_id 从 chemistry_question_bank 查询题目解析。"""
    row = db.execute(
        text(
            """
            SELECT analysis, hint
            FROM chemistry_question_bank
            WHERE question_id = :qid
            LIMIT 1
            """
        ),
        {"qid": question_id},
    ).mappings().first()
    if not row:
        return ""
    return (row.get("analysis") or row.get("hint") or "").strip()


def get_user_proficiency(db: Session, user_id: int) -> dict:
    """知识点熟练度"""
    _sync_questions_from_chemistry_bank(db)
    rows = (
        db.query(
            Question.knowledge_point,
            func.sum(case((AnswerRecord.is_correct == True, 1), else_=0)).label("correct"),
            func.count(AnswerRecord.id).label("total"),
        )
        .join(AnswerRecord, Question.id == AnswerRecord.question_id)
        .filter(
            AnswerRecord.user_id == user_id,
            Question.knowledge_point.isnot(None),
            Question.knowledge_point != "",
        )
        .group_by(Question.knowledge_point)
        .all()
    )
    return {r.knowledge_point: int(r.correct / r.total * 100) if r.total else 0 for r in rows}


def get_wrong_question_ids(db: Session, user_id: int, limit: int = 50) -> List[int]:
    """错题集 ID 列表"""
    records = (
        db.query(AnswerRecord.question_id)
        .join(Question, Question.id == AnswerRecord.question_id)
        .filter(AnswerRecord.user_id == user_id, AnswerRecord.is_correct == False)
        .filter(Question.id >= QUESTION_MIN_ID)
        .order_by(desc(AnswerRecord.answered_at))
        .all()
    )
    seen = set()
    ids = []
    for (qid,) in records:
        if qid not in seen:
            seen.add(qid)
            ids.append(qid)
        if len(ids) >= limit:
            break
    return ids


def get_wrong_questions_for_user(db: Session, user_id: int, limit: int = 50) -> List[Question]:
    """
    错题集（第二阶段）：
    1）优先从 wrong_questions 表取“最近答错原题”
    2）再用 question_variants 做“错题→同知识点变式题”
    3）如果变式数据为空，则退回“原题重做”
    """
    _sync_questions_from_chemistry_bank(db)
    # wrong_questions 里取最近错题（base question）
    base_qids_rows = (
        db.query(WrongQuestion.question_id)
        .filter(WrongQuestion.user_id == user_id)
        .filter(WrongQuestion.question_id >= QUESTION_MIN_ID)
        .order_by(desc(func.coalesce(WrongQuestion.last_wrong_at, WrongQuestion.created_at)))
        .limit(max(limit * 2, 20))
        .all()
    )
    base_qids = []
    seen_base = set()
    for (qid,) in base_qids_rows:
        if qid not in seen_base:
            seen_base.add(qid)
            base_qids.append(qid)

    if not base_qids:
        # 兼容：表未初始化时退回 answer_records 逻辑
        wrong_ids = get_wrong_question_ids(db, user_id, limit)
        questions = db.query(Question).filter(Question.id.in_(wrong_ids)).all() if wrong_ids else []
        id_order = {qid: i for i, qid in enumerate(wrong_ids)}
        return sorted(questions, key=lambda q: id_order.get(q.id, 999))

    result_qids: List[int] = []
    seen_qids = set()

    for base_qid in base_qids:
        # 从变式映射中优先拿 variant_question_id
        variant_rows = (
            db.query(QuestionVariant.variant_question_id)
            .filter(QuestionVariant.base_question_id == base_qid)
            .order_by(desc(QuestionVariant.weight))
            .limit(10)
            .all()
        )
        added_any = False
        for (vid,) in variant_rows:
            if vid == base_qid:
                continue
            if vid in seen_qids:
                continue
            result_qids.append(vid)
            seen_qids.add(vid)
            added_any = True
            if len(result_qids) >= limit:
                break

        if len(result_qids) >= limit:
            break

        # 不足则加入原错题
        if not added_any or (len(result_qids) < limit):
            if base_qid not in seen_qids:
                result_qids.append(base_qid)
                seen_qids.add(base_qid)
                if len(result_qids) >= limit:
                    break

    # 拉取题目并保持 result_qids 顺序
    questions = db.query(Question).filter(Question.id.in_(result_qids)).all() if result_qids else []
    id_order = {qid: i for i, qid in enumerate(result_qids)}
    return sorted(questions, key=lambda q: id_order.get(q.id, 999))


def get_recommended_questions(
    db: Session, user_id: int, count: int = 5
) -> List[Question]:
    """习题测评推荐算法：错题优先、薄弱知识点补充"""
    _sync_questions_from_chemistry_bank(db)
    wrong_ids = get_wrong_question_ids(db, user_id, 20)
    prof = get_user_proficiency(db, user_id)
    weak = [k for k, v in sorted(prof.items(), key=lambda x: x[1])[:3]]

    recommended = []
    if wrong_ids:
        qs = (
            db.query(Question)
            .filter(Question.id.in_(wrong_ids), Question.id >= QUESTION_MIN_ID)
            .order_by(func.rand())
            .limit(count)
            .all()
        )
        recommended = [q.id for q in qs]

    exclude = list(set(recommended + wrong_ids))
    if len(recommended) < count and weak:
        extra = db.query(Question).filter(Question.knowledge_point.in_(weak), Question.id >= QUESTION_MIN_ID)
        if exclude:
            extra = extra.filter(~Question.id.in_(exclude))
        extra = extra.order_by(func.rand()).limit(count - len(recommended)).all()
        recommended.extend([q.id for q in extra])
        exclude = list(set(recommended + wrong_ids))

    if len(recommended) < count:
        fallback = db.query(Question).filter(Question.id >= QUESTION_MIN_ID)
        if exclude:
            fallback = fallback.filter(~Question.id.in_(exclude))
        fallback = fallback.order_by(func.rand()).limit(count - len(recommended)).all()
        recommended.extend([q.id for q in fallback])

    if not recommended:
        return []
    return db.query(Question).filter(Question.id.in_(recommended[:count])).all()


# --- 每日一练：基于弱点权重的推荐 ---
TOP_WEAKNESS_N = 5
MASTERY_EXCLUDE_THRESHOLD = 80


def _get_question_ids_for_knowledge_point(
    db: Session,
    user_id: int,
    knowledge_point_id: int,
    exclude_ids: List[int],
    limit: int,
    preferred_difficulties: Optional[List[int]] = None,
    prefer_variants: bool = True,
) -> List[int]:
    """
    _sync_questions_from_chemistry_bank(db)
    获取某知识点下的题目 ID：
    1）优先用错题变式（question_variants + wrong_questions）
    2）再用 question_knowledge_points
    3）回退 Question.knowledge_point
    """
    kp = db.query(KnowledgePoint).filter(KnowledgePoint.id == knowledge_point_id).first()
    kp_name = kp.name if kp else None

    qids = []

    # 0) 优先取错题的变式题（可在诊断测评中关闭）
    if prefer_variants:
        try:
            base_wrong_subq = db.query(WrongQuestion.question_id).filter(
                WrongQuestion.user_id == user_id
            ).subquery()

            variant_q = db.query(
                QuestionVariant.variant_question_id, QuestionVariant.weight
            ).filter(
                QuestionVariant.base_question_id.in_(base_wrong_subq),
                QuestionVariant.variant_question_id >= QUESTION_MIN_ID,
                (QuestionVariant.knowledge_point_id == knowledge_point_id)
                | (QuestionVariant.knowledge_point_id.is_(None)),
            )

            if preferred_difficulties:
                variant_q = variant_q.join(
                    Question, Question.id == QuestionVariant.variant_question_id
                ).filter(Question.difficulty.in_(preferred_difficulties))

            variant_rows = (
                variant_q.filter(~QuestionVariant.variant_question_id.in_(exclude_ids))
                .order_by(desc(QuestionVariant.weight))
                .limit(limit)
                .all()
            )

            for qid, _w in variant_rows:
                if qid in exclude_ids or qid in qids:
                    continue
                qids.append(qid)
                if len(qids) >= limit:
                    return qids
        except Exception:
            # 变式表为空/列不匹配等情况不阻塞主流程
            pass

    # 1) 从 question_knowledge_points
    qkps_q = (
        db.query(QuestionKnowledgePoint.question_id)
        .filter(QuestionKnowledgePoint.knowledge_point_id == knowledge_point_id)
        .filter(QuestionKnowledgePoint.question_id >= QUESTION_MIN_ID)
        .distinct()
    )
    if preferred_difficulties:
        qkps_q = qkps_q.join(
            Question, Question.id == QuestionKnowledgePoint.question_id
        ).filter(Question.difficulty.in_(preferred_difficulties))
    qkps = qkps_q.all()
    for (qid,) in qkps:
        if qid not in exclude_ids:
            qids.append(qid)
        if len(qids) >= limit:
            return qids

    # 2) 回退：Question.knowledge_point 匹配
    if kp_name:
        qs = (
            db.query(Question.id)
            .filter(Question.knowledge_point == kp_name)
            .filter(~Question.id.in_(exclude_ids + qids))
            .all()
        )
        # 注意：上面 .all() 已经把数据拉出来，这里为了保持原逻辑简单不再回填难度过滤
        # （真正需要时建议把这一段改成不 .all() 的链式查询）
        # 但为了保证准确，我们用 Python 再做难度过滤
        if preferred_difficulties:
            qs = (
                db.query(Question.id)
                .filter(Question.knowledge_point == kp_name)
                .filter(~Question.id.in_(exclude_ids + qids))
                .filter(Question.difficulty.in_(preferred_difficulties))
                .order_by(func.rand())
                .limit(limit - len(qids))
                .all()
            )
        else:
            qs = (
                db.query(Question.id)
                .filter(Question.knowledge_point == kp_name)
                .filter(~Question.id.in_(exclude_ids + qids))
                .order_by(func.rand())
                .limit(limit - len(qids))
                .all()
            )
        for (qid,) in qs:
            qids.append(qid)
    return qids


def get_or_generate_daily_recommendations(
    db: Session,
    user_id: int,
    count: int = 10,
    today: date = None,
) -> List[Question]:
    """
    _sync_questions_from_chemistry_bank(db)
    获取今日推荐题，若不存在则按弱点权重生成并写入 daily_recommendations。
    策略：weakness_weight 降序取 Top N 知识点 → 每知识点选题 → 不足则随机补题。
    """
    if today is None:
        today = date.today()

    # 1) 已有今日推荐：
    # - 若数量足够：直接返回
    # - 若数量不足：在保留已存在记录的前提下补齐剩余题目
    recs = (
        db.query(DailyRecommendation)
        .filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.recommend_date == today,
        )
        .order_by(DailyRecommendation.id.asc())
        .all()
    )
    if recs and len(recs) >= count:
        qids = [r.question_id for r in recs]
        questions = (
            db.query(Question)
            .filter(Question.id.in_(qids), Question.id >= QUESTION_MIN_ID)
            .all()
        )
        id_order = {qid: i for i, qid in enumerate(qids)}
        return sorted(questions, key=lambda q: id_order.get(q.id, 999))
    existing_qids: List[int] = [r.question_id for r in recs] if recs else []
    missing_count = max(0, count - len(existing_qids))
    if missing_count == 0 and existing_qids:
        questions = db.query(Question).filter(Question.id.in_(existing_qids), Question.id >= QUESTION_MIN_ID).all()
        id_order = {qid: i for i, qid in enumerate(existing_qids)}
        return sorted(questions, key=lambda q: id_order.get(q.id, 999))

    # 2) 生成新推荐
    # 2a) 弱点知识点：weakness_weight 降序，排除 mastery_score > 80
    weakness_rows = (
        db.query(StudentKnowledgeMastery)
        .filter(StudentKnowledgeMastery.user_id == user_id)
        .filter(
            (StudentKnowledgeMastery.mastery_score.is_(None))
            | (StudentKnowledgeMastery.mastery_score <= MASTERY_EXCLUDE_THRESHOLD)
        )
        .order_by(desc(StudentKnowledgeMastery.weakness_weight))
        .limit(TOP_WEAKNESS_N)
        .all()
    )

    # 2b) 近期已做过的题（7 天内）避免重复
    from datetime import datetime, timedelta
    since = datetime.utcnow() - timedelta(days=7)
    recent_qids = [
        r[0]
        for r in db.query(AnswerRecord.question_id)
        .filter(AnswerRecord.user_id == user_id, AnswerRecord.answered_at >= since)
        .filter(AnswerRecord.question_id >= QUESTION_MIN_ID)
        .distinct()
        .all()
    ]
    # exclude 需要同时排除：已存在的今日题 + 近期做过的题
    exclude = list(set(existing_qids + recent_qids))

    recommended_qids = []
    # 仅补齐缺失数量
    per_kp = max(2, missing_count // max(1, len(weakness_rows)))  # 每知识点至少 2 题
    target_count = missing_count

    for idx, row in enumerate(weakness_rows):
        # 难度匹配：按掌握度从中等难度起步，逐步向更高难度推进
        # 约定：Question.difficulty 1/2/3 对应 easy/medium/hard（若你的题库含义不同可再调）
        if row.mastery_score is None:
            preferred = [2, 3]
        elif row.mastery_score <= 35:
            preferred = [1, 2]
        elif row.mastery_score <= 65:
            preferred = [2]
        else:
            preferred = [2, 3]

        ids = _get_question_ids_for_knowledge_point(
            db,
            user_id,
            row.knowledge_point_id,
            exclude,
            per_kp,
            preferred_difficulties=preferred,
            prefer_variants=True,
        )
        for qid in ids:
            if qid not in recommended_qids:
                recommended_qids.append(qid)
                exclude.append(qid)
        if len(recommended_qids) >= target_count:
            break

    # 2c) 不足则随机补题
    if len(recommended_qids) < target_count:
        fallback = (
            db.query(Question)
            .filter(~Question.id.in_(exclude), Question.id >= QUESTION_MIN_ID)
            .order_by(func.rand())
            .limit(target_count - len(recommended_qids))
            .all()
        )
        for q in fallback:
            recommended_qids.append(q.id)
            exclude.append(q.id)

    # 2d) 兜底：题库较小时，若“近7天排除”导致候选不足，则允许重复旧题补齐
    if len(recommended_qids) < target_count:
        extra = (
            db.query(Question)
            .filter(Question.id >= QUESTION_MIN_ID, ~Question.id.in_(recommended_qids))
            .order_by(func.rand())
            .limit(target_count - len(recommended_qids))
            .all()
        )
        for q in extra:
            recommended_qids.append(q.id)

    recommended_qids = recommended_qids[:target_count]

    # 3) 写入 daily_recommendations
    strategy = "weak_kp" if weakness_rows else "mixed"
    for qid in recommended_qids:
        dr = DailyRecommendation(
            user_id=user_id,
            question_id=qid,
            recommend_date=today,
            strategy=strategy,
            is_done=0,
        )
        db.add(dr)
    db.commit()

    # 返回：今日所有推荐题（包含已存在的记录 + 本次新增的补齐题）
    all_recs = (
        db.query(DailyRecommendation)
        .filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.recommend_date == today,
        )
        .order_by(DailyRecommendation.id.asc())
        .all()
    )
    qids = [r.question_id for r in all_recs]
    questions = (
        db.query(Question)
        .filter(Question.id.in_(qids), Question.id >= QUESTION_MIN_ID)
        .all()
    )
    id_order = {qid: i for i, qid in enumerate(qids)}
    return sorted(questions, key=lambda q: id_order.get(q.id, 999))


def _get_preferred_difficulties_for_session(session_kind: str) -> List[int]:
    """
    诊断/再测评的难度偏好（先用适中难度覆盖，再逐步提升）。
    约定 Question.difficulty 1/2/3 对应 easy/medium/hard。
    """
    if session_kind == "re_evaluate":
        return [2, 3]
    return [2]


def get_diagnostic_questions(
    db: Session,
    user_id: int,
    count: int = 10,
    session_kind: str = "diagnostic",
) -> List[Question]:
    """
    诊断测评/再测评选题：
    - 选取弱点知识点（weakness_weight 高优先），覆盖广
    - 诊断中关闭“错题变式优先”（避免变式题影响诊断覆盖）
    - 难度按 session_kind 偏好过滤
    """
    _sync_questions_from_chemistry_bank(db)
    preferred = _get_preferred_difficulties_for_session(session_kind=session_kind)

    from datetime import datetime, timedelta

    since = datetime.utcnow() - timedelta(days=7)
    recent_qids = [
        r[0]
        for r in db.query(AnswerRecord.question_id)
        .filter(AnswerRecord.user_id == user_id, AnswerRecord.answered_at >= since)
        .filter(AnswerRecord.question_id >= QUESTION_MIN_ID)
        .distinct()
        .all()
    ]
    exclude = list(set(recent_qids))

    kp_rows = (
        db.query(StudentKnowledgeMastery)
        .filter(StudentKnowledgeMastery.user_id == user_id)
        .order_by(desc(StudentKnowledgeMastery.weakness_weight))
        .limit(max(3, count // 2))
        .all()
    )

    # 用户全新/数据少：退回按难度随机抽
    if not kp_rows:
        fallback_q = db.query(Question).filter(Question.difficulty.in_(preferred), Question.id >= QUESTION_MIN_ID)
        if exclude:
            fallback_q = fallback_q.filter(~Question.id.in_(exclude))
        qs = fallback_q.order_by(func.rand()).limit(count).all()
        if qs:
            return qs
        # 兜底：如果题库缺少 preferred 难度，退回到全难度随机
        fallback_q = db.query(Question).filter(Question.id >= QUESTION_MIN_ID)
        if exclude:
            fallback_q = fallback_q.filter(~Question.id.in_(exclude))
        qs = fallback_q.order_by(func.rand()).limit(count).all()
        if qs:
            return qs
        # 再兜底：如果“近7天排除”后为空，允许重复题
        return (
            db.query(Question)
            .filter(Question.id >= QUESTION_MIN_ID)
            .order_by(func.rand())
            .limit(count)
            .all()
        )

    recommended_qids: List[int] = []
    per_kp = max(1, count // max(1, len(kp_rows)))

    for row in kp_rows:
        ids = _get_question_ids_for_knowledge_point(
            db,
            user_id,
            row.knowledge_point_id,
            exclude,
            per_kp,
            preferred_difficulties=preferred,
            prefer_variants=False,
        )
        for qid in ids:
            if qid in recommended_qids:
                continue
            recommended_qids.append(qid)
            exclude.append(qid)
            if len(recommended_qids) >= count:
                break
        if len(recommended_qids) >= count:
            break

    if len(recommended_qids) < count:
        fallback_q = db.query(Question).filter(Question.difficulty.in_(preferred), Question.id >= QUESTION_MIN_ID)
        if exclude:
            fallback_q = fallback_q.filter(~Question.id.in_(exclude))
        fallback = fallback_q.order_by(func.rand()).limit(count - len(recommended_qids)).all()
        for q in fallback:
            recommended_qids.append(q.id)
            exclude.append(q.id)

    # 兜底：若排除近7天后仍不足，允许重复旧题补齐
    if len(recommended_qids) < count:
        fallback = (
            db.query(Question)
            .filter(
                Question.id >= QUESTION_MIN_ID,
                ~Question.id.in_(recommended_qids),
            )
            .order_by(func.rand())
            .limit(count - len(recommended_qids))
            .all()
        )
        for q in fallback:
            recommended_qids.append(q.id)

    recommended_qids = recommended_qids[:count]
    if not recommended_qids:
        fallback_q = db.query(Question).filter(Question.id >= QUESTION_MIN_ID)
        if exclude:
            fallback_q = fallback_q.filter(~Question.id.in_(exclude))
        qs = fallback_q.order_by(func.rand()).limit(count).all()
        if qs:
            return qs
        # 最终兜底：完全不排除近7天
        return (
            db.query(Question)
            .filter(Question.id >= QUESTION_MIN_ID)
            .order_by(func.rand())
            .limit(count)
            .all()
        )

    questions = db.query(Question).filter(Question.id.in_(recommended_qids)).all()
    id_order = {qid: i for i, qid in enumerate(recommended_qids)}
    return sorted(questions, key=lambda q: id_order.get(q.id, 999))


def _ensure_knowledge_points_from_questions(db: Session) -> None:
    """若 knowledge_points 为空，则从 Question.knowledge_point 同步创建"""
    if db.query(KnowledgePoint).first():
        return
    kp_names = (
        db.query(Question.knowledge_point)
        .filter(
            Question.knowledge_point.isnot(None),
            Question.knowledge_point != "",
        )
        .distinct()
        .all()
    )
    for (name,) in kp_names:
        if name and not db.query(KnowledgePoint).filter(KnowledgePoint.name == name).first():
            db.add(KnowledgePoint(name=name))
    db.commit()


def get_mastery_matrix(db: Session, user_id: int) -> List[dict]:
    """
    学生端知识点掌握度矩阵数据，用于可视化（颜色分级、雷达图）。
    返回：[{ knowledgePointId, name, masteryScore, weaknessWeight, chapter }]
    """
    _ensure_knowledge_points_from_questions(db)
    rows = (
        db.query(
            KnowledgePoint.id,
            KnowledgePoint.name,
            KnowledgePoint.chapter,
            StudentKnowledgeMastery.mastery_score,
            StudentKnowledgeMastery.weakness_weight,
            StudentKnowledgeMastery.wrong_count,
            StudentKnowledgeMastery.last_practiced_at,
        )
        .outerjoin(
            StudentKnowledgeMastery,
            (StudentKnowledgeMastery.knowledge_point_id == KnowledgePoint.id)
            & (StudentKnowledgeMastery.user_id == user_id),
        )
        .all()
    )
    return [
        {
            "knowledgePointId": r.id,
            "name": r.name,
            "chapter": r.chapter or "",
            "masteryScore": r.mastery_score if r.mastery_score is not None else 0,
            "weaknessWeight": r.weakness_weight if r.weakness_weight is not None else 0,
            "wrongCount": r.wrong_count if r.wrong_count is not None else 0,
            "lastPracticedAt": (
                r.last_practiced_at.isoformat() if r.last_practiced_at else None
            ),
        }
        for r in rows
    ]


def get_motivation_summary(db: Session, user_id: int) -> dict:
    """
    学习激励汇总（不新增表）：
    - streakDays: 连续打卡天数（存在 is_done=1 的 daily_recommendations 记为当日打卡）
    - greenModules: 全绿模块（章节下所有知识点 mastery_score >= 80）
    - moduleAverages: 章节平均掌握度（用于雷达图/柱状图）
    """
    matrix = get_mastery_matrix(db, user_id)
    from datetime import datetime, timedelta

    # 1) 按章节聚合
    by_chapter: dict[str, list[dict]] = {}
    for m in matrix:
        ch = (m.get("chapter") or "").strip() or "其他"
        by_chapter.setdefault(ch, []).append(m)

    module_averages = []
    green_modules = []
    for chapter, items in by_chapter.items():
        scores = [int(x.get("masteryScore") or 0) for x in items]
        if not scores:
            continue
        avg = int(round(sum(scores) / len(scores)))
        module_averages.append({"chapter": chapter, "avgScore": avg})
        if all(s >= 80 for s in scores):
            green_modules.append(chapter)

    module_averages.sort(key=lambda x: x["avgScore"], reverse=True)

    # 2) 连续打卡（从今天往前，只要当日有完成记录即算打卡）
    done_dates = {
        d[0]
        for d in db.query(DailyRecommendation.recommend_date)
        .filter(
            DailyRecommendation.user_id == user_id,
            DailyRecommendation.is_done == 1,
            DailyRecommendation.recommend_date.isnot(None),
        )
        .distinct()
        .all()
    }

    streak_days = 0
    cursor = date.today()
    while cursor in done_dates:
        streak_days += 1
        cursor = cursor - timedelta(days=1)

    # 3) 最近练习章节掌握度（过滤掉未练习知识点）
    # - “未练习”在这里等价于 last_practiced_at 为空
    # - 用最近 7 天作为口径（与其它推荐逻辑一致）
    since = datetime.utcnow() - timedelta(days=7)
    recent_rows = (
        db.query(
            KnowledgePoint.chapter,
            StudentKnowledgeMastery.mastery_score,
        )
        .join(
            StudentKnowledgeMastery,
            (StudentKnowledgeMastery.knowledge_point_id == KnowledgePoint.id)
            & (StudentKnowledgeMastery.user_id == user_id),
        )
        .filter(StudentKnowledgeMastery.last_practiced_at.isnot(None))
        .filter(StudentKnowledgeMastery.last_practiced_at >= since)
        .all()
    )

    recent_by_ch: dict[str, list[int]] = {}
    for ch, ms in recent_rows:
        ch_name = (ch or "").strip() or "其他"
        recent_by_ch.setdefault(ch_name, []).append(int(ms or 0))

    recent_module_averages = []
    recent_green_modules = []
    for chapter, items in recent_by_ch.items():
        if not items:
            continue
        avg = int(round(sum(items) / len(items)))
        recent_module_averages.append({"chapter": chapter, "avgScore": avg})
        if all(s >= 80 for s in items):
            recent_green_modules.append(chapter)

    recent_module_averages.sort(key=lambda x: x["avgScore"], reverse=True)

    return {
        "streakDays": streak_days,
        "greenModules": green_modules,
        "moduleAverages": module_averages,
        "moduleAveragesRecent": recent_module_averages,
    }