"""学生管理接口（需求 2.1、4.1）。""" from __future__ import annotations from typing import Annotated from fastapi import APIRouter, Body, File, Query, Response, UploadFile from app.core.deps import DbSession, ReadAccount, WriteAccount from app.core.response import Resp, ok from app.dao.clazz_dao import ClazzDao from app.dao.student_dao import StudentDao from app.model import Education from app.model.constants import STUDENT_STATUS_TEXT from app.schema.common import PageResult from app.schema.student_schema import ( StudentCreate, StudentImportResult, StudentOut, StudentUpdate, ) from app.service.score_service import ScoreService from app.service.student_service import StudentService router = APIRouter(prefix="/students", tags=["2.1 学生基本信息管理"]) @router.get( "", response_model=Resp[PageResult[StudentOut]], summary="学生列表（支持编号/姓名/班级/状态/性别/年龄区间筛选）", ) def list_students( db: DbSession, _: ReadAccount, page: Annotated[int, Query(ge=1, description="页码")] = 1, page_size: Annotated[int, Query(ge=1, le=200, description="每页条数")] = 10, keyword: Annotated[str | None, Query(description="模糊搜索：姓名/学号/电话/专业/院校/籍贯")] = None, class_id: Annotated[int | None, Query(description="班级ID")] = None, class_no: Annotated[str | None, Query(description="班级编号")] = None, status: Annotated[int | None, Query(ge=1, le=3, description="状态 1=在读 2=进入就业 3=已就业")] = None, gender: Annotated[int | None, Query(ge=1, le=2, description="性别 1=男 2=女")] = None, advisor_id: Annotated[int | None, Query(description="顾问ID")] = None, education: Annotated[str | None, Query(description="学历")] = None, age_min: Annotated[int | None, Query(ge=1, le=100, description="年龄下限")] = None, age_max: Annotated[int | None, Query(ge=1, le=100, description="年龄上限")] = None, order_by: Annotated[str, Query(description="排序字段 id/stu_no/name/age/enroll_date/status")] = "id", order: Annotated[str, Query(pattern="^(asc|desc)$", description="排序方向")] = "desc", ): stmt = StudentDao.build_stmt( keyword=keyword, class_id=class_id, class_no=class_no, status=status, gender=gender, advisor_id=advisor_id, education=education, age_min=age_min, age_max=age_max, order_by=order_by, order=order, ) items, total, page, pages = StudentDao.paginate(db, stmt, page, page_size) return ok({ "total": total, "page": page, "page_size": page_size, "pages": pages, "items": [StudentOut.model_validate(s) for s in items], }) @router.post("", response_model=Resp[StudentOut], summary="创建学生（学号按规则自动生成）") def create_student(db: DbSession, account: WriteAccount, payload: StudentCreate): student = StudentService.create(db, payload) db.commit() db.refresh(student) return ok(StudentOut.model_validate(student), msg=f"学生 {student.name} 创建成功，学号 {student.stu_no}") @router.get("/meta/options", summary="下拉选项：班级 / 顾问 / 学历 / 状态") def student_options(db: DbSession, _: ReadAccount): from app.dao.advisor_dao import AdvisorDao classes = ClazzDao.all(db, ClazzDao.build_stmt(order_by="class_no", order="asc")) advisors = AdvisorDao.all(db, AdvisorDao.build_stmt(order_by="id", order="asc")) return ok({ "classes": [ {"id": c.id, "name": c.name, "class_no": c.class_no, "status_text": c.status_text} for c in classes ], "advisors": [{"id": a.id, "name": a.name} for a in advisors], "educations": [e.value for e in Education], "statuses": [{"value": k, "label": v} for k, v in sorted(STUDENT_STATUS_TEXT.items())], }) @router.get("/import/template", summary="下载 Excel 导入模板") def download_template(_: ReadAccount): content = StudentService.build_import_template() return Response( content=content, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": 'attachment; filename="student_import_template.xlsx"'}, ) @router.post( "/import", response_model=Resp[StudentImportResult], summary="Excel 批量导入学生（逐行校验，失败行有行号和原因）", ) async def import_students( db: DbSession, _: WriteAccount, file: Annotated[UploadFile, File(description="按模板填写的 .xlsx")], dry_run: Annotated[bool, Query(description="只校验不落库")] = False, ): if not file.filename or not file.filename.lower().endswith((".xlsx", ".xlsm")): return Resp.fail("只支持 .xlsx 文件，请先用「下载模板」拿到标准格式") content = await file.read() result = StudentService.import_from_excel(db, content, dry_run=dry_run) return ok(result, msg=f"导入完成：成功 {result.success} 行，失败 {result.failed} 行") @router.get("/{stu_id}", response_model=Resp[StudentOut], summary="学生详情") def get_student(db: DbSession, _: ReadAccount, stu_id: int): student = StudentDao.get_or_404(db, stu_id, "学生") return ok(StudentOut.model_validate(student)) @router.get("/{stu_id}/scores", summary="学生成绩明细 + 汇总") def get_student_scores(db: DbSession, _: ReadAccount, stu_id: int): StudentDao.get_or_404(db, stu_id, "学生") return ok(ScoreService.student_summary(db, stu_id)) @router.put("/{stu_id}", response_model=Resp[StudentOut], summary="更新学生信息") def update_student(db: DbSession, _: WriteAccount, stu_id: int, payload: StudentUpdate): student = StudentDao.get_or_404(db, stu_id, "学生") StudentService.update(db, student, payload) db.commit() db.refresh(student) return ok(StudentOut.model_validate(student), msg="更新成功") @router.delete("/{stu_id}", summary="逻辑删除学生（成绩与就业记录一并逻辑删除）") def delete_student(db: DbSession, _: WriteAccount, stu_id: int): student = StudentDao.get_or_404(db, stu_id, "学生") StudentService.delete(db, student) db.commit() return ok(msg=f"已删除学生 {student.name}（可联系管理员恢复）") @router.post("/{stu_id}/restore", summary="恢复被逻辑删除的学生") def restore_student(db: DbSession, _: WriteAccount, stu_id: int): student = StudentDao.get(db, stu_id, with_deleted=True) if student is None: return Resp.fail("学生不存在", 404) if student.is_del == 0: return Resp.fail("该学生未被删除，无需恢复") student.is_del = 0 db.commit() return ok(msg=f"已恢复学生 {student.name}") @router.post("/batch/delete", summary="批量逻辑删除") def batch_delete(db: DbSession, _: WriteAccount, ids: Annotated[list[int], Body(embed=True, description="学生ID列表")]): deleted = 0 missing = [] for stu_id in ids: student = StudentDao.get(db, stu_id) if student is None: missing.append(stu_id) continue StudentService.delete(db, student) deleted += 1 db.commit() return ok({"deleted": deleted, "missing": missing}, msg=f"已删除 {deleted} 名学生")