Zhang Jing Xuan
0841b1a103
- 前端：Vue3 + TS + Element Plus，24 个页面路由（课件/组题/教案/动画/思维导图/作文批改/命题/课堂/资源/社区等） - 后端：FastAPI + SQLAlchemy + SQLite，17 个路由模块，AI 服务层含降级模板 - AI：glm-5.x 推理模型已禁用思维链，确保输出真实内容 - 修复：ai_service 两处请求体注入 enable_thinking/thinking disabled - 测试账号：13900999999 / test1234
199 lines
7.2 KiB
Python
199 lines
7.2 KiB
Python
import re
|
||
from io import BytesIO
|
||
from urllib.parse import quote
|
||
|
||
from docx import Document
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy.orm import Session
|
||
|
||
from database import get_db
|
||
from models.exam import Exam
|
||
from models.resource import Resource
|
||
from models.user import User
|
||
from schemas.exam import ExamCreate, ExamOut, ExamUpdate
|
||
from services.auth import get_current_user, get_optional_current_user
|
||
|
||
router = APIRouter(prefix="/api/exams", tags=["智能命题"])
|
||
|
||
|
||
def _safe_filename(name: str, suffix: str) -> str:
|
||
stem = re.sub(r"[\\/:*?\"<>|\r\n]+", "_", name).strip(" .") or "试卷"
|
||
return f"{stem[:80]}{suffix}"
|
||
|
||
|
||
def _as_list(value) -> list:
|
||
if isinstance(value, list):
|
||
return value
|
||
if value in (None, ""):
|
||
return []
|
||
return [value]
|
||
|
||
|
||
def _exam_or_404(exam_id: int, current_user: User | None, db: Session) -> Exam:
|
||
exam = db.query(Exam).filter(Exam.id == exam_id).first()
|
||
if not exam:
|
||
raise HTTPException(status_code=404, detail="试卷不存在")
|
||
if (not current_user or exam.user_id != current_user.id) and not has_public_resource(db, exam_id):
|
||
raise HTTPException(status_code=403, detail="无权访问该试卷")
|
||
return exam
|
||
|
||
|
||
def has_public_resource(db: Session, exam_id: int) -> bool:
|
||
return db.query(Resource).filter(
|
||
Resource.content_ref == f"exam:{exam_id}",
|
||
Resource.status == "active",
|
||
Resource.is_public == 1,
|
||
).first() is not None
|
||
|
||
|
||
@router.post("/", response_model=ExamOut, status_code=201)
|
||
def create_exam(data: ExamCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
||
exam = Exam(user_id=current_user.id, **data.model_dump())
|
||
db.add(exam)
|
||
db.commit()
|
||
db.refresh(exam)
|
||
return exam
|
||
|
||
|
||
@router.get("/", response_model=list[ExamOut])
|
||
def list_exams(
|
||
subject: str = "",
|
||
skip: int = Query(0, ge=0),
|
||
limit: int = Query(20, ge=1, le=100),
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
query = db.query(Exam).filter(Exam.user_id == current_user.id)
|
||
if subject:
|
||
query = query.filter(Exam.subject == subject)
|
||
return query.order_by(Exam.updated_at.desc()).offset(skip).limit(limit).all()
|
||
|
||
|
||
@router.get("/{exam_id}", response_model=ExamOut)
|
||
def get_exam(
|
||
exam_id: int,
|
||
current_user: User | None = Depends(get_optional_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
return _exam_or_404(exam_id, current_user, db)
|
||
|
||
|
||
@router.get("/{exam_id}/export-docx")
|
||
def export_exam_docx(
|
||
exam_id: int,
|
||
current_user: User | None = Depends(get_optional_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
exam = _exam_or_404(exam_id, current_user, db)
|
||
document = Document()
|
||
document.add_heading(exam.title or "智能试卷", level=1)
|
||
meta = [
|
||
exam.subject,
|
||
exam.grade,
|
||
f"{exam.duration or 90}分钟",
|
||
f"总分{exam.total_score or 100}分",
|
||
f"难度：{exam.difficulty or 'medium'}",
|
||
]
|
||
if exam.knowledge_points:
|
||
meta.append("知识点：" + "、".join(str(item) for item in exam.knowledge_points if str(item).strip()))
|
||
document.add_paragraph(" / ".join(str(part) for part in meta if part))
|
||
|
||
for index, question in enumerate(exam.questions or [], start=1):
|
||
if not isinstance(question, dict):
|
||
document.add_paragraph(f"{index}. {question}")
|
||
continue
|
||
q_type = question.get("type") or question.get("question_type") or "题目"
|
||
score = question.get("score")
|
||
heading = f"{index}. [{q_type}]"
|
||
if score:
|
||
heading += f"（{score}分）"
|
||
document.add_paragraph(heading)
|
||
document.add_paragraph(str(question.get("content") or question.get("question") or question.get("stem") or ""))
|
||
options = _as_list(question.get("options"))
|
||
for option_index, option in enumerate(options):
|
||
prefix = chr(65 + option_index) if option_index < 26 else str(option_index + 1)
|
||
document.add_paragraph(f"{prefix}. {option}")
|
||
|
||
answers = exam.answers or [question.get("answer") for question in exam.questions or [] if isinstance(question, dict)]
|
||
if answers:
|
||
document.add_page_break()
|
||
document.add_heading("参考答案与解析", level=1)
|
||
for index, question in enumerate(exam.questions or [], start=1):
|
||
answer = answers[index - 1] if index - 1 < len(answers) else ""
|
||
analysis = question.get("analysis") or question.get("explanation") if isinstance(question, dict) else ""
|
||
document.add_paragraph(f"{index}. 答案：{answer}")
|
||
if analysis:
|
||
document.add_paragraph(f"解析：{analysis}")
|
||
|
||
buffer = BytesIO()
|
||
document.save(buffer)
|
||
buffer.seek(0)
|
||
filename = _safe_filename(f"{exam.title or '智能试卷'}-试卷", ".docx")
|
||
return StreamingResponse(
|
||
buffer,
|
||
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||
headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"},
|
||
)
|
||
|
||
|
||
@router.put("/{exam_id}", response_model=ExamOut)
|
||
def update_exam(
|
||
exam_id: int,
|
||
data: ExamUpdate,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
exam = db.query(Exam).filter(Exam.id == exam_id, Exam.user_id == current_user.id).first()
|
||
if not exam:
|
||
raise HTTPException(status_code=404, detail="试卷不存在")
|
||
for key, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(exam, key, value)
|
||
db.commit()
|
||
db.refresh(exam)
|
||
return exam
|
||
|
||
|
||
@router.post("/{exam_id}/remix", response_model=ExamOut, status_code=201)
|
||
def remix_exam(
|
||
exam_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
source = db.query(Exam).filter(Exam.id == exam_id).first()
|
||
if not source:
|
||
raise HTTPException(status_code=404, detail="试卷不存在")
|
||
if source.user_id != current_user.id and not has_public_resource(db, exam_id):
|
||
raise HTTPException(status_code=403, detail="无权改编该试卷")
|
||
|
||
clone = Exam(
|
||
user_id=current_user.id,
|
||
title=f"{source.title}（改编）",
|
||
subject=source.subject or "",
|
||
grade=source.grade or "",
|
||
questions=list(source.questions or []),
|
||
answers=list(source.answers or []),
|
||
duration=source.duration or 90,
|
||
total_score=source.total_score or 100,
|
||
difficulty=source.difficulty or "medium",
|
||
knowledge_points=list(source.knowledge_points or []),
|
||
)
|
||
db.add(clone)
|
||
db.commit()
|
||
db.refresh(clone)
|
||
return clone
|
||
|
||
|
||
@router.delete("/{exam_id}", status_code=204)
|
||
def delete_exam(exam_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
||
exam = db.query(Exam).filter(Exam.id == exam_id, Exam.user_id == current_user.id).first()
|
||
if not exam:
|
||
raise HTTPException(status_code=404, detail="试卷不存在")
|
||
db.query(Resource).filter(
|
||
Resource.user_id == current_user.id,
|
||
Resource.content_ref == f"exam:{exam_id}",
|
||
Resource.status == "active",
|
||
).update({"status": "archived"}, synchronize_session=False)
|
||
db.delete(exam)
|
||
db.commit()
|