import random from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import cast, String, or_, and_, func from sqlalchemy.orm import Session from app.database import get_db from app.models.quiz import Quiz from app.models.question import Question as QuestionModel from app.models.section import Section from app.models.attempt import QuizAttempt from app.models.user import User from app.schemas.quiz import QuizCreate, QuizUpdate, QuizResponse, QuizDetail, QuizLearningDetail, QuizReview from app.services import quiz_service from app.utils.auth import get_current_user, require_moderator from app.utils.quiz_questions import get_quiz_questions, question_in_quiz, remove_question_from_quiz router = APIRouter() @router.post("/") def create_quiz( quiz_data: QuizCreate, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Start async quiz extraction. Returns {job_id} immediately; poll /quizzes/job/{job_id} for progress.""" import uuid 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'") job_id = str(uuid.uuid4()) try: from app.tasks.quiz_tasks import extract_quiz import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) r.set(f"extraction:status:{job_id}", "pending", ex=3600) # Store job under user so any session can query it title = quiz_data.title r.lpush(f"extraction:user_jobs:{current_user.id}", job_id) r.expire(f"extraction:user_jobs:{current_user.id}", 86400) r.set(f"extraction:job_title:{job_id}", title, ex=3600) extract_quiz.delay( job_id=job_id, 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, model_id=quiz_data.model_id, question_category_id=quiz_data.question_category_id, extraction_mode=quiz_data.extraction_mode, ) except Exception: # Celery/Redis unavailable — fall back to synchronous extraction 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, model_id=quiz_data.model_id, question_category_id=quiz_data.question_category_id, ) return {"job_id": job_id, "status": "completed", "quiz_id": quiz.id} except (ValueError, RuntimeError) as e: raise HTTPException(status_code=400, detail=str(e)) return {"job_id": job_id, "status": "pending"} @router.get("/jobs") def list_user_jobs(current_user: User = Depends(get_current_user)): """Return all recent extraction jobs for the logged-in user (any browser/session).""" import json as _json import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) job_ids = r.lrange(f"extraction:user_jobs:{current_user.id}", 0, 19) jobs = [] for jid in job_ids: status = r.get(f"extraction:status:{jid}") if not status: continue steps = r.lrange(f"extraction:steps:{jid}", 0, -1) last_step = "" if steps: last_step = _json.loads(steps[-1]).get("message", "") quiz_id = r.get(f"extraction:quiz_id:{jid}") title = r.get(f"extraction:job_title:{jid}") or "Quiz" jobs.append({ "job_id": jid, "title": title, "status": status, "steps_count": len(steps), "last_step": last_step[:80], "quiz_id": int(quiz_id) if quiz_id else None, }) return jobs @router.get("/job/{job_id}") def get_extraction_job(job_id: str, current_user: User = Depends(require_moderator)): """Poll extraction job progress. Returns steps list, status, and quiz_id when done.""" import json as _json import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) status = r.get(f"extraction:status:{job_id}") or "unknown" raw_steps = r.lrange(f"extraction:steps:{job_id}", 0, -1) steps = [_json.loads(s) for s in raw_steps] result = {"job_id": job_id, "status": status, "steps": steps} if status == "completed": result["quiz_id"] = int(r.get(f"extraction:quiz_id:{job_id}") or 0) if status == "failed": result["error"] = r.get(f"extraction:error:{job_id}") or "Unknown error" return result @router.post("/job/{job_id}/cancel") def cancel_extraction_job(job_id: str, current_user: User = Depends(get_current_user)): """Cancel a running extraction job.""" import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) # Verify this job belongs to the user user_jobs = r.lrange(f"extraction:user_jobs:{current_user.id}", 0, 19) if job_id not in user_jobs: raise HTTPException(status_code=403, detail="Job not found") status = r.get(f"extraction:status:{job_id}") if status != "running": raise HTTPException(status_code=400, detail=f"Job is not running (status: {status})") r.set(f"extraction:status:{job_id}", "cancelled") return {"status": "cancelled"} @router.get("/search") def search_quizzes( q: str = Query(..., min_length=2, max_length=200), mode: str = Query("all"), # "title" | "questions" | "all" db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Hybrid semantic + keyword search across quiz titles and questions.""" from sqlalchemy import text as sa_text from app.services import embedding_service phrase = q.strip() if not phrase: return [] results = {} # quiz_id -> result dict def _ensure_quiz(quiz_id: int, match_type: str): if quiz_id not in results: quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first() if not quiz or quiz.course_id is not None: return False results[quiz_id] = { "quiz_id": quiz.id, "quiz_title": quiz.title, "questions_count": quiz.questions_count, "mode": quiz.mode, "time_limit_minutes": quiz.time_limit_minutes, "match_type": match_type, "matching_questions": [], } elif results[quiz_id]["match_type"] != match_type and match_type != "title": results[quiz_id]["match_type"] = "both" return True # ── Title search ───────────────────────────────────────────── if mode in ("title", "all"): for quiz in db.query(Quiz).filter(Quiz.title.ilike(f"%{phrase}%"), Quiz.course_id.is_(None)).limit(30).all(): _ensure_quiz(quiz.id, "title") # ── Semantic (vector) search ────────────────────────────────── seen_question_ids = set() if mode in ("questions", "all"): query_emb = embedding_service.generate_embedding(phrase) if query_emb: # Use f-string for the vector literal — safe because it's a list of floats emb_literal = "[" + ",".join(str(float(x)) for x in query_emb) + "]" rows = db.execute(sa_text(""" SELECT q.id, q.quiz_id, q.question_text, q.options, q.correct_answer, q.explanation, 1 - (q.embedding <=> CAST(:vec AS vector)) AS similarity FROM questions q WHERE q.embedding IS NOT NULL ORDER BY q.embedding <=> CAST(:vec AS vector) LIMIT 40 """), {"vec": emb_literal}).fetchall() for row in rows: similarity = float(row.similarity) if similarity < 0.30: continue quiz_id_val = row.quiz_id # DB column still named quiz_id if quiz_id_val and _ensure_quiz(quiz_id_val, "questions"): seen_question_ids.add(row.id) results[quiz_id_val]["matching_questions"].append({ "id": row.id, "question_text": row.question_text, "options": row.options, "correct_answer": row.correct_answer, "explanation": row.explanation, "similarity": round(similarity, 3), "match_source": "semantic", }) # ── Keyword (ILIKE) search ──────────────────────────────────── if mode in ("questions", "all"): q_filter = or_( QuestionModel.question_text.ilike(f"%{phrase}%"), cast(QuestionModel.options, String).ilike(f"%{phrase}%"), ) keyword_rows = ( db.query(QuestionModel) .filter(q_filter) .order_by(QuestionModel.source_quiz_id, QuestionModel.id) .limit(200) .all() ) for question in keyword_rows: src_qid = question.source_quiz_id if src_qid and _ensure_quiz(src_qid, "questions"): if question.id not in seen_question_ids: seen_question_ids.add(question.id) results[src_qid]["matching_questions"].append({ "id": question.id, "question_text": question.question_text, "options": question.options, "correct_answer": question.correct_answer, "explanation": question.explanation, "similarity": None, "match_source": "keyword", }) # Sort each quiz's questions: semantic first (by similarity desc), then keyword for r in results.values(): r["matching_questions"].sort( key=lambda x: (0 if x.get("match_source") == "semantic" else 1, -(x.get("similarity") or 0)) ) # Sort results: title matches first, then by number of semantic hits desc sorted_results = sorted( results.values(), key=lambda r: ( 0 if r["match_type"] == "title" else 1, -sum(1 for q in r["matching_questions"] if q.get("match_source") == "semantic"), ), ) return sorted_results @router.get("/", response_model=list[QuizResponse]) def list_quizzes( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """List quizzes. Moderators see all; regular users only see published.""" q = db.query(Quiz).filter(Quiz.deleted_at.is_(None), Quiz.course_id.is_(None)) if not current_user.is_moderator: q = q.filter(Quiz.is_published == 1) return q.order_by(Quiz.created_at.desc()).all() @router.get("/{quiz_id}") def get_quiz( quiz_id: int, study: bool = Query(False), attempt_id: int | None = Query(None), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get quiz for taking. study=true forces answers/explanations to be included. attempt_id filters to the question subset selected for that attempt (question pool).""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") if study or quiz.mode == "learning": result = QuizLearningDetail.model_validate(quiz) else: result = QuizDetail.model_validate(quiz) # Filter questions if attempt has a question pool selection if attempt_id: from app.models.attempt import QuizAttempt attempt = db.query(QuizAttempt).filter( QuizAttempt.id == attempt_id, QuizAttempt.user_id == current_user.id, ).first() if attempt and attempt.selected_question_ids: sel = set(attempt.selected_question_ids) result.questions = [q for q in result.questions if q.id in sel] result.questions_count = len(result.questions) return result @router.patch("/{quiz_id}", response_model=QuizResponse) def update_quiz( quiz_id: int, data: QuizUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Update quiz metadata (title, etc.) — owner or admin only.""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") if quiz.user_id != current_user.id and current_user.role != "admin": raise HTTPException(status_code=403, detail="Not authorized to edit this quiz") if data.title: quiz.title = data.title.strip() db.commit() db.refresh(quiz) return quiz @router.post("/{quiz_id}/shuffle", response_model=QuizDetail) def shuffle_quiz( quiz_id: int, shuffle_options: bool = Query(True, description="Also shuffle answer options within each question"), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Return quiz with question order and optionally option order shuffled. Does not modify DB.""" from app.models.question import Question as QuestionModel from app.schemas.quiz import QuestionResponse quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") questions = get_quiz_questions(db, quiz_id) shuffled_questions = questions.copy() random.shuffle(shuffled_questions) result = [] for q in shuffled_questions: options = list(q.options) if q.options else None if shuffle_options and options: random.shuffle(options) result.append(QuestionResponse( id=q.id, question_text=q.question_text, question_type=q.question_type, options=options, image_path=q.image_path, )) return { "id": quiz.id, "section_id": quiz.section_id, "user_id": quiz.user_id, "title": quiz.title, "questions_count": quiz.questions_count, "mode": quiz.mode, "time_limit_minutes": quiz.time_limit_minutes, "created_at": quiz.created_at, "questions": result, } @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.get("/{quiz_id}/questions") def get_quiz_questions_for_edit( quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Get all questions with answers for editing — moderator/admin only.""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") questions = get_quiz_questions(db, quiz_id) return [ { "id": q.id, "question_text": q.question_text, "question_type": q.question_type, "options": q.options, "correct_answer": q.correct_answer, "explanation": q.explanation, } for q in questions ] @router.patch("/{quiz_id}/questions/{question_id}") def update_question( quiz_id: int, question_id: int, data: dict, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Update a quiz question — change is shared: affects all quizzes using this question.""" if not question_in_quiz(db, quiz_id, question_id): raise HTTPException(status_code=404, detail="Question not found in this quiz") question = db.query(QuestionModel).filter(QuestionModel.id == question_id).first() if not question: raise HTTPException(status_code=404, detail="Question not found") allowed = {"question_text", "options", "correct_answer", "explanation", "question_type"} for key, value in data.items(): if key in allowed: setattr(question, key, value) # Validate correct_answer is one of the options if question.options and question.correct_answer not in question.options: raise HTTPException( status_code=400, detail=f"correct_answer must match one of the options exactly. Options: {question.options}" ) db.commit() db.refresh(question) return { "id": question.id, "question_text": question.question_text, "question_type": question.question_type, "options": question.options, "correct_answer": question.correct_answer, "explanation": question.explanation, } @router.delete("/{quiz_id}/questions/{question_id}", status_code=204) def delete_question( quiz_id: int, question_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Remove a question from this quiz. If it's shared, only removes the link (question stays in bank). If it's only in this quiz, deletes it from the bank too.""" if not question_in_quiz(db, quiz_id, question_id): raise HTTPException(status_code=404, detail="Question not found in this quiz") from app.models.quiz_question_link import QuizQuestionLink # Check if used by any OTHER quiz other_uses = db.query(QuizQuestionLink).filter( QuizQuestionLink.question_id == question_id, QuizQuestionLink.quiz_id != quiz_id, ).count() remove_question_from_quiz(db, quiz_id, question_id) quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first() if quiz and quiz.questions_count > 0: quiz.questions_count -= 1 # Only delete the question record if no other quiz references it if other_uses == 0: db.query(QuestionModel).filter(QuestionModel.id == question_id).delete() db.commit() @router.delete("/{quiz_id}", status_code=204) def delete_quiz( quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Soft-delete quiz — moves to trash. Restore via PATCH /{quiz_id}/restore.""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") quiz.deleted_at = datetime.utcnow() db.commit() @router.patch("/{quiz_id}/publish") def set_quiz_published( quiz_id: int, published: bool = True, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Hide (published=false) or show (published=true) a quiz for regular users.""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") quiz.is_published = 1 if published else 0 db.commit() return {"quiz_id": quiz_id, "is_published": quiz.is_published} @router.get("/trash", response_model=list[QuizResponse]) def list_trash( db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """List soft-deleted quizzes — admin/moderator only.""" return db.query(Quiz).filter(Quiz.deleted_at.isnot(None)).order_by(Quiz.deleted_at.desc()).all() @router.patch("/{quiz_id}/restore", response_model=QuizResponse) def restore_quiz( quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Restore a soft-deleted quiz from trash.""" quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.isnot(None)).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found in trash") quiz.deleted_at = None db.commit() db.refresh(quiz) return quiz @router.delete("/{quiz_id}/permanent", status_code=204) def permanently_delete_quiz( quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(require_moderator), ): """Permanently delete a quiz that's already in trash. Questions stay in bank.""" from app.models.quiz_question_link import QuizQuestionLink from datetime import datetime as dt quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first() if not quiz: raise HTTPException(status_code=404, detail="Quiz not found") if quiz.deleted_at is None: raise HTTPException(status_code=400, detail="Move to trash first before permanently deleting") # Detach exclusive questions to bank links = db.query(QuizQuestionLink).filter(QuizQuestionLink.quiz_id == quiz_id).all() for link in links: other = db.query(QuizQuestionLink).filter( QuizQuestionLink.question_id == link.question_id, QuizQuestionLink.quiz_id != quiz_id, ).count() if other == 0: db.query(QuestionModel).filter(QuestionModel.id == link.question_id).update( {QuestionModel.source_quiz_id: None}, synchronize_session=False ) db.delete(quiz) db.commit()