- FastAPI backend with JWT auth, roles (admin/moderator/user) - PDF upload (up to 500MB) with streaming, PyMuPDF text extraction - ChromaDB vectorization per page with metadata - LiteLLM AI question extraction from PDF (not generation) - Image extraction from PDF pages, graceful fallback - Quiz modes: timed (countdown timer) + learning (answers shown inline) - Page-by-page question navigation with dot navigator - TTS endpoint using LiteLLM (Google Vertex / OpenAI voices) - Admin dashboard: AI model management per task, user role management - Moderator role: upload PDFs, create sections, generate quizzes - Spaced repetition reminders via SMTP email (SM-2 intervals) - APScheduler daily reminder jobs - Celery + Redis for background PDF processing - React frontend with all pages - Docker Compose deployment (nginx + backend + celery + redis) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models.quiz import Quiz
|
|
from app.models.section import Section
|
|
from app.models.attempt import QuizAttempt
|
|
from app.models.user import User
|
|
from app.schemas.quiz import QuizCreate, QuizResponse, QuizDetail, QuizLearningDetail, QuizReview
|
|
from app.services import quiz_service
|
|
from app.utils.auth import get_current_user, require_moderator
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/", response_model=QuizResponse)
|
|
def create_quiz(
|
|
quiz_data: QuizCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Create quiz by extracting questions from PDF section. Moderator/Admin only."""
|
|
section = db.query(Section).filter(Section.id == quiz_data.section_id).first()
|
|
if not section:
|
|
raise HTTPException(status_code=404, detail="Section not found")
|
|
if not current_user.is_admin and section.document.user_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not your document")
|
|
|
|
if quiz_data.mode not in ("timed", "learning"):
|
|
raise HTTPException(status_code=400, detail="Mode must be 'timed' or 'learning'")
|
|
|
|
try:
|
|
quiz = quiz_service.create_quiz_from_section(
|
|
db=db,
|
|
user_id=current_user.id,
|
|
section_id=quiz_data.section_id,
|
|
title=quiz_data.title,
|
|
mode=quiz_data.mode,
|
|
time_limit_minutes=quiz_data.time_limit_minutes,
|
|
)
|
|
return quiz
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except RuntimeError as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/", response_model=list[QuizResponse])
|
|
def list_quizzes(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""List all available quizzes. Admins/mods see all, users see all published."""
|
|
if current_user.is_moderator:
|
|
quizzes = db.query(Quiz).order_by(Quiz.created_at.desc()).all()
|
|
else:
|
|
quizzes = db.query(Quiz).order_by(Quiz.created_at.desc()).all()
|
|
return quizzes
|
|
|
|
|
|
@router.get("/{quiz_id}")
|
|
def get_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get quiz for taking. Learning mode includes answers."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
if quiz.mode == "learning":
|
|
return QuizLearningDetail.model_validate(quiz)
|
|
return QuizDetail.model_validate(quiz)
|
|
|
|
|
|
@router.get("/{quiz_id}/review", response_model=QuizReview)
|
|
def review_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get quiz with answers — only if user has completed an attempt."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
has_attempt = db.query(QuizAttempt).filter(
|
|
QuizAttempt.quiz_id == quiz_id,
|
|
QuizAttempt.user_id == current_user.id,
|
|
QuizAttempt.completed_at.isnot(None),
|
|
).first()
|
|
if not has_attempt and not current_user.is_moderator:
|
|
raise HTTPException(status_code=403, detail="Complete an attempt first to review answers")
|
|
|
|
return quiz
|
|
|
|
|
|
@router.delete("/{quiz_id}", status_code=204)
|
|
def delete_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
db.delete(quiz)
|
|
db.commit()
|