import logging 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.services import site_settings from app.services.question_figures import figures_for_questions from app.services.study_plan_context import plan_context_for_quizzes from app.utils.upload_access import validate_image_attachments from app.utils.quiz_questions import validate_option_explanations from app.database import get_db from app.models.quiz import PLAN_ORIGIN, Quiz from app.models.quiz_category import QuizCategory 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.models.question_category import QuestionCategory from app.services.quiz_builder import category_breadcrumbs from app.schemas.quiz import QuizCreate, QuizUpdate, QuizResponse, QuizDetail, QuizLearningDetail, QuizReview from app.services import quiz_service from app.utils.quiz_access import can_access_quiz, general_quiz_visibility, require_quiz_access, set_quiz_shared, quiz_shareable_predicate 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() logger = logging.getLogger(__name__) @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", "") # Extraction lands in a review batch now rather than a quiz. quiz_id # is still read for jobs that ran before that change and whose keys are # still in Redis. batch_id = r.get(f"extraction:batch_id:{jid}") quiz_id = r.get(f"extraction:quiz_id:{jid}") title = r.get(f"extraction:job_title:{jid}") or "Extraction" jobs.append({ "job_id": jid, "title": title, "status": status, "steps_count": len(steps), "last_step": last_step[:80], "batch_id": int(batch_id) if batch_id else None, "quiz_id": int(quiz_id) if quiz_id else None, }) return jobs @router.delete("/jobs/{job_id}", status_code=204) def forget_job(job_id: str, current_user: User = Depends(get_current_user)): """Take one job off this person's list. A job is a record of work, not the work itself: removing it stops nothing that is running and deletes nothing it produced. It exists because the list is a list of what happened, and one somebody has read and dealt with is clutter in front of the ones they have not. Only from your own list — the id alone is not authority to touch anybody else's, and the keys behind it are shared. """ import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) key = f"extraction:user_jobs:{current_user.id}" if not r.lrem(key, 0, job_id): raise HTTPException(404, "That job is not in your list") # The steps and status go with it. They are keyed by job id and nothing # else points at them once the list entry is gone, so leaving them would # simply be rubbish with an hour to live. for suffix in ("status", "steps", "error", "job_title", "batch_id", "quiz_id", "article"): r.delete(f"extraction:{suffix}:{job_id}") @router.get("/job/{job_id}") def get_extraction_job(job_id: str, current_user: User = Depends(require_moderator)): """Poll extraction job progress. Returns the steps, the status, and — when it finishes — the review batch the drafts landed in.""" 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": batch_id = r.get(f"extraction:batch_id:{job_id}") result["batch_id"] = int(batch_id) if batch_id else None # Jobs that ran before extraction was staged still point at a quiz. quiz_id = r.get(f"extraction:quiz_id:{job_id}") result["quiz_id"] = int(quiz_id) if quiz_id else None 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 not can_access_quiz(db, quiz, current_user): 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"): title_query = db.query(Quiz).filter(Quiz.title.ilike(f"%{phrase}%")) title_query = title_query.filter(general_quiz_visibility(current_user)) for quiz in title_query.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, q.explanation_image_path, 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, "explanation_image_path": row.explanation_image_path, "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, "explanation_image_path": question.explanation_image_path, "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("/sessions") def list_quiz_sessions( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """One management row per accessible quiz: attempt state, progress and last score. Powers the session-management view (start / resume / review + analysis, repeat, rename, delete) without the client fanning out per-quiz requests. """ quizzes = ( db.query(Quiz) .filter(Quiz.deleted_at.is_(None)) .filter(general_quiz_visibility(current_user)) .order_by(Quiz.created_at.desc()) .all() ) if not quizzes: return [] quiz_ids = [q.id for q in quizzes] attempts = ( db.query(QuizAttempt) .filter(QuizAttempt.user_id == current_user.id, QuizAttempt.quiz_id.in_(quiz_ids)) .order_by(QuizAttempt.started_at) .all() ) active: dict[int, QuizAttempt] = {} finished: dict[int, list[QuizAttempt]] = {} for attempt in attempts: if attempt.completed_at is None: active[attempt.quiz_id] = attempt # newest wins (ordered by started_at) elif not attempt.expired: finished.setdefault(attempt.quiz_id, []).append(attempt) # Answered counts for live attempts come from the Redis progress blob; a # Redis outage degrades to "0 answered" rather than failing the whole page. answered: dict[int, int] = {} if active: try: 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) live_attempts = list(active.values()) keys = [f"quiz_progress:{current_user.id}:{a.id}" for a in live_attempts] for attempt, raw in zip(live_attempts, r.mget(keys)): if not raw: continue saved = _json.loads(raw) # Counted, not settled. Listing sessions used to hand in any # exam whose clock had run out, so simply opening this page # could mark a paper — the learner's next sight of a block they # had walked away from was a score. An out-of-time block stays # open until it is opened, where Time's Up is shown and closing # it hands the paper in. answered[attempt.id] = len(saved.get("answers", {}) or {}) except Exception: logger.warning("Redis unavailable for session progress", exc_info=True) categories = {c.id: c.name for c in db.query(QuizCategory).all()} plans = plan_context_for_quizzes(db, current_user.id, quiz_ids) rows = [] for quiz in quizzes: live = active.get(quiz.id) done = finished.get(quiz.id, []) # Material belonging to a study plan is not a session of its own. It is # listed, and started, on the plan — showing it here as well put the # same twelve titles under two headings with no way to tell them apart. # Once the learner has actually sat it, it is history and belongs here. if quiz.origin == PLAN_ORIGIN and not live and not done: continue last = done[-1] if done else None per_attempt = quiz.questions_per_attempt or quiz.questions_count or 0 def pct(attempt): return round(attempt.score / attempt.total_questions * 100) if attempt.total_questions else 0 last_activity = ( live.started_at if live else last.completed_at if last else quiz.created_at ) rows.append({ "quiz_id": quiz.id, "title": quiz.title, "mode": quiz.mode, "origin": quiz.origin, "questions_count": quiz.questions_count, "questions_per_attempt": per_attempt, "time_limit_minutes": quiz.time_limit_minutes, "category_id": quiz.category_id, "category_name": categories.get(quiz.category_id), "is_published": quiz.is_published, "is_shared": quiz.is_shared, "is_owner": quiz.user_id == current_user.id, "created_at": quiz.created_at.isoformat() if quiz.created_at else None, "state": "in_progress" if live else "completed" if last else "not_started", "active_attempt_id": live.id if live else None, "answered": answered.get(live.id, 0) if live else (last.total_questions if last else 0), "total": (live.total_questions or per_attempt) if live else per_attempt, "attempts_count": len(done), "last_attempt_id": last.id if last else None, "last_percentage": pct(last) if last else None, "last_score": last.score if last else None, "last_total": last.total_questions if last else None, "last_completed_at": last.completed_at.isoformat() if last else None, "best_percentage": max((pct(a) for a in done), default=None), "last_activity": last_activity.isoformat() if last_activity else None, "plan": plans.get(quiz.id), }) rows.sort(key=lambda r: (r["last_activity"] or ""), reverse=True) return rows @router.get("/share-policy") def share_policy(current_user: User = Depends(get_current_user)): """Whether this site lets a learner hand a session to somebody. Asked before the share dialog offers to make a link, so a switch an administrator has thrown reads as "not offered" rather than as a button that fails when pressed. """ return {"enabled": site_settings.get_flag("sharing_enabled")} @router.get("/", response_model=list[QuizResponse]) def list_quizzes( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """List accessible general quizzes (owned, published, or shared).""" q = db.query(Quiz).filter(Quiz.deleted_at.is_(None)) q = q.filter(general_quiz_visibility(current_user)) 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), ): """Return selected questions; only an authorized study/completed attempt can reveal answers.""" 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 not can_access_quiz(db, quiz, current_user): raise HTTPException(status_code=403, detail="This quiz is private") attempt = None if attempt_id is not None: attempt = db.query(QuizAttempt).filter_by(id=attempt_id, quiz_id=quiz.id, user_id=current_user.id).first() if not attempt: raise HTTPException(404, "Attempt not found for this quiz") reveal = attempt is not None and (attempt.mode == "study" or attempt.completed_at is not None) if study and not reveal: raise HTTPException(403, "Answers are hidden until the exam is submitted") if reveal: require_quiz_access(db, quiz, current_user, review=True) result = (QuizLearningDetail if reveal else QuizDetail).model_validate(quiz) result.attempt_mode = (attempt.mode or "exam") if attempt else None if attempt and attempt.selected_question_ids is not None: selected = set(attempt.selected_question_ids) result.questions = [q for q in result.questions if q.id in selected] result.questions_count = len(result.questions) categories = db.query(QuestionCategory).all() figures = figures_for_questions(db, [q.id for q in result.questions]) for question in result.questions: question.category_breadcrumbs = category_breadcrumbs(categories, question.question_category_id) # Stem figures always; explanation figures only where answers are # already revealed, so a figure cannot give away what the stem hides. question.figures = [figure for figure in figures.get(question.id, []) if reveal or figure["role"] == "stem"] 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") if not can_access_quiz(db, quiz, current_user): raise HTTPException(status_code=403, detail="This quiz is private") 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") require_quiz_access(db, quiz, current_user, review=True) 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, "image_path": q.image_path, "explanation_image_path": q.explanation_image_path, } 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", "image_path", "explanation_image_path", "option_explanations"} data = validate_image_attachments(db, current_user, data) if "option_explanations" in data: data["option_explanations"] = validate_option_explanations( data.get("options", question.options), data["option_explanations"]) 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, "image_path": question.image_path, "explanation_image_path": question.explanation_image_path, } @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 if not published: quiz.is_shared = 0 db.commit() return {"quiz_id": quiz_id, "is_published": quiz.is_published, "is_shared": quiz.is_shared} @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() @router.patch("/{quiz_id}/share") def share_quiz(quiz_id: int, shared: bool = Query(...), db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(404, "Quiz not found") return set_quiz_shared(db, quiz, current_user, shared) @router.post("/{quiz_id}/share-link") def create_share_link(quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): """Mark a general quiz shareable and return its public token.""" # An institution can switch sharing off for everyone; a link already issued # keeps working, but no new one is made. if not site_settings.get_flag("sharing_enabled"): raise HTTPException(403, "Sharing is turned off for this site") quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(404, "Quiz not found") if quiz.user_id != current_user.id and not current_user.is_moderator: raise HTTPException(403, "Only the owner or a moderator can share this quiz") if not db.query(Quiz.id).filter(Quiz.id == quiz.id, quiz_shareable_predicate()).first(): raise HTTPException(400, "This test contains private questions") if not quiz.share_token: import uuid quiz.share_token = uuid.uuid4().hex quiz.is_shared = 1 db.commit() return {"token": quiz.share_token, "enabled": True} @router.delete("/{quiz_id}/share-link", status_code=204) def revoke_share_link(quiz_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first() if not quiz: raise HTTPException(404, "Quiz not found") if quiz.user_id != current_user.id and not current_user.is_moderator: raise HTTPException(403, "Only the owner or a moderator can revoke sharing") quiz.share_token = None quiz.is_shared = 0 quiz.is_published = 0 db.commit()