Initial commit: secondary school grade archive system.

Add FastAPI/React app with Docker deployment, Ubuntu one-click install, and docs for junior/senior high score tracking and mistake bank. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-28 11:18:58 +08:00
commit e329d3398a
76 changed files with 8506 additions and 0 deletions
@@ -0,0 +1,67 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from jose import JWTError, jwt
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.core.database import get_db
+from app.core.deps import get_current_user
+from app.core.security import (
+    create_access_token,
+    create_refresh_token,
+    get_password_hash,
+    verify_password,
+)
+from app.models.user import User
+from app.schemas import RefreshRequest, TokenResponse, UserLogin, UserOut, UserRegister
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+
+
+@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
+def register(data: UserRegister, db: Session = Depends(get_db)):
+    if db.query(User).filter(User.username == data.username).first():
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已存在")
+    user = User(username=data.username, password_hash=get_password_hash(data.password))
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+@router.post("/login", response_model=TokenResponse)
+def login(data: UserLogin, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.username == data.username).first()
+    if user is None or not verify_password(data.password, user.password_hash):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
+    return TokenResponse(
+        access_token=create_access_token(str(user.id)),
+        refresh_token=create_refresh_token(str(user.id)),
+    )
+
+
+@router.post("/refresh", response_model=TokenResponse)
+def refresh(data: RefreshRequest, db: Session = Depends(get_db)):
+    try:
+        payload = jwt.decode(data.refresh_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        user_id = payload.get("sub")
+        token_type = payload.get("type")
+        if user_id is None or token_type != "refresh":
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效刷新令牌")
+    except JWTError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效刷新令牌") from exc
+
+    user = db.get(User, uuid.UUID(user_id))
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在")
+
+    return TokenResponse(
+        access_token=create_access_token(str(user.id)),
+        refresh_token=create_refresh_token(str(user.id)),
+    )
+
+
+@router.get("/me", response_model=UserOut)
+def me(current_user: User = Depends(get_current_user)):
+    return current_user
@@ -0,0 +1,182 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session, joinedload
+
+from app.core.database import get_db
+from app.core.deps import get_current_user
+from app.models.user import ExamRecord, SubjectScore, User
+from app.schemas import ExamCreate, ExamOut, ExamUpdate, ScoreOut, TrendResponse
+from app.services.score_trend import build_trend
+from app.services.student_access import get_student_for_user
+
+router = APIRouter(tags=["exams"])
+
+
+def _score_to_out(score: SubjectScore) -> ScoreOut:
+    return ScoreOut(
+        id=score.id,
+        subject_id=score.subject_id,
+        subject_name=score.subject.name if score.subject else None,
+        total_score=float(score.total_score),
+        obtained_score=float(score.obtained_score),
+        ratio=float(score.ratio),
+    )
+
+
+def _exam_to_out(exam: ExamRecord) -> ExamOut:
+    return ExamOut(
+        id=exam.id,
+        exam_type=exam.exam_type,
+        exam_date=exam.exam_date,
+        title=exam.title,
+        created_at=exam.created_at,
+        scores=[_score_to_out(s) for s in exam.scores],
+    )
+
+
+def _apply_scores(db: Session, exam: ExamRecord, scores_data):
+    exam.scores.clear()
+    for item in scores_data:
+        ratio = round(item.obtained_score / item.total_score, 4)
+        exam.scores.append(
+            SubjectScore(
+                subject_id=item.subject_id,
+                total_score=item.total_score,
+                obtained_score=item.obtained_score,
+                ratio=ratio,
+            )
+        )
+
+
+@router.get("/students/{student_id}/exams", response_model=list[ExamOut])
+def list_exams(
+    student_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    get_student_for_user(db, student_id, current_user.id)
+    exams = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .filter(ExamRecord.student_id == student_id)
+        .order_by(ExamRecord.exam_date.desc(), ExamRecord.created_at.desc())
+        .all()
+    )
+    return [_exam_to_out(e) for e in exams]
+
+
+@router.post("/students/{student_id}/exams", response_model=ExamOut, status_code=status.HTTP_201_CREATED)
+def create_exam(
+    student_id: uuid.UUID,
+    data: ExamCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    get_student_for_user(db, student_id, current_user.id)
+    exam = ExamRecord(
+        student_id=student_id,
+        exam_type=data.exam_type,
+        exam_date=data.exam_date,
+        title=data.title,
+    )
+    db.add(exam)
+    db.flush()
+    if data.scores:
+        _apply_scores(db, exam, data.scores)
+    db.commit()
+    db.refresh(exam)
+    exam = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .filter(ExamRecord.id == exam.id)
+        .first()
+    )
+    return _exam_to_out(exam)
+
+
+@router.get("/exams/{exam_id}", response_model=ExamOut)
+def get_exam(
+    exam_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    exam = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .join(ExamRecord.student)
+        .filter(ExamRecord.id == exam_id)
+        .first()
+    )
+    if exam is None or exam.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="考试记录不存在")
+    return _exam_to_out(exam)
+
+
+@router.patch("/exams/{exam_id}", response_model=ExamOut)
+def update_exam(
+    exam_id: uuid.UUID,
+    data: ExamUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    exam = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .join(ExamRecord.student)
+        .filter(ExamRecord.id == exam_id)
+        .first()
+    )
+    if exam is None or exam.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="考试记录不存在")
+
+    if data.exam_type is not None:
+        exam.exam_type = data.exam_type
+    if data.exam_date is not None:
+        exam.exam_date = data.exam_date
+    if data.title is not None:
+        exam.title = data.title
+    if data.scores is not None:
+        _apply_scores(db, exam, data.scores)
+
+    db.commit()
+    db.refresh(exam)
+    exam = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .filter(ExamRecord.id == exam.id)
+        .first()
+    )
+    return _exam_to_out(exam)
+
+
+@router.delete("/exams/{exam_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_exam(
+    exam_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    exam = (
+        db.query(ExamRecord)
+        .join(ExamRecord.student)
+        .filter(ExamRecord.id == exam_id)
+        .first()
+    )
+    if exam is None or exam.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="考试记录不存在")
+    db.delete(exam)
+    db.commit()
+
+
+@router.get("/students/{student_id}/scores/trend", response_model=TrendResponse)
+def get_score_trend(
+    student_id: uuid.UUID,
+    subject_id: int = Query(...),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    get_student_for_user(db, student_id, current_user.id)
+    try:
+        return build_trend(db, student_id, subject_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
@@ -0,0 +1,54 @@
+import csv
+import io
+import uuid
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import StreamingResponse
+from sqlalchemy.orm import Session, joinedload
+
+from app.core.database import get_db
+from app.core.deps import get_current_user
+from app.models.user import ExamRecord, SubjectScore, User
+from app.services.student_access import get_student_for_user
+
+router = APIRouter(tags=["export"])
+
+
+@router.get("/students/{student_id}/scores/export")
+def export_scores_csv(
+    student_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    student = get_student_for_user(db, student_id, current_user.id)
+    exams = (
+        db.query(ExamRecord)
+        .options(joinedload(ExamRecord.scores).joinedload(SubjectScore.subject))
+        .filter(ExamRecord.student_id == student_id)
+        .order_by(ExamRecord.exam_date.asc())
+        .all()
+    )
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["考试日期", "考试类型", "标题", "科目", "总分", "得分", "占比"])
+    type_map = {"weekly": "周考", "monthly": "月考", "final": "期末"}
+    for exam in exams:
+        for score in exam.scores:
+            writer.writerow([
+                exam.exam_date.isoformat(),
+                type_map.get(exam.exam_type.value, exam.exam_type.value),
+                exam.title or "",
+                score.subject.name if score.subject else "",
+                float(score.total_score),
+                float(score.obtained_score),
+                f"{float(score.ratio) * 100:.2f}%",
+            ])
+
+    output.seek(0)
+    filename = f"{student.name}_scores.csv"
+    return StreamingResponse(
+        iter([output.getvalue().encode("utf-8-sig")]),
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
@@ -0,0 +1,73 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.deps import get_current_user
+from app.models.user import Student, User
+from app.schemas import StudentCreate, StudentOut, StudentUpdate
+from app.services.student_access import get_student_for_user
+
+router = APIRouter(prefix="/students", tags=["students"])
+
+
+@router.get("", response_model=list[StudentOut])
+def list_students(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    return (
+        db.query(Student)
+        .filter(Student.user_id == current_user.id)
+        .order_by(Student.created_at.desc())
+        .all()
+    )
+
+
+@router.post("", response_model=StudentOut, status_code=status.HTTP_201_CREATED)
+def create_student(
+    data: StudentCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    student = Student(user_id=current_user.id, **data.model_dump())
+    db.add(student)
+    db.commit()
+    db.refresh(student)
+    return student
+
+
+@router.get("/{student_id}", response_model=StudentOut)
+def get_student(
+    student_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    return get_student_for_user(db, student_id, current_user.id)
+
+
+@router.patch("/{student_id}", response_model=StudentOut)
+def update_student(
+    student_id: uuid.UUID,
+    data: StudentUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    student = get_student_for_user(db, student_id, current_user.id)
+    for key, value in data.model_dump(exclude_unset=True).items():
+        setattr(student, key, value)
+    db.commit()
+    db.refresh(student)
+    return student
+
+
+@router.delete("/{student_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_student(
+    student_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    student = get_student_for_user(db, student_id, current_user.id)
+    db.delete(student)
+    db.commit()
@@ -0,0 +1,17 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.deps import get_current_user
+from app.models.user import Subject, User
+from app.schemas import SubjectOut
+
+router = APIRouter(prefix="/subjects", tags=["subjects"])
+
+
+@router.get("", response_model=list[SubjectOut])
+def list_subjects(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    return db.query(Subject).order_by(Subject.id).all()
@@ -0,0 +1,313 @@
+import uuid
+from pathlib import Path
+
+from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, HTTPException, Query, UploadFile, status
+from fastapi.responses import FileResponse
+from sqlalchemy.orm import Session, joinedload
+
+from app.core.config import settings
+from app.core.database import SessionLocal, get_db
+from app.core.deps import get_current_user
+from app.models.user import Subject, User, WrongQuestion, WrongQuestionStatus
+from app.schemas import WrongQuestionOut, WrongQuestionUpdate
+from app.services import ocr as ocr_service
+from app.services import ollama as ollama_service
+from app.services.student_access import get_student_for_user
+
+router = APIRouter(tags=["wrong_questions"])
+
+
+def _wq_to_out(wq: WrongQuestion) -> WrongQuestionOut:
+    return WrongQuestionOut(
+        id=wq.id,
+        student_id=wq.student_id,
+        subject_id=wq.subject_id,
+        subject_name=wq.subject.name if wq.subject else None,
+        image_path=wq.image_path,
+        ocr_raw_text=wq.ocr_raw_text,
+        question_text=wq.question_text,
+        solution_text=wq.solution_text,
+        status=wq.status,
+        created_at=wq.created_at,
+    )
+
+
+def _process_wrong_question(question_id: uuid.UUID):
+    db = SessionLocal()
+    try:
+        wq = (
+            db.query(WrongQuestion)
+            .options(joinedload(WrongQuestion.subject), joinedload(WrongQuestion.student))
+            .filter(WrongQuestion.id == question_id)
+            .first()
+        )
+        if wq is None:
+            return
+
+        image_full = Path(settings.UPLOAD_DIR) / wq.image_path
+        try:
+            ocr_text = ocr_service.run_ocr(str(image_full))
+            wq.ocr_raw_text = ocr_text or None
+            wq.status = WrongQuestionStatus.ocr_done if ocr_text else WrongQuestionStatus.failed
+            db.commit()
+        except Exception:
+            wq.status = WrongQuestionStatus.failed
+            db.commit()
+            return
+
+        if not ocr_text:
+            return
+
+        subject_name = wq.subject.name if wq.subject else "综合"
+        school_level = wq.student.school_level if wq.student else None
+        import asyncio
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            question_text = loop.run_until_complete(
+                ollama_service.format_question(subject_name, ocr_text, school_level)
+            )
+            solution_text = loop.run_until_complete(
+                ollama_service.generate_solution(subject_name, question_text, school_level)
+            )
+            wq.question_text = question_text
+            wq.solution_text = solution_text
+            wq.status = WrongQuestionStatus.solved
+            db.commit()
+        except Exception:
+            wq.status = WrongQuestionStatus.ocr_done
+            db.commit()
+        finally:
+            loop.close()
+    finally:
+        db.close()
+
+
+@router.get("/students/{student_id}/wrong-questions", response_model=list[WrongQuestionOut])
+def list_wrong_questions(
+    student_id: uuid.UUID,
+    subject_id: int | None = Query(None),
+    q: str | None = Query(None),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    get_student_for_user(db, student_id, current_user.id)
+    query = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject))
+        .filter(WrongQuestion.student_id == student_id)
+    )
+    if subject_id is not None:
+        query = query.filter(WrongQuestion.subject_id == subject_id)
+    if q:
+        pattern = f"%{q}%"
+        query = query.filter(
+            (WrongQuestion.question_text.ilike(pattern))
+            | (WrongQuestion.solution_text.ilike(pattern))
+            | (WrongQuestion.ocr_raw_text.ilike(pattern))
+        )
+    items = query.order_by(WrongQuestion.created_at.desc()).all()
+    return [_wq_to_out(w) for w in items]
+
+
+@router.post(
+    "/students/{student_id}/wrong-questions",
+    response_model=WrongQuestionOut,
+    status_code=status.HTTP_201_CREATED,
+)
+async def upload_wrong_question(
+    student_id: uuid.UUID,
+    background_tasks: BackgroundTasks,
+    subject_id: int = Form(...),
+    file: UploadFile = File(...),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    get_student_for_user(db, student_id, current_user.id)
+    subject = db.get(Subject, subject_id)
+    if subject is None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="科目不存在")
+
+    content = await file.read()
+    if len(content) > settings.MAX_UPLOAD_SIZE:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="文件超过10MB限制")
+
+    wq = WrongQuestion(
+        student_id=student_id,
+        subject_id=subject_id,
+        image_path="",
+        status=WrongQuestionStatus.pending,
+    )
+    db.add(wq)
+    db.flush()
+
+    rel_path = ocr_service.save_upload_file(
+        str(current_user.id), str(wq.id), file.filename or "image.jpg", content
+    )
+    wq.image_path = rel_path
+    db.commit()
+    db.refresh(wq)
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject))
+        .filter(WrongQuestion.id == wq.id)
+        .first()
+    )
+
+    background_tasks.add_task(_process_wrong_question, wq.id)
+    return _wq_to_out(wq)
+
+
+@router.get("/wrong-questions/{question_id}", response_model=WrongQuestionOut)
+def get_wrong_question(
+    question_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject), joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+    return _wq_to_out(wq)
+
+
+@router.patch("/wrong-questions/{question_id}", response_model=WrongQuestionOut)
+def update_wrong_question(
+    question_id: uuid.UUID,
+    data: WrongQuestionUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject), joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+
+    if data.subject_id is not None:
+        if db.get(Subject, data.subject_id) is None:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="科目不存在")
+        wq.subject_id = data.subject_id
+    if data.question_text is not None:
+        wq.question_text = data.question_text
+    if data.solution_text is not None:
+        wq.solution_text = data.solution_text
+
+    db.commit()
+    db.refresh(wq)
+    return _wq_to_out(wq)
+
+
+@router.delete("/wrong-questions/{question_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_wrong_question(
+    question_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+
+    image_full = Path(settings.UPLOAD_DIR) / wq.image_path
+    db.delete(wq)
+    db.commit()
+    if image_full.exists():
+        image_full.unlink()
+
+
+@router.post("/wrong-questions/{question_id}/retry-ocr", response_model=WrongQuestionOut)
+def retry_ocr(
+    question_id: uuid.UUID,
+    background_tasks: BackgroundTasks,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject), joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+
+    wq.status = WrongQuestionStatus.pending
+    db.commit()
+    background_tasks.add_task(_process_wrong_question, wq.id)
+    return _wq_to_out(wq)
+
+
+@router.post("/wrong-questions/{question_id}/regenerate-solution", response_model=WrongQuestionOut)
+async def regenerate_solution(
+    question_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.subject), joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+
+    if not wq.question_text and not wq.ocr_raw_text:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="缺少题目内容")
+
+    subject_name = wq.subject.name if wq.subject else "综合"
+    school_level = wq.student.school_level if wq.student else None
+    question_text = wq.question_text or wq.ocr_raw_text or ""
+
+    try:
+        if not wq.question_text and wq.ocr_raw_text:
+            wq.question_text = await ollama_service.format_question(
+                subject_name, wq.ocr_raw_text, school_level
+            )
+            question_text = wq.question_text
+        wq.solution_text = await ollama_service.generate_solution(
+            subject_name, question_text, school_level
+        )
+        wq.status = WrongQuestionStatus.solved
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Ollama 调用失败: {exc}"
+        ) from exc
+
+    db.commit()
+    db.refresh(wq)
+    return _wq_to_out(wq)
+
+
+@router.get("/wrong-questions/{question_id}/image")
+def get_wrong_question_image(
+    question_id: uuid.UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    wq = (
+        db.query(WrongQuestion)
+        .options(joinedload(WrongQuestion.student))
+        .filter(WrongQuestion.id == question_id)
+        .first()
+    )
+    if wq is None or wq.student.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="错题不存在")
+
+    image_full = Path(settings.UPLOAD_DIR) / wq.image_path
+    if not image_full.exists():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在")
+    return FileResponse(image_full)