pdf-quiz-generator/backend/app/routers/exams.py

"""Exams — the study objective a learner is working towards.

An exam sits above systems and disciplines: the same paediatric cardiology
question can count towards a paediatrics board and a step exam, so membership is
a link table rather than a column on the question.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import func
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.article import Article
from app.models.exam import (ArticleExamLink, BlueprintCategoryLink, Exam,
                             ExamBlueprint, QuestionExamLink)
from app.models.question import Question
from app.models.question_category import QuestionCategory, QuestionCategoryLink
from app.models.user import User
from app.services import exam_blueprint
from app.utils.auth import get_current_user, require_admin, require_moderator

router = APIRouter()


# The three readings an article can offer. An objective may show a subset.
ARTICLE_VIEWS = ("short", "long", "clinical")


class ExamWrite(BaseModel):
    name: str
    slug: str
    sort_order: int = 100
    is_active: int = 1
    family: str | None = None
    description: str | None = None
    article_views: list[str] | None = None


def views_for(exam: Exam | None) -> list[str]:
    """Which article views this objective shows. No objective means all of them."""
    if exam is None or not exam.article_views:
        return list(ARTICLE_VIEWS)
    chosen = [v for v in ARTICLE_VIEWS if v in exam.article_views]
    # An objective that shows nothing would leave every article blank, which is
    # a configuration mistake rather than a preference worth honouring.
    return chosen or list(ARTICLE_VIEWS)


@router.get("/")
def list_exams(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
    """Every selectable exam, with how many bank questions it covers."""
    counts = dict(
        db.query(QuestionExamLink.exam_id, func.count(QuestionExamLink.question_id))
        .group_by(QuestionExamLink.exam_id).all()
    )
    exams = db.query(Exam).filter(Exam.is_active == 1).order_by(Exam.sort_order, Exam.name).all()
    active = db.get(Exam, current_user.active_exam_id) if current_user.active_exam_id else None
    return {
        "active_exam_id": current_user.active_exam_id,
        "active_exam_name": active.name if active else None,
        # What the current objective actually changes, so the interface can say so
        # rather than leaving the learner to guess whether it did anything.
        "article_views": views_for(active),
        "exams": [{"id": e.id, "slug": e.slug, "name": e.name,
                   "family": e.family or "Other", "description": e.description,
                   "article_views": views_for(e),
                   "question_count": counts.get(e.id, 0)} for e in exams],
    }


class ActiveExam(BaseModel):
    exam_id: int | None = None


@router.put("/active")
def set_active_exam(
    data: ActiveExam,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Switch which exam the learner is studying for. Persisted on the user."""
    if data.exam_id is not None:
        exam = db.get(Exam, data.exam_id)
        if not exam or not exam.is_active:
            raise HTTPException(404, "Exam not found")
    current_user.active_exam_id = data.exam_id
    db.commit()
    return {"active_exam_id": current_user.active_exam_id}


def _clean_views(views: list[str] | None) -> list[str] | None:
    """Null means every view. An empty or nonsense list would blank every
    article, which is a mistake rather than a preference."""
    if views is None:
        return None
    kept = [v for v in ARTICLE_VIEWS if v in views]
    return kept or None


def _as_json(exam: Exam) -> dict:
    return {"id": exam.id, "slug": exam.slug, "name": exam.name,
            "family": exam.family, "description": exam.description,
            "sort_order": exam.sort_order, "is_active": exam.is_active,
            "article_views": views_for(exam)}


