pdf-quiz-generator/backend/app/services/prepared_session.py

"""The prepared session: one action, and a plain statement of why.

The manual builder asks a learner to choose topics, a count, a difficulty and a
state before it will give them anything. Most of those are decisions nobody
opening the app at seven in the morning has the information to make, and the
answers are already in their own record of answers. This turns that record into
a session.

Two things are produced together and must never disagree: the questions, and
the account of why those questions. The account is the product — an adaptive
session a learner cannot see the reasoning behind is one they abandon for a
manual session they can — so a plan is built, stated, and only then committed,
and the same `CandidateRanking` produces both. There is no language model
anywhere in here. This is arithmetic over the learner's own answers, and it has
to be reproducible and legible; a model that could not show its working would
make the explanation impossible, which is the whole of what is being built.

Selection itself lives in `quiz_builder.CandidateRanking` — this module decides
how long the session should be, groups what was chosen, and says why.
"""
import random
import statistics
from datetime import datetime

from fastapi import HTTPException
from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.models.attempt import AttemptAnswer, QuizAttempt
from app.models.exam import Exam
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.services import exam_blueprint
from app.services.knowledge_groups import category_ancestry
from app.services.quiz_builder import (CandidateRanking, bank_question_predicate,
                                       general_question_predicate, require_objective)

#: Where a session starts before the learner has finished enough of them for
#: their own habit to be a measurement rather than an anecdote.
DEFAULT_SESSION_LENGTH = 20

#: How many finished sessions it takes before length is set from the learner's
#: own median. Three, because one session is not a habit and two cannot have a
#: median that is not simply the mean of the pair.
SESSIONS_BEFORE_LENGTH_IS_PERSONAL = 3

#: Bounds on an offered length. Not on what the learner may then ask for — the
#: builder's own 1–200 still governs that — only on what is offered unasked.
#: Below five a session says nothing about a topic; above sixty it stops being
#: something anyone finishes in a sitting, and length is set here from what
#: this learner actually finishes.
MIN_SESSION_LENGTH = 5
MAX_SESSION_LENGTH = 60

#: Accuracy below which a topic is described to the learner as a weak area.
WEAK_ACCURACY = 0.6

#: Decayed answers a topic needs behind it before that word is used. Below
#: this the prior in `combined_accuracy` is most of the number, and a topic
#: last answered — correctly — in the spring would come out under the
#: threshold and be reported as a weakness on no evidence at all. Three is the
#: point at which the learner's own answers outweigh the prior.
WEAK_EVIDENCE_ANSWERS = 3.0


def habitual_length(db: Session, user) -> tuple[int | None, int]:
    """The learner's own median finished session, and how many they have finished.

    The median rather than the mean: one abandoned three-question session and
    one marathon both pull a mean somewhere neither of them is, and a learner
    who reliably does thirty should be offered thirty.

    Counted as answers recorded, not as questions the session held, because
    what is wanted is the length they *finish*. Expired attempts are
    out for the same reason they are out of every other figure here.
    """
    rows = db.query(func.count(AttemptAnswer.id)).select_from(QuizAttempt).join(
        AttemptAnswer, AttemptAnswer.attempt_id == QuizAttempt.id).join(
        Quiz, Quiz.id == QuizAttempt.quiz_id).filter(
        QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
        or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)),
    ).group_by(QuizAttempt.id).all()
    lengths = sorted(row[0] for row in rows if row[0])
    if len(lengths) < SESSIONS_BEFORE_LENGTH_IS_PERSONAL:
        return None, len(lengths)
    median = int(round(statistics.median(lengths)))
    return max(MIN_SESSION_LENGTH, min(MAX_SESSION_LENGTH, median)), len(lengths)


