item_bank/app/api/notebook.py

"""首页概览与复习清单。

刻意不做正确率统计：这是笔记本，不是刷题软件。需要回头看的东西由
"我标记的" ∪ "最近一次答错的" 决定，而不是分数。
"""
from fastapi import APIRouter, Depends, Query
from sqlalchemy import Integer, case, desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.db import get_db
from app.deps import get_current_user
from app.models.practice import PracticeRecord
from app.models.question import Question, QuestionTag, Tag
from app.models.user import User
from app.schemas.notebook import NotebookOverview, ReviewItem, TagStat
from app.schemas.question import QuestionOut

router = APIRouter(prefix="/api/notebook", tags=["notebook"])


def _latest_wrong_subquery(user_id: str):
    """最近一次作答判错的题目 id。

    created_at 是 Python 侧生成、带微秒的（见 app/db.py），所以同一秒内
    多次作答也能正确排序。
    """
    ranked = (
        select(
            PracticeRecord.question_id,
            PracticeRecord.is_correct,
            func.row_number()
            .over(
                partition_by=PracticeRecord.question_id,
                order_by=PracticeRecord.created_at.desc(),
            )
            .label("rn"),
        )
        .where(PracticeRecord.user_id == user_id)
        .subquery()
    )
    return (
        select(ranked.c.question_id)
        .where(ranked.c.rn == 1, ranked.c.is_correct.is_(False))
        .subquery()
    )


@router.get("/overview", response_model=NotebookOverview)
async def overview(
    db: AsyncSession = Depends(get_db),
    current: User = Depends(get_current_user),
    recent_limit: int = Query(10, ge=1, le=50),
):
    """首页：以知识点导航为主。"""
    total_notes = await db.scalar(
        select(func.count(Question.id)).where(Question.user_id == current.id)
    )

    latest_wrong = _latest_wrong_subquery(current.id)
    wrong_ids = select(latest_wrong.c.question_id)

    # 复习清单大小 = 手动标记 ∪ 最近答错
    review_count = await db.scalar(
        select(func.count(Question.id)).where(
            Question.user_id == current.id,
            (Question.needs_review.is_(True)) | (Question.id.in_(wrong_ids)),
        )
    )

    untagged_count = await db.scalar(
        select(func.count(Question.id)).where(
            Question.user_id == current.id, ~Question.tags.any()
        )
    )

    # 每个知识点：总数 + 其中待复习数
    tag_rows = (
        await db.execute(
            select(
                Tag.name,
                func.count(Question.id).label("cnt"),
                func.sum(
                    case(
                        (
                            (Question.needs_review.is_(True))
                            | (Question.id.in_(wrong_ids)),
                            1,
                        ),
                        else_=0,
                    )
                )
                .cast(Integer)
                .label("review_cnt"),
            )
            .join(QuestionTag, QuestionTag.tag_id == Tag.id)
            .join(Question, Question.id == QuestionTag.question_id)
            .where(Tag.user_id == current.id)
            .group_by(Tag.name)
            .order_by(desc("cnt"), Tag.name)
        )
    ).all()

    recent_rows = (
        await db.scalars(
            select(Question)
            .where(Question.user_id == current.id)
            .options(selectinload(Question.tags))
            .order_by(Question.created_at.desc())
            .limit(recent_limit)
        )
    ).all()

    return NotebookOverview(
        total_notes=total_notes or 0,
        review_count=review_count or 0,
        untagged_count=untagged_count or 0,
        tags=[
            TagStat(name=name, count=cnt, needs_review_count=rc or 0)
            for name, cnt, rc in tag_rows
        ],
        recent=[QuestionOut.from_model(q) for q in recent_rows],
    )


@router.get("/review", response_model=list[ReviewItem])
async def review_list(
    db: AsyncSession = Depends(get_db),
    current: User = Depends(get_current_user),
    tag: str | None = None,
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
):
    """复习清单：我手动标记的 ∪ 最近一次答错的。"""
    latest_wrong = _latest_wrong_subquery(current.id)
    wrong_ids_stmt = select(latest_wrong.c.question_id)
    wrong_ids = set((await db.scalars(wrong_ids_stmt)).all())

    stmt = (
        select(Question)
        .where(
            Question.user_id == current.id,
            (Question.needs_review.is_(True))
            | (Question.id.in_(wrong_ids_stmt)),
        )
        .options(selectinload(Question.tags))
    )
    if tag:
        stmt = stmt.where(Question.tags.any(Tag.name == tag))

    rows = (
        await db.scalars(
            stmt.order_by(Question.updated_at.desc()).limit(limit).offset(offset)
        )
    ).all()
    if not rows:
        return []

    # 答错次数与最近答错时间
    qids = [q.id for q in rows]
    agg_rows = (
        await db.execute(
            select(
                PracticeRecord.question_id,
                func.sum(case((PracticeRecord.is_correct.is_(False), 1), else_=0))
                .cast(Integer)
                .label("wrong_count"),
                func.max(
                    case(
                        (
                            PracticeRecord.is_correct.is_(False),
                            PracticeRecord.created_at,
                        )
                    )
                ).label("last_wrong_at"),
            )
            .where(
                PracticeRecord.user_id == current.id,
                PracticeRecord.question_id.in_(qids),
            )
            .group_by(PracticeRecord.question_id)
        )
    ).all()
    agg = {r.question_id: r for r in agg_rows}

    items = []
    for q in rows:
        a = agg.get(q.id)
        items.append(
            ReviewItem(
                question=QuestionOut.from_model(q),
                marked=q.needs_review,
                last_wrong=q.id in wrong_ids,
                wrong_count=(a.wrong_count if a else 0) or 0,
                last_wrong_at=(
                    a.last_wrong_at.isoformat() if a and a.last_wrong_at else None
                ),
            )
        )
    return items