test/app/api/score_api.py

"""成绩管理接口（需求 2.2、4.2）。"""

from __future__ import annotations

from typing import Annotated

from fastapi import APIRouter, Query

from app.core.config import settings
from app.core.deps import DbSession, ReadAccount, WriteAccount
from app.core.response import Resp, ok
from app.dao.score_dao import ScoreDao
from app.dao.student_dao import StudentDao
from app.schema.common import PageResult
from app.schema.score_schema import ScoreBatchCreate, ScoreCreate, ScoreOut, ScoreRecordResult, ScoreUpdate
from app.service.score_service import ScoreService

router = APIRouter(prefix="/scores", tags=["2.2 学生考核成绩管理"])


@router.get("", response_model=Resp[PageResult[ScoreOut]], summary="成绩列表（多条件筛选）")
def list_scores(
    db: DbSession,
    _: ReadAccount,
    page: Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=200)] = 10,
    stu_id: Annotated[int | None, Query(description="学生ID")] = None,
    exam_seq: Annotated[int | None, Query(description="考核序次")] = None,
    class_id: Annotated[int | None, Query(description="班级ID")] = None,
    flag: Annotated[int | None, Query(ge=0, le=1, description="1=只看预警")] = None,
    min_score: Annotated[float | None, Query(description="分数下限")] = None,
    max_score: Annotated[float | None, Query(description="分数上限")] = None,
    keyword: Annotated[str | None, Query(description="学生姓名/学号")] = None,
    order_by: Annotated[str, Query()] = "id",
    order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
):
    stmt = ScoreDao.build_stmt(
        stu_id=stu_id, exam_seq=exam_seq, class_id=class_id, flag=flag,
        min_score=min_score, max_score=max_score, keyword=keyword, order_by=order_by, order=order,
    )
    items, total, page, pages = ScoreDao.paginate(db, stmt, page, page_size)
    return ok({
        "total": total, "page": page, "page_size": page_size, "pages": pages,
        "items": [ScoreOut.model_validate(s) for s in items],
    })


@router.get("/meta", summary="成绩模块元信息：已用过的考核序次 / 红线 / 课程")
def score_meta(db: DbSession, _: ReadAccount):
    from sqlalchemy import select

    from app.model import Score

    exam_names = [
        name
        for name in db.scalars(
            select(Score.exam_name).where(Score.alive(), Score.exam_name.is_not(None)).distinct()
        ).all()
        if name
    ]
    return ok({
        "exam_seqs": ScoreDao.exam_seq_list(db),
        "max_exam_seq": ScoreDao.max_exam_seq(db),
        "exam_names": exam_names,
        "pass_line": settings.SCORE_PASS_LINE,
        "warn_line": settings.SCORE_WARN_LINE,
    })


@router.get("/student/{stu_id}", summary="某个学生的全部成绩 + 汇总（需求 4.2 的 GET /score/{stu_id}）")
def scores_of_student(db: DbSession, _: ReadAccount, stu_id: int):
    StudentDao.get_or_404(db, stu_id, "学生")
    return ok(ScoreService.student_summary(db, stu_id))


@router.get("/warnings", summary="红线预警名单（低于红线的学生汇总）")
def warning_list(
    db: DbSession,
    _: ReadAccount,
    page: Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=200)] = 10,
):
    stmt = ScoreDao.build_stmt(flag=1, order_by="score", order="asc")
    items, total, page, pages = ScoreDao.paginate(db, stmt, page, page_size)
    return ok({
        "total": total, "page": page, "page_size": page_size, "pages": pages,
        "warn_line": settings.SCORE_WARN_LINE,
        "items": [ScoreOut.model_validate(s) for s in items],
    })


@router.post("", response_model=Resp[ScoreRecordResult], summary="录入成绩（低于红线自动触发预警）")
def create_score(db: DbSession, _: WriteAccount, payload: ScoreCreate):
    score, warning, warning_msg = ScoreService.create(db, payload)
    db.commit()
    db.refresh(score)
    summary = ScoreService.student_summary(db, score.stu_id)
    return ok(
        {
            "record": ScoreOut.model_validate(score),
            "warning": warning,
            "warning_msg": warning_msg,
            "avg_score": summary["avg"],
        },
        msg=warning_msg or "成绩录入成功",
    )


@router.post("/batch", summary="一次录入同一个学生的多次成绩")
def create_scores_batch(db: DbSession, _: WriteAccount, payload: ScoreBatchCreate):
    results = ScoreService.batch_create(db, payload.stu_id, payload.scores)
    db.commit()
    items = []
    for score, warning, warning_msg in results:
        db.refresh(score)
        items.append(
            {
                "record": ScoreOut.model_validate(score),
                "warning": warning,
                "warning_msg": warning_msg,
            }
        )
    warnings = [i for i in items if i["warning"]]
    return ok(
        {"count": len(items), "warning_count": len(warnings), "items": items},
        msg=f"录入 {len(items)} 条成绩" + (f"，其中 {len(warnings)} 条触发红线预警" if warnings else ""),
    )


@router.put("/{score_id}", response_model=Resp[ScoreRecordResult], summary="修改成绩（重新计算预警标记）")
def update_score(db: DbSession, _: WriteAccount, score_id: int, payload: ScoreUpdate):
    score = ScoreDao.get_or_404(db, score_id, "成绩记录")
    score, warning, warning_msg = ScoreService.update(db, score, payload)
    db.commit()
    db.refresh(score)
    summary = ScoreService.student_summary(db, score.stu_id)
    return ok(
        {
            "record": ScoreOut.model_validate(score),
            "warning": warning,
            "warning_msg": warning_msg,
            "avg_score": summary["avg"],
        },
        msg="修改成功" + (f"（{warning_msg}）" if warning_msg else ""),
    )


@router.delete("/{score_id}", summary="删除成绩（逻辑删除，同序次可重新录入）")
def delete_score(db: DbSession, _: WriteAccount, score_id: int):
    score = ScoreDao.get_or_404(db, score_id, "成绩记录")
    student_name = score.student_name
    exam_seq = score.exam_seq
    ScoreService.delete(db, score)
    db.commit()
    return ok(msg=f"已删除 {student_name} 第 {exam_seq} 次考核成绩")


@router.post("/student/{stu_id}/clear", summary="清空某个学生的全部成绩")
def clear_student_scores(db: DbSession, _: WriteAccount, stu_id: int):
    StudentDao.get_or_404(db, stu_id, "学生")
    scores = ScoreDao.list_by_student(db, stu_id)
    for score in scores:
        score.soft_delete()
    db.commit()
    return ok({"deleted": len(scores)}, msg=f"已清空 {len(scores)} 条成绩")