from collections import defaultdict
|
|
from fastapi import APIRouter, Depends, HTTPException
|
from sqlalchemy import select
|
from sqlalchemy.orm import Session
|
|
from app.auth import get_current_user
|
from app.database import get_db
|
from app.grader_scope import assert_subject_allowed
|
from app.student_no import ensure_student_no, find_attempt_by_student_no
|
from app.system_settings import get_or_create_settings
|
from app.models import (
|
ExamPaperMap,
|
ExaminationExamExaminee,
|
ExaminationExaminee,
|
ExaminationExamineeExamAnswerItem,
|
ExaminationPaperItem,
|
ExaminationPaperSource,
|
ImportBatch,
|
User,
|
)
|
from app.schemas import AttemptDetail, PartGroup, QuestionDetail
|
|
router = APIRouter(prefix="/api", tags=["attempts"])
|
|
|
@router.get("/attempts/lookup/{student_no}")
|
def lookup_attempt(
|
student_no: str,
|
db: Session = Depends(get_db),
|
user: User = Depends(get_current_user),
|
):
|
"""根据编码反查答卷，返回 attempt_id 供前端跳转。"""
|
attempt = find_attempt_by_student_no(db, student_no)
|
if not attempt:
|
raise HTTPException(404, detail="编码不存在")
|
batch = db.get(ImportBatch, attempt.import_batch_id)
|
assert_subject_allowed(db, user, batch.subject_name if batch else None)
|
return {"attempt_id": str(attempt.id)}
|
|
|
@router.get("/attempts/{attempt_id}", response_model=AttemptDetail)
|
def get_attempt(
|
attempt_id: str,
|
db: Session = Depends(get_db),
|
user: User = Depends(get_current_user),
|
):
|
try:
|
attempt_pk = int(attempt_id)
|
except ValueError:
|
raise HTTPException(404, detail="记录不存在")
|
attempt = db.get(ExaminationExamExaminee, attempt_pk)
|
if not attempt:
|
raise HTTPException(404, detail="记录不存在")
|
examinee = db.get(ExaminationExaminee, attempt.examinee_id)
|
batch = db.get(ImportBatch, attempt.import_batch_id)
|
assert_subject_allowed(db, user, batch.subject_name if batch else None)
|
|
settings = get_or_create_settings(db)
|
hide_privacy = user.role == "grader" and not settings.grader_can_see_privacy
|
|
pmap = db.scalar(
|
select(ExamPaperMap).where(ExamPaperMap.exam_id == attempt.exam_id)
|
)
|
paper_id = pmap.paper_source_id if pmap else (batch.paper_source_id if batch else None)
|
paper = db.get(ExaminationPaperSource, paper_id) if paper_id else None
|
|
items = list(
|
db.scalars(
|
select(ExaminationPaperItem)
|
.where(ExaminationPaperItem.paper_source_id == paper_id)
|
.order_by(ExaminationPaperItem.part_ref, ExaminationPaperItem.item_order)
|
)
|
) if paper else []
|
|
answer_items = {
|
ai.external_item_id: ai
|
for ai in db.scalars(
|
select(ExaminationExamineeExamAnswerItem).where(
|
ExaminationExamineeExamAnswerItem.exam_examinee_id == attempt_pk
|
)
|
)
|
}
|
|
groups: dict[str, list[QuestionDetail]] = defaultdict(list)
|
part_order: list[str] = []
|
for it in items:
|
pn = it.part_name or "题目"
|
if pn not in part_order:
|
part_order.append(pn)
|
ai = answer_items.get(it.external_item_id)
|
groups[pn].append(
|
QuestionDetail(
|
external_item_id=it.external_item_id,
|
part_name=it.part_name,
|
item_type=it.item_type,
|
item_order=it.item_order,
|
title_html=it.title_html,
|
detail_json=it.detail_json,
|
max_score=it.max_score,
|
reference_answer=it.reference_answer,
|
user_answer=ai.answer_raw if ai else None,
|
is_correct=ai.is_correct if ai else False,
|
score=ai.score if ai else 0,
|
)
|
)
|
|
student_no = ensure_student_no(db, attempt)
|
|
return AttemptDetail(
|
attempt_id=str(attempt.id),
|
student_no=student_no,
|
examinee_name=None if hide_privacy else (examinee.nickname if examinee else None),
|
id_card=None if hide_privacy else (examinee.id_card if examinee else None),
|
score=attempt.score,
|
paper_total_score=paper.total_score if paper else None,
|
paper_name=paper.paper_name if paper else None,
|
subject_name=batch.subject_name if batch else None,
|
room_id=attempt.room_id,
|
seat_number=attempt.seat_number,
|
start_time=attempt.start_time,
|
end_time=attempt.end_time,
|
parts=[PartGroup(part_name=p, questions=groups[p]) for p in part_order],
|
)
|