@router.post("/", status_code=201)
def create_exam(
    data: ExamWrite,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Create an objective, with everything it was given.

    This used to write name, slug, sort order and active, and silently drop
    family, description and article views — so a new objective landed in
    "Other" showing every view, whatever was asked for.
    """
    slug = data.slug.strip().lower()
    if not slug:
        raise HTTPException(422, "An objective needs a slug")
    if db.query(Exam.id).filter(Exam.slug == slug).first():
        raise HTTPException(409, "That slug is already in use")
    exam = Exam(slug=slug, name=data.name.strip(), sort_order=data.sort_order,
                is_active=data.is_active, family=(data.family or None),
                description=(data.description or None),
                article_views=_clean_views(data.article_views))
    db.add(exam)
    db.commit()
    db.refresh(exam)
    return _as_json(exam)


class ExamEdit(BaseModel):
    name: str | None = None
    sort_order: int | None = None
    is_active: int | None = None
    family: str | None = None
    description: str | None = None
    article_views: list[str] | None = None


@router.patch("/{exam_id}")
def update_exam(
    exam_id: int,
    data: ExamEdit,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    exam = db.get(Exam, exam_id)
    if not exam:
        raise HTTPException(404, "Exam not found")
    fields = data.model_dump(exclude_unset=True)
    if "article_views" in fields:
        fields["article_views"] = _clean_views(fields["article_views"])
    # The slug is the address other things hold; it is not edited here.
    for key, value in fields.items():
        setattr(exam, key, value)
    db.commit()
    db.refresh(exam)
    return _as_json(exam)


class AssignByCategory(BaseModel):
    category_ids: list[int] = Field(min_length=1)
    #: Whether to take everything beneath the named categories too. Almost
    #: always yes — naming "Cardiology" and getting only what was filed
    #: directly on it, not its conditions, is nobody's intent.
    include_descendants: bool = True
    questions: bool = True
    articles: bool = True


@router.post("/{exam_id}/assign")
def assign_by_category(
    exam_id: int,
    data: AssignByCategory,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Put everything under these categories into this objective.

    Populating a new objective one link row at a time is not a thing anyone
    would do for three thousand questions. Adding is idempotent — running it
    twice adds nothing the second time — so it is safe to widen a selection
    and run it again.
    """
    exam = db.get(Exam, exam_id)
    if not exam:
        raise HTTPException(404, "Exam not found")

    known = {row[0] for row in db.query(QuestionCategory.id).filter(
        QuestionCategory.id.in_(data.category_ids)).all()}
    missing = sorted(set(data.category_ids) - known)
    if missing:
        raise HTTPException(400, f"No such categor{'y' if len(missing) == 1 else 'ies'}: "
                                 f"{', '.join(str(m) for m in missing)}")

    scope = (exam_blueprint._descendants(db, list(known)) if data.include_descendants
             else set(known))

    added_questions = added_articles = 0

    if data.questions:
        linked = db.query(QuestionCategoryLink.question_id).filter(
            QuestionCategoryLink.category_id.in_(scope))
        ids = {row[0] for row in db.query(Question.id).filter(
            Question.deleted_at.is_(None),
            Question.question_category_id.in_(scope) | Question.id.in_(linked)).all()}
        have = {row[0] for row in db.query(QuestionExamLink.question_id).filter(
            QuestionExamLink.exam_id == exam_id).all()}
        fresh = ids - have
        db.add_all([QuestionExamLink(question_id=qid, exam_id=exam_id) for qid in fresh])
        added_questions = len(fresh)

    if data.articles:
        ids = {row[0] for row in db.query(Article.id).filter(
            Article.category_id.in_(scope)).all()}
        have = {row[0] for row in db.query(ArticleExamLink.article_id).filter(
            ArticleExamLink.exam_id == exam_id).all()}
        fresh = ids - have
        db.add_all([ArticleExamLink(article_id=aid, exam_id=exam_id) for aid in fresh])
        added_articles = len(fresh)

    db.commit()
    return {"exam_id": exam_id, "categories": len(scope),
            "questions_added": added_questions, "articles_added": added_articles}


@router.delete("/{exam_id}/assign")
def unassign_by_category(
    exam_id: int,
    data: AssignByCategory,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Take them back out. The questions and articles are untouched — only
    their membership of this objective goes."""
    if not db.get(Exam, exam_id):
        raise HTTPException(404, "Exam not found")
    scope = (exam_blueprint._descendants(db, data.category_ids) if data.include_descendants
             else set(data.category_ids))
    removed_questions = removed_articles = 0
    if data.questions:
        linked = db.query(QuestionCategoryLink.question_id).filter(
            QuestionCategoryLink.category_id.in_(scope))
        ids = [row[0] for row in db.query(Question.id).filter(
            Question.question_category_id.in_(scope) | Question.id.in_(linked)).all()]
        removed_questions = db.query(QuestionExamLink).filter(
            QuestionExamLink.exam_id == exam_id,
            QuestionExamLink.question_id.in_(ids)).delete(synchronize_session=False)
    if data.articles:
        ids = [row[0] for row in db.query(Article.id).filter(Article.category_id.in_(scope)).all()]
        removed_articles = db.query(ArticleExamLink).filter(
            ArticleExamLink.exam_id == exam_id,
            ArticleExamLink.article_id.in_(ids)).delete(synchronize_session=False)
    db.commit()
    return {"exam_id": exam_id, "questions_removed": removed_questions,
            "articles_removed": removed_articles}


@router.get("/{exam_id}/blueprint")
def read_blueprint(
    exam_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """The published shape of this exam, and what the bank can supply for it."""
    if not db.get(Exam, exam_id):
        raise HTTPException(404, "Exam not found")
    rows = exam_blueprint.coverage(db, exam_id)
    return {
        "exam_id": exam_id,
        "total_weight": exam_blueprint.total_weight(db, exam_id),
        "domains": rows,
        # Said plainly, because a domain weighted at 12% with nine questions
        # behind it will never carry a paper.
        "empty": [row["code"] for row in rows if row["questions"] == 0],
    }


class BlueprintCategories(BaseModel):
    category_ids: list[int]


@router.put("/{exam_id}/blueprint/{blueprint_id}/categories")
def set_blueprint_categories(
    exam_id: int,
    blueprint_id: int,
    data: BlueprintCategories,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Which of our categories feed one line of the outline.

    Replaces rather than adds: the mapping is a statement of what this domain
    is, and editing it by hand should not require remembering what was there.
    """
    line = db.query(ExamBlueprint).filter(
        ExamBlueprint.id == blueprint_id, ExamBlueprint.exam_id == exam_id).first()
    if not line:
        raise HTTPException(404, "No such blueprint line for this exam")
    known = {row[0] for row in db.query(QuestionCategory.id).filter(
        QuestionCategory.id.in_(data.category_ids)).all()}
    missing = sorted(set(data.category_ids) - known)
    if missing:
        raise HTTPException(400, f"No such category: {', '.join(str(m) for m in missing)}")
    db.query(BlueprintCategoryLink).filter(
        BlueprintCategoryLink.blueprint_id == blueprint_id).delete(synchronize_session=False)
    db.add_all([BlueprintCategoryLink(blueprint_id=blueprint_id, category_id=cid)
                for cid in known])
    db.commit()
    return {"blueprint_id": blueprint_id, "category_ids": sorted(known)}