from fastapi import APIRouter, Depends, HTTPException, Query
|
from sqlalchemy import func, select
|
from sqlalchemy.orm import Session
|
|
from app.auth import get_current_user
|
from app.database import get_db
|
from app.grader_scope import allowed_subjects_for_user, apply_subject_scope
|
from app.models import (
|
ExamPaperMap,
|
ExaminationExamExaminee,
|
ExaminationExaminee,
|
ExaminationPaperSource,
|
ImportBatch,
|
User,
|
)
|
from app.schemas import ScoreRow, ScoresResponse
|
|
router = APIRouter(prefix="/api", tags=["scores"])
|
|
|
@router.get("/subjects")
|
def list_subjects(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
|
q = select(ImportBatch.subject_name).distinct().order_by(ImportBatch.subject_name)
|
allowed = allowed_subjects_for_user(db, user)
|
if allowed is not None:
|
if not allowed:
|
return {"items": []}
|
q = q.where(ImportBatch.subject_name.in_(allowed))
|
rows = db.scalars(q)
|
return {"items": list(rows)}
|
|
|
@router.get("/scores", response_model=ScoresResponse)
|
def list_scores(
|
page: int = Query(1, ge=1),
|
page_size: int = Query(20, ge=1, le=100),
|
name: str | None = None,
|
id_card: str | None = None,
|
subject: str | None = None,
|
db: Session = Depends(get_db),
|
user: User = Depends(get_current_user),
|
):
|
allowed = allowed_subjects_for_user(db, user)
|
if subject and allowed is not None and subject not in allowed:
|
raise HTTPException(403, detail="无权查看该科目")
|
q = (
|
select(
|
ExaminationExamExaminee,
|
ExaminationExaminee,
|
ImportBatch.subject_name,
|
ExaminationPaperSource.paper_name,
|
)
|
.join(ExaminationExaminee, ExaminationExaminee.id == ExaminationExamExaminee.examinee_id)
|
.join(ImportBatch, ImportBatch.id == ExaminationExamExaminee.import_batch_id)
|
.outerjoin(ExamPaperMap, ExamPaperMap.exam_id == ExaminationExamExaminee.exam_id)
|
.outerjoin(
|
ExaminationPaperSource,
|
ExaminationPaperSource.id == ExamPaperMap.paper_source_id,
|
)
|
)
|
if name:
|
q = q.where(ExaminationExaminee.nickname.contains(name))
|
if id_card:
|
q = q.where(ExaminationExaminee.id_card.contains(id_card))
|
if subject:
|
q = q.where(ImportBatch.subject_name == subject)
|
q = apply_subject_scope(q, ImportBatch.subject_name, allowed)
|
|
count_q = select(func.count()).select_from(q.subquery())
|
total = db.scalar(count_q) or 0
|
rows = db.execute(
|
q.order_by(ExaminationExamExaminee.id.desc())
|
.offset((page - 1) * page_size)
|
.limit(page_size)
|
).all()
|
|
items = [
|
ScoreRow(
|
examinee_id=str(ex.examinee_id),
|
name=ee.nickname,
|
id_card=ee.id_card,
|
subject_name=sn,
|
score=ex.score,
|
attempt_id=str(ex.id),
|
room_id=ex.room_id,
|
paper_name=pn,
|
start_time=ex.start_time,
|
end_time=ex.end_time,
|
)
|
for ex, ee, sn, pn in rows
|
]
|
return ScoresResponse(total=total, page=page, page_size=page_size, items=items)
|
