import re from io import BytesIO from urllib.parse import quote from docx import Document from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from typing import Optional from database import get_db from models.exercise import Exercise, ExerciseAttempt from models.resource import Resource from models.user import User from schemas.exercise import ExerciseCreate, ExerciseAIGenerate, ExerciseAttemptCreate, ExerciseAttemptOut, ExerciseOut, ExerciseUpdate from services.credits import credits_payload, spend_credits from services.limiter import limiter from services.auth import get_current_user, get_optional_current_user from services.ai_service import AIService router = APIRouter(prefix="/api/exercises", tags=["互动练习"]) ai_service = AIService() 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 _exercise_or_404(exercise_id: int, current_user: User | None, db: Session) -> Exercise: ex = db.query(Exercise).filter(Exercise.id == exercise_id).first() if not ex: raise HTTPException(status_code=404, detail="练习不存在") if (not current_user or ex.user_id != current_user.id) and not has_public_resource(db, exercise_id): raise HTTPException(status_code=403, detail="无权访问该练习") return ex def has_public_resource(db: Session, exercise_id: int) -> bool: return db.query(Resource).filter( Resource.content_ref == f"exercise:{exercise_id}", Resource.status == "active", Resource.is_public == 1, ).first() is not None @limiter.limit("10/minute") @router.post("/ai-generate", response_model=dict) async def ai_generate_exercise( request: Request, data: ExerciseAIGenerate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): account = spend_credits(db, current_user, "exercise_generate", f"生成互动练习：{data.prompt[:80]}") db.commit() result = await ai_service.generate_exercise( prompt=data.prompt, exercise_type=data.exercise_type, subject=data.subject, knowledge_points=data.knowledge_points, count=data.count, ) return {"success": True, "data": result, "credits": credits_payload(account, "exercise_generate")} @router.post("/", response_model=ExerciseOut, status_code=201) def create_exercise(data: ExerciseCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): ex = Exercise(user_id=current_user.id, **data.model_dump()) db.add(ex) db.commit() db.refresh(ex) return ex @router.get("/", response_model=list[ExerciseOut]) def list_exercises( exercise_type: Optional[str] = None, subject: Optional[str] = None, 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(Exercise).filter(Exercise.user_id == current_user.id) if exercise_type: query = query.filter(Exercise.exercise_type == exercise_type) if subject: query = query.filter(Exercise.subject == subject) return query.order_by(Exercise.updated_at.desc()).offset(skip).limit(limit).all() @router.get("/{exercise_id}", response_model=ExerciseOut) def get_exercise( exercise_id: int, current_user: User | None = Depends(get_optional_current_user), db: Session = Depends(get_db), ): return _exercise_or_404(exercise_id, current_user, db) @router.get("/{exercise_id}/export-docx") def export_exercise_docx( exercise_id: int, current_user: User | None = Depends(get_optional_current_user), db: Session = Depends(get_db), ): ex = _exercise_or_404(exercise_id, current_user, db) document = Document() document.add_heading(ex.title or "课堂练习", level=1) meta_parts = [ex.subject, ex.exercise_type, f"{len(ex.questions or [])}题"] if ex.knowledge_points: meta_parts.append("知识点：" + "、".join(str(item) for item in ex.knowledge_points if str(item).strip())) document.add_paragraph(" / ".join(str(part) for part in meta_parts if part)) for index, question in enumerate(ex.questions or [], start=1): if not isinstance(question, dict): document.add_paragraph(f"{index}. {question}") continue q_type = question.get("type") or ex.exercise_type or "练习" difficulty = question.get("difficulty") heading = f"{index}. [{q_type}]" if difficulty: heading += f"（{difficulty}）" document.add_paragraph(heading) document.add_paragraph(str(question.get("question") or question.get("content") 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}") answer = question.get("answer") if answer not in (None, ""): document.add_paragraph(f"答案：{answer}") explanation = question.get("explanation") or question.get("analysis") if explanation: document.add_paragraph(f"解析：{explanation}") buffer = BytesIO() document.save(buffer) buffer.seek(0) filename = _safe_filename(f"{ex.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("/{exercise_id}", response_model=ExerciseOut) def update_exercise( exercise_id: int, data: ExerciseUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): ex = db.query(Exercise).filter(Exercise.id == exercise_id, Exercise.user_id == current_user.id).first() if not ex: raise HTTPException(status_code=404, detail="练习不存在") for key, value in data.model_dump(exclude_unset=True).items(): setattr(ex, key, value) db.commit() db.refresh(ex) return ex @router.post("/{exercise_id}/remix", response_model=ExerciseOut, status_code=201) def remix_exercise( exercise_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): source = db.query(Exercise).filter(Exercise.id == exercise_id).first() if not source: raise HTTPException(status_code=404, detail="练习不存在") if source.user_id != current_user.id and not has_public_resource(db, exercise_id): raise HTTPException(status_code=403, detail="无权改编该练习") clone = Exercise( user_id=current_user.id, title=f"{source.title}（改编）", exercise_type=source.exercise_type or "choice", subject=source.subject or "", knowledge_points=list(source.knowledge_points or []), questions=list(source.questions or []), settings=dict(source.settings or {}), ) db.add(clone) db.commit() db.refresh(clone) return clone @router.delete("/{exercise_id}", status_code=204) def delete_exercise(exercise_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): ex = db.query(Exercise).filter(Exercise.id == exercise_id, Exercise.user_id == current_user.id).first() if not ex: raise HTTPException(status_code=404, detail="练习不存在") db.query(Resource).filter( Resource.user_id == current_user.id, Resource.content_ref == f"exercise:{exercise_id}", Resource.status == "active", ).update({"status": "archived"}, synchronize_session=False) db.delete(ex) db.commit() @router.post("/{exercise_id}/attempt", response_model=dict, status_code=201) def submit_attempt(exercise_id: int, data: ExerciseAttemptCreate, db: Session = Depends(get_db)): exercise = db.query(Exercise).filter(Exercise.id == exercise_id).first() if not exercise: raise HTTPException(status_code=404, detail="练习不存在") attempt = ExerciseAttempt( exercise_id=exercise_id, student_name=data.student_name, answers=data.answers, score=data.score, total=data.total, duration=data.duration, ) db.add(attempt) db.commit() db.refresh(attempt) return {"success": True, "id": attempt.id} @router.get("/{exercise_id}/attempts", response_model=list[ExerciseAttemptOut]) def list_attempts( exercise_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): exercise = db.query(Exercise).filter( Exercise.id == exercise_id, Exercise.user_id == current_user.id, ).first() if not exercise: raise HTTPException(status_code=404, detail="练习不存在") return ( db.query(ExerciseAttempt) .filter(ExerciseAttempt.exercise_id == exercise_id) .order_by(ExerciseAttempt.created_at.desc()) .limit(200) .all() )