4aa1352da7
**The badge is off the navbar.** It sat in the header of every page, for everybody, polling every thirty seconds — for a number that means something to the handful of people who start an extraction and nothing at all to a learner sitting a session. Extraction is workbench business and it lives there now: Settings → Tools → Jobs, and a link from the workbench itself, which is where one is started. To answer the question it raised: a job ages off the list after a day, and the steps behind it after an hour. Which is to say it disappears when Redis forgets it, on its own, with nothing to tell you it had. **So there is now a way to clear one.** "Forget" takes a job off your list, with a confirm beside it. It stops nothing that is running — the button is not offered for a running job — and deletes nothing the job produced; it clears a line somebody has read and dealt with so the ones they have not are not buried under it. Only from your own list: the id alone is not authority over anybody else's, and the keys behind it are shared. **And a way to read one.** The details panel is called "Log" now, because that is what it is — every step the job took, in order, and the reason it stopped if it stopped. That reason has been recorded all along and shown nowhere. Also, on the deck: the way out is a back link above the title like every other page rather than a small grey button in the bar of card controls, and the four buttons that are the whole interaction are full size. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
824 lines
34 KiB
Python
824 lines
34 KiB
Python
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()
|