def _ago(days: float) -> str:
    """How long ago, at the resolution a learner thinks in."""
    if days < 7:
        return "this week"
    weeks = int(days // 7)
    if weeks < 9:
        return f"{weeks} week{'s' if weeks != 1 else ''} ago"
    months = max(2, int(round(days / 30.44)))
    return f"{months} months ago"


def _percent(weight: float | None) -> str | None:
    return None if weight is None else f"{round(weight, 1):g}"


def _blueprint_paper(db: Session, user, count: int, seed: str) -> list[int]:
    """A paper shaped like the published exam, for a learner with no record yet.

    Seeded, and seeded on the day rather than the moment, because the plan and
    the session it commits are two requests: an unseeded draw would show one
    paper and start another, and the explanation would then be a fiction.
    """
    exam_id = getattr(user, "active_exam_id", None)
    if not exam_id or not exam_blueprint.domains(db, exam_id):
        return []
    ids, _ = exam_blueprint.sample(
        db, exam_id, count, predicate=bank_question_predicate(user) & general_question_predicate(),
        rng=random.Random(seed))
    return ids


def _topic_rows(db: Session, ranking: CandidateRanking, question_ids: list[int],
                cold_start: bool) -> list[dict]:
    """Group the chosen questions by discipline, and say why each is there.

    Counted from the questions actually chosen rather than predicted from the
    ranking, so the plan cannot drift from the session: every count below is a
    tally of `question_ids`.

    Grouped by the top-level category — the discipline — because that is the
    vocabulary the Analysis page already reports a learner's weaknesses in, and
    a plan phrased in leaf topics would be twenty rows of one question each.
    """
    categories = db.query(QuestionCategory).all()
    names = {cat.id: cat.name for cat in categories}
    ancestry = category_ancestry(categories)
    discipline_of = {cid: chain[-1] for cid, chain in ancestry.items() if chain}

    groups: dict[int | None, dict] = {}
    for question_id in question_ids:
        category = ranking.category_of.get(question_id)
        key = discipline_of.get(category) if category is not None else None
        row = groups.setdefault(key, {
            "category_id": key, "name": names.get(key, "Unfiled") if key else "Unfiled",
            "count": 0, "new_count": 0, "review_count": 0,
            "_leaves": set(), "_ages": [],
        })
        row["count"] += 1
        if question_id in ranking.recall:
            row["review_count"] += 1
            row["_ages"].append(ranking.latest[question_id][2])
        else:
            row["new_count"] += 1
        if category is not None:
            row["_leaves"].add(category)

    rows = []
    for row in groups.values():
        leaves = row.pop("_leaves")
        ages = row.pop("_ages")
        # A discipline's own numbers, not the average of its leaves': what the
        # learner is told about "Cardiology" has to be about cardiology.
        scored = leaves | ({row["category_id"]} if row["category_id"] else set())
        accuracy = ranking.combined_accuracy(scored)
        evidence = ranking.evidence_weight(scored)
        weight = ranking.weights.get(row["category_id"]) if ranking.weights else None
        row["accuracy"] = round(100 * accuracy) if accuracy is not None else None
        row["weight"] = round(weight, 1) if weight is not None else None
        row["reason"] = _reason(row, accuracy, evidence, weight, ages, cold_start)
        rows.append(row)
    # Biggest share of the session first: the plan should read in the order the
    # learner's attention is being spent.
    rows.sort(key=lambda row: (-row["count"], row["name"]))
    return rows


def _reason(row: dict, accuracy: float | None, evidence: float, weight: float | None,
            ages: list[float], cold_start: bool) -> str:
    """One line saying why this topic is in the session.

    Strongest reason first, so the sentence a learner reads is the reason and
    not merely a true statement about the topic. A weak topic is usually also
    due for review — being told the accuracy is the more useful of the two.
    """
    share = _percent(weight)
    if cold_start:
        return f"Worth {share}% of the exam" if share else "Broad coverage — no history to go on yet"
    if accuracy is not None and accuracy < WEAK_ACCURACY and evidence >= WEAK_EVIDENCE_ANSWERS:
        return f"Weak area — {round(100 * accuracy)}% correct so far"
    if ages and row["review_count"] >= row["new_count"]:
        return f"Due for review — last answered {_ago(min(ages))}"
    if accuracy is None:
        return f"Not attempted yet, worth {share}% of the exam" if share else "Not attempted yet"
    return f"Worth {share}% of the exam" if share else "Keeping your coverage even"


def prepare_session(db: Session, user, count: int | None = None, now: datetime | None = None) -> dict:
    """Build a session for this learner and the account of why it is that one.

    Nothing is written. The same call is made again to commit, which is what
    makes the plan honest: the questions named here are the questions started.
    """
    require_objective(db, user)
    now = now or datetime.utcnow()
    ranking = CandidateRanking(db, user, now=now)
    available = len(ranking.rows)
    if not available:
        raise HTTPException(400, "There are no questions in your bank yet")

    habitual, finished = habitual_length(db, user)
    asked = count is not None
    if asked:
        length = max(1, min(200, count))
    else:
        length = habitual or DEFAULT_SESSION_LENGTH
    length = min(length, available)

    # No finished session is the honest test for cold start, not "no answers":
    # a learner who abandoned one question has told us nothing to personalize
    # from, and being handed a "weak area" on the strength of it is the fake
    # personalization this is meant to avoid.
    cold_start = finished == 0
    seed = f"{user.id}:{length}:{now.date().isoformat()}"
    question_ids = _blueprint_paper(db, user, length, seed) if cold_start else []
    blueprint_led = bool(question_ids)
    if not question_ids:
        question_ids = ranking.select(length)
    if not question_ids:
        raise HTTPException(400, "There are no questions in your bank yet")

    topics = _topic_rows(db, ranking, question_ids, cold_start)
    review_count = sum(row["review_count"] for row in topics)
    new_count = sum(row["new_count"] for row in topics)
    exam_name = None
    if getattr(user, "active_exam_id", None):
        exam = db.get(Exam, user.active_exam_id)
        exam_name = exam.name if exam else None

    return {
        "count": len(question_ids),
        "question_ids": question_ids,
        "basis": "cold_start" if cold_start else "personalized",
        "new_count": new_count,
        "review_count": review_count,
        "available": available,
        "exam_name": exam_name,
        "summary": _summary(cold_start, blueprint_led, new_count, review_count),
        "length_reason": _length_reason(asked, habitual, finished, len(question_ids), available),
        "topics": topics,
    }


def _summary(cold_start: bool, blueprint_led: bool, new_count: int, review_count: int) -> str:
    if cold_start:
        spread = "the exam blueprint" if blueprint_led else "your whole question bank"
        return ("You haven't finished a session yet, so this is an even spread across "
                f"{spread} rather than a personalized one.")
    if review_count:
        return (f"{new_count} new and {review_count} due for review, ranked by what would "
                "move your score the most.")
    return f"{new_count} new questions, ranked by what would move your score the most."


def _length_reason(asked: bool, habitual: int | None, finished: int,
                   length: int, available: int) -> str:
    if asked:
        reason = f"{length} questions, your choice."
    elif habitual:
        reason = f"You usually finish {habitual} questions, so that's the length."
    else:
        remaining = SESSIONS_BEFORE_LENGTH_IS_PERSONAL - finished
        reason = (f"{DEFAULT_SESSION_LENGTH} to start with — after {remaining} more finished "
                  f"session{'s' if remaining != 1 else ''} this matches your own median.")
    if length >= available:
        reason += " That is everything left in your bank."
    return reason