3279e14bb2
There will be no courses. What was there: one draft called "jk" with two empty lessons, and 4,000 lines of code around it — courses, modules, lessons, enrolments, per-lesson progress, SCORM, BigBlueButton, completion certificates, three React pages, a router, two models. Its real cost was everywhere else. Every query that measured practice had to remember `Quiz.course_id.is_(None)`, and forgetting it in one place would have silently mixed course attempts into a learner's analytics; the bank predicate carried a subquery to exclude a course's own questions from every search, recommendation and share; quiz access had a second, parallel rule about enrolment. All of that is gone, so the remaining rules say what they mean. `quizzes.allow_review` goes with it. It was only ever enforced for a course quiz, so it had become a promise nothing keeps — the public session page was still offering "no answer review" about sessions that review fine. The fixtures' question 5 lived in a course quiz and stood for "a question that exists but is not in your bank". There is no such thing now — a question is in the bank unless it is deleted — so the counts it kept out of the numbers are back in, and the tests that turned on it now turn on deletion or on the attempt that actually holds a question. Files the LMS uploaded stay on disk and stay protected: LEGACY_LMS_PREFIXES in app/utils/upload_access.py is what keeps them unreachable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
698 lines
32 KiB
Python
698 lines
32 KiB
Python
"""Permission-safe, saved general-bank tests and category selection."""
|
||
import random
|
||
import statistics
|
||
from collections import defaultdict
|
||
from datetime import datetime
|
||
from typing import Literal
|
||
|
||
from fastapi import HTTPException
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from sqlalchemy import func, or_, select
|
||
|
||
from app.models.attempt import AttemptAnswer, QuizAttempt
|
||
from app.models.favorite import Favorite
|
||
from app.models.question import Question
|
||
from app.models.question_category import QuestionCategory, QuestionCategoryLink
|
||
from app.models.quiz import Quiz
|
||
from app.utils.quiz_questions import add_questions_to_quiz
|
||
|
||
|
||
def category_descendants(categories, selected):
|
||
parents = {c.id: c.parent_id for c in categories}
|
||
result = set(selected)
|
||
if result - parents.keys():
|
||
raise HTTPException(400, "Category not found")
|
||
while True:
|
||
expanded = result | {cid for cid, parent in parents.items() if parent in result}
|
||
if expanded == result:
|
||
return result
|
||
result = expanded
|
||
|
||
|
||
def category_breadcrumbs(categories, category_id):
|
||
by_id = {c.id: c for c in categories}
|
||
path, seen = [], set()
|
||
while category_id in by_id and category_id not in seen:
|
||
seen.add(category_id)
|
||
cat = by_id[category_id]
|
||
path.append({"id": cat.id, "name": cat.name})
|
||
category_id = cat.parent_id
|
||
return list(reversed(path))
|
||
|
||
|
||
def validate_parent(categories, category_id, parent_id):
|
||
if parent_id is None:
|
||
return
|
||
if parent_id not in {c.id for c in categories}:
|
||
raise HTTPException(400, "Parent category not found")
|
||
if category_id is not None and parent_id in category_descendants(categories, [category_id]):
|
||
raise HTTPException(400, "A category cannot be its own parent or a descendant's child")
|
||
|
||
|
||
def general_question_predicate():
|
||
# A deleted question is out of every path at once — bank, builder, search,
|
||
# recommendations, share — because it is excluded here rather than at each
|
||
# call site, where one of them would eventually be forgotten.
|
||
return Question.deleted_at.is_(None)
|
||
|
||
|
||
def shareable_question_predicate():
|
||
"""What a public share link may contain.
|
||
|
||
Once it was "questions whose author ticked shared". Per-question sharing is
|
||
gone: a question in the bank is in the bank, and who may *manage* one is
|
||
decided by the grant tree rather than by a flag its author set. So the only
|
||
thing still excluded here is what is excluded everywhere — a deleted
|
||
question.
|
||
"""
|
||
return general_question_predicate()
|
||
|
||
|
||
def bank_question_predicate(user):
|
||
"""What a learner's bank holds.
|
||
|
||
Every question in it. The flag this used to consult defaulted to 1 and was
|
||
settable only from a route nothing called, so in practice it divided the
|
||
bank into "everything" and "everything, plus your own private ones" — a
|
||
distinction that cost every recommendation denominator a join and never
|
||
changed an answer.
|
||
"""
|
||
del user # Kept in the signature: the exam scope and grants still take one.
|
||
return general_question_predicate()
|
||
|
||
|
||
def exam_scope_predicate(db, user):
|
||
"""Limit the bank to the learner's active exam, if they have chosen one.
|
||
|
||
No selection means the whole bank, so existing behaviour is unchanged until
|
||
someone picks an exam. A question with no exam links stays visible rather
|
||
than disappearing, since unlinked content is unclassified, not excluded.
|
||
"""
|
||
from sqlalchemy import select as sa_select
|
||
|
||
from app.models.exam import QuestionExamLink
|
||
|
||
exam_id = getattr(user, "active_exam_id", None)
|
||
if not exam_id:
|
||
return None
|
||
in_exam = sa_select(QuestionExamLink.question_id).where(QuestionExamLink.exam_id == exam_id)
|
||
unlinked = ~sa_select(QuestionExamLink.question_id).where(
|
||
QuestionExamLink.question_id == Question.id).exists()
|
||
return Question.id.in_(in_exam) | unlinked
|
||
|
||
|
||
def bank_query(db, user):
|
||
return db.query(Question).filter(bank_question_predicate(user))
|
||
|
||
|
||
def filtered_bank_query(db, user, category_ids=(), state="all", difficulty=None, article_ids=(), tag_ids=(), system_ids=()):
|
||
"""The bank a learner can draw a session from, narrowed by their filters.
|
||
|
||
Scoped to their exam, like browsing and searching already were. It was not,
|
||
and the two disagreed in the worst possible direction: a learner studying
|
||
for an exam with no content linked to it saw an empty question bank and was
|
||
then handed a full session built from every question in it. Whatever the
|
||
right pool is, it has to be the same pool in both places.
|
||
"""
|
||
query = bank_query(db, user)
|
||
scope = exam_scope_predicate(db, user)
|
||
if scope is not None:
|
||
query = query.filter(scope)
|
||
if difficulty:
|
||
query = query.filter(Question.difficulty == difficulty)
|
||
if article_ids:
|
||
from app.models.article import QuestionArticleLink
|
||
query = query.filter(Question.id.in_(select(QuestionArticleLink.question_id).where(
|
||
QuestionArticleLink.article_id.in_(article_ids))))
|
||
if tag_ids:
|
||
from sqlalchemy import text as sa_text
|
||
tag_list = list(dict.fromkeys(tag_ids))
|
||
matching = list(db.execute(sa_text("""
|
||
SELECT question_id FROM question_tag_links
|
||
WHERE tag_id = ANY(:tag_ids)
|
||
GROUP BY question_id
|
||
HAVING COUNT(DISTINCT tag_id) = :cnt
|
||
"""), {"tag_ids": tag_list, "cnt": len(tag_list)}).scalars())
|
||
if matching:
|
||
query = query.filter(Question.id.in_(matching))
|
||
else:
|
||
query = query.filter(Question.id.is_(None)) # No questions match all tags.
|
||
if system_ids:
|
||
# A question is never filed under an organ system directly — the topic
|
||
# it sits under carries one. It used to reach a system through a
|
||
# symptom keyword the question happened to mention, and only about half
|
||
# of them mentioned one that had been filed.
|
||
#
|
||
# ANY of these systems, unlike tag_ids above, which is an AND: asking
|
||
# for cardiovascular and respiratory means either, because no question
|
||
# is both.
|
||
wanted = list(dict.fromkeys(system_ids))
|
||
named = [cid for (cid,) in db.query(QuestionCategory.id).filter(
|
||
QuestionCategory.system_id.in_(wanted)).all()]
|
||
if not named:
|
||
return query.filter(Question.id.is_(None))
|
||
# And everything beneath them: a subtopic with no system of its own
|
||
# belongs to the system of the topic above it, which is the same rule
|
||
# the analysis groups by.
|
||
topics = category_descendants(db.query(QuestionCategory).all(), named)
|
||
query = query.filter(or_(
|
||
Question.question_category_id.in_(topics),
|
||
Question.id.in_(select(QuestionCategoryLink.question_id).where(
|
||
QuestionCategoryLink.category_id.in_(topics)))))
|
||
if category_ids:
|
||
ids = category_descendants(db.query(QuestionCategory).all(), category_ids)
|
||
query = query.filter(or_(
|
||
Question.question_category_id.in_(ids),
|
||
Question.id.in_(select(QuestionCategoryLink.question_id).where(QuestionCategoryLink.category_id.in_(ids))),
|
||
))
|
||
if state == "bookmarked":
|
||
query = query.filter(Question.id.in_(select(Favorite.question_id).where(Favorite.user_id == user.id)))
|
||
elif state in ("unused", "incorrect"):
|
||
# Latest completed, nonexpired general-bank answer; deterministic ties.
|
||
answers = db.query(
|
||
AttemptAnswer.question_id.label("question_id"), AttemptAnswer.is_correct.label("is_correct"),
|
||
func.row_number().over(partition_by=AttemptAnswer.question_id, order_by=(
|
||
QuizAttempt.completed_at.desc(), QuizAttempt.id.desc(), AttemptAnswer.id.desc(),
|
||
)).label("rank"),
|
||
).join(QuizAttempt, AttemptAnswer.attempt_id == QuizAttempt.id).join(Quiz, QuizAttempt.quiz_id == Quiz.id).filter(
|
||
QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
|
||
or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)),
|
||
).subquery()
|
||
if state == "unused":
|
||
query = query.filter(~Question.id.in_(select(answers.c.question_id)))
|
||
else:
|
||
query = query.filter(Question.id.in_(select(answers.c.question_id).where(
|
||
answers.c.rank == 1, answers.c.is_correct.is_(False),
|
||
)))
|
||
elif state != "all":
|
||
raise HTTPException(400, "Invalid question state")
|
||
return query
|
||
|
||
|
||
class TestOptions(BaseModel):
|
||
title: str = Field(min_length=1, max_length=200)
|
||
mode: Literal["timed", "learning"] = "timed"
|
||
time_limit_minutes: int | None = Field(default=None, gt=0)
|
||
is_shared: bool = False
|
||
#: Sitting the same questions again. Analysed on its own page, left out of
|
||
#: the figures that claim to say how much of the bank you know.
|
||
is_repetition: bool = False
|
||
|
||
@field_validator("title")
|
||
@classmethod
|
||
def nonblank_title(cls, value):
|
||
if not value.strip():
|
||
raise ValueError("Title is required")
|
||
return value.strip()
|
||
|
||
|
||
class CreateFromBankRequest(TestOptions):
|
||
question_ids: list[int] = Field(min_length=1)
|
||
|
||
|
||
class GenerateTestRequest(TestOptions):
|
||
category_ids: list[int] = Field(default_factory=list)
|
||
state: Literal["all", "unused", "incorrect", "bookmarked"] = "all"
|
||
count: int = Field(ge=1, le=200)
|
||
expected_count: int | None = Field(default=None, ge=0)
|
||
difficulty: Literal["easy", "medium", "hard"] | None = None
|
||
#: "blueprint" draws a paper shaped like the real exam — the examining
|
||
#: board's published weights, rather than a uniform draw from the bank.
|
||
algorithm: Literal["random", "adaptive", "blueprint"] = "random"
|
||
article_ids: list[int] = Field(default_factory=list)
|
||
tag_ids: list[int] = Field(default_factory=list)
|
||
#: Organ systems. Matched as "any tag beneath this system", where
|
||
#: tag_ids is "every one of these tags".
|
||
system_ids: list[int] = Field(default_factory=list)
|
||
explicit_ids: list[int] = Field(default_factory=list)
|
||
|
||
|
||
#: Seconds a timed block allows per question. The pace a real paper is sat at,
|
||
#: so a block of forty runs an hour — and so a learner rehearsing on this bank
|
||
#: is rehearsing the clock as well as the questions.
|
||
SECONDS_PER_QUESTION = 90
|
||
|
||
|
||
def exam_minutes(data, count: int) -> int | None:
|
||
"""How long a timed block gets, rounded up to the minute.
|
||
|
||
Set from the number of questions rather than asked for. Choosing a limit
|
||
is a decision nobody has the information to make — the pace is a property
|
||
of the exam being rehearsed, not a preference — and a block sat at the
|
||
wrong pace teaches the wrong pace. An explicit limit is still honoured for
|
||
the cases that genuinely differ.
|
||
"""
|
||
if getattr(data, "mode", None) != "timed":
|
||
return None
|
||
asked = getattr(data, "time_limit_minutes", None)
|
||
if asked:
|
||
return asked
|
||
return max(1, -(-(count * SECONDS_PER_QUESTION) // 60))
|
||
|
||
|
||
def create_saved_test(db, user, data, question_ids):
|
||
ids = list(dict.fromkeys(question_ids))
|
||
if not 1 <= len(ids) <= 200:
|
||
raise HTTPException(400, "Select between 1 and 200 questions")
|
||
query = bank_query(db, user).filter(Question.id.in_(ids))
|
||
if data.is_shared:
|
||
query = query.filter(shareable_question_predicate())
|
||
if query.count() != len(ids):
|
||
raise HTTPException(400, "Some questions are missing, private, or unavailable for this test")
|
||
quiz = Quiz(user_id=user.id, title=data.title, mode=data.mode,
|
||
time_limit_minutes=exam_minutes(data, len(ids)),
|
||
questions_count=len(ids), is_published=0, is_shared=int(data.is_shared),
|
||
is_repetition=int(getattr(data, "is_repetition", False)))
|
||
db.add(quiz)
|
||
db.flush()
|
||
add_questions_to_quiz(db, quiz.id, ids)
|
||
db.commit()
|
||
db.refresh(quiz)
|
||
return {"id": quiz.id, "title": quiz.title, "questions_count": quiz.questions_count}
|
||
|
||
|
||
def blueprint_weights(db, user) -> dict[int, float]:
|
||
"""Each category mapped to the share of the paper its domain accounts for.
|
||
|
||
Empty when the learner has no study objective, or when the objective has no
|
||
published weights — in which case selection falls back to being about
|
||
weakness alone, which is what it always was.
|
||
|
||
`categories_for` already returns the descendants of a mapped category, so a
|
||
leaf topic under a weighted domain is in here too.
|
||
"""
|
||
exam_id = getattr(user, "active_exam_id", None)
|
||
if not exam_id:
|
||
return {}
|
||
from app.services import exam_blueprint
|
||
|
||
weights: dict[int, float] = {}
|
||
for line in exam_blueprint.domains(db, exam_id):
|
||
if line.weight is None:
|
||
continue
|
||
for category_id in exam_blueprint.categories_for(db, line.id):
|
||
weights[category_id] = float(line.weight)
|
||
return weights
|
||
|
||
|
||
#: How long one answer keeps half its weight as evidence about the learner.
|
||
#:
|
||
#: The curve is exponential, `0.5 ** (age / half_life)`. A fixed window was the
|
||
#: obvious first thing and is wrong in a way that shows: it makes an answer
|
||
#: twenty-nine days old count in full and one thirty-one days old count for
|
||
#: nothing, so a topic crosses a cliff overnight and the ranking lurches
|
||
#: without the learner having done anything. Exponential also has the property
|
||
#: that matters for an order recomputed on every visit — it is memoryless, so
|
||
#: an answer's weight depends only on its own age and not on what has been
|
||
#: answered since, which is what keeps two consecutive sessions consistent with
|
||
#: each other. A power law fits very long retention slightly better, but it
|
||
#: needs an arbitrary offset to avoid a singularity at age zero and a second
|
||
#: parameter nothing here could justify; one named half-life describes the
|
||
#: whole of this curve.
|
||
#:
|
||
#: Thirty days because that is about the turn of a revision cycle. It puts a
|
||
#: ninety-day-old answer at an eighth of the weight of a fresh one — so what
|
||
#: was missed last week clearly outranks what was missed in spring — while not
|
||
#: writing off a topic revised last month as forgotten.
|
||
EVIDENCE_HALF_LIFE_DAYS = 30.0
|
||
|
||
#: What the last outcome says about whether a question is still known, at the
|
||
#: moment it was answered. Not 1 and 0: one answer is one observation, and a
|
||
#: right answer can be a guess as easily as a wrong one can be a slip. These
|
||
#: are the numbers the recycling order always used, named here because time
|
||
#: now moves them.
|
||
RECALL_AFTER_CORRECT = 0.85
|
||
RECALL_AFTER_WRONG = 0.25
|
||
|
||
#: Recall of a question there is no useful evidence about either way — and the
|
||
#: accuracy assumed for a topic never answered in, so it sorts between the
|
||
#: learner's strong and weak areas rather than jumping the queue.
|
||
NEUTRAL_RECALL = 0.5
|
||
|
||
#: Recall below which a question is due to come round again. A correct answer
|
||
#: decays past this at about three and a half weeks, which is the review
|
||
#: interval this is meant to express; anything ever answered wrongly is below
|
||
#: it from the moment it was answered.
|
||
DUE_RECALL = 0.7
|
||
|
||
#: The most of one session that may be spent on questions already seen. Review
|
||
#: is not what you do once the new material runs out — that rule meant a
|
||
#: learner with three thousand unseen questions never saw a repeat, which is
|
||
#: no spaced repetition at all. But somebody who opens the app and is handed
|
||
#: twenty questions they have already answered does not open it again, so the
|
||
#: majority of any session is still new.
|
||
MAX_REVIEW_SHARE = 0.4
|
||
|
||
#: Answers' worth of "no idea" mixed into every topic's accuracy. Without it a
|
||
#: single correct answer made a topic 100% known and it never came back, which
|
||
#: is the one thing a ranking that claims to decay must not do.
|
||
PRIOR_ANSWERS = 2.0
|
||
|
||
#: How fast a topic's priority falls as the session keeps drawing from it.
|
||
CATEGORY_DAMPING = 0.5
|
||
|
||
#: The three labels, in order, and the readiness at which each becomes the
|
||
#: right level to be working at.
|
||
#:
|
||
#: A learner getting a third of a topic right is not helped by its hardest
|
||
#: questions — they will miss those too and learn nothing from the miss — and
|
||
#: one getting nine in ten right is not helped by its easiest. The bands are
|
||
#: readiness, which is shrunk accuracy, so a topic answered twice sits near
|
||
#: 0.5 and gets medium: the middle is also the honest default when almost
|
||
#: nothing is known about somebody.
|
||
DIFFICULTY_ORDER = ("easy", "medium", "hard")
|
||
#: 0.45 and not 0.55 for the first edge: a topic with no evidence sits at
|
||
#: exactly NEUTRAL_RECALL, and a boundary above that would hand every untouched
|
||
#: topic its easiest questions — which is a poor way to find out what somebody
|
||
#: knows and a slow way to start.
|
||
DIFFICULTY_BANDS = ((0.45, "easy"), (0.78, "medium"), (1.01, "hard"))
|
||
|
||
#: What a question at the wrong level is still worth. One step away is most of
|
||
#: the value — a medium question is a perfectly good thing to answer while you
|
||
#: are on easy ones — and two steps is half.
|
||
#:
|
||
#: A multiplier and not a filter, deliberately, for the same reason the
|
||
#: reranker is a permutation: a bank thinned to one difficulty is a bank three
|
||
#: times smaller, and on a narrow topic that means the same eight questions
|
||
#: every time. Nothing is ever excluded for being the wrong level; it is
|
||
#: preferred against.
|
||
DIFFICULTY_FIT = (1.0, 0.72, 0.5)
|
||
|
||
#: And what an unlabelled question is worth. Just under a perfect fit, so a
|
||
#: labelled question at the right level wins a tie, and comfortably above a
|
||
#: two-step miss — an unknown difficulty is unknown, not wrong. Every question
|
||
#: in the bank carries a label today; this is for what is written tomorrow.
|
||
UNRATED_FIT = 0.92
|
||
|
||
|
||
def target_difficulty(readiness: float | None) -> str:
|
||
"""The level a learner working at this readiness should be asked at."""
|
||
if readiness is None:
|
||
return "medium"
|
||
for ceiling, level in DIFFICULTY_BANDS:
|
||
if readiness < ceiling:
|
||
return level
|
||
return "hard"
|
||
|
||
|
||
def recency_weight(age_days: float) -> float:
|
||
"""How much evidence that old still counts for."""
|
||
return 0.5 ** (max(0.0, age_days) / EVIDENCE_HALF_LIFE_DAYS)
|
||
|
||
|
||
def recall_probability(was_correct: bool, age_days: float) -> float:
|
||
"""Chance a question is still known, given how it last went and how long ago.
|
||
|
||
Decays towards a coin flip rather than towards zero. Forgetting a right
|
||
answer does not turn it into a wrong one, and time does not turn a wrong
|
||
answer into a right one either; both outcomes end up saying nothing, which
|
||
is exactly the state in which the question is worth asking again.
|
||
"""
|
||
settled = RECALL_AFTER_CORRECT if was_correct else RECALL_AFTER_WRONG
|
||
return NEUTRAL_RECALL + (settled - NEUTRAL_RECALL) * recency_weight(age_days)
|
||
|
||
|
||
class CandidateRanking:
|
||
"""One learner, one filtered bank, and everything needed to order it.
|
||
|
||
Built once and read many times. Selection and the plan that describes it
|
||
are two readings of this one object rather than two calculations that would
|
||
have to be kept in step — a plan that does not describe the session it
|
||
starts is worse than no plan.
|
||
|
||
The rules, in the order they apply:
|
||
|
||
**Unseen material is most of the session.** A question never met teaches
|
||
more than one already answered, so it takes every slot review is not
|
||
holding.
|
||
|
||
**Review takes the rest, up to `MAX_REVIEW_SHARE`, and only what is due.**
|
||
Due means recall has decayed below `DUE_RECALL` — everything answered
|
||
wrongly, and everything answered correctly long enough ago to be worth
|
||
checking.
|
||
|
||
**Within either, highest value first, damped per topic.** Value for unseen
|
||
material is the topic's impact, `(1 − accuracy) × blueprint weight`; for
|
||
review it is `(1 − recall) × blueprint weight`. Each pick halves its
|
||
topic's priority, which stops a session of twenty becoming twenty
|
||
questions from one subject — and gives a learner with no history at all a
|
||
spread across the paper instead of the heaviest domain entire.
|
||
|
||
Accuracy and recall both fade with time; see `EVIDENCE_HALF_LIFE_DAYS`.
|
||
"""
|
||
|
||
def __init__(self, db, user, category_ids=(), state="all", difficulty=None, now=None):
|
||
self.now = now or datetime.utcnow()
|
||
query = filtered_bank_query(db, user, category_ids, state, difficulty)
|
||
# No cap. This used to take the first 2,000 rows, which on a
|
||
# 2,948-question bank meant adaptive selection could not see about a
|
||
# third of it, and which third depended on database order. Two integer
|
||
# columns per question is not a size worth protecting against.
|
||
#
|
||
# Sorted by id so that every scan below, and so every tie, resolves the
|
||
# same way twice running: the plan and the session it commits are
|
||
# separate calls, and a ranking that reshuffles between them would make
|
||
# the plan a guess.
|
||
self.rows = sorted(((row[0], row[1]) for row in query.with_entities(
|
||
Question.id, Question.question_category_id).all()), key=lambda row: row[0])
|
||
self.category_of = dict(self.rows)
|
||
#: question id → "easy" | "medium" | "hard", where one is recorded. A
|
||
#: third integer column per question, fetched with the other two.
|
||
self.difficulty_of = {
|
||
row[0]: row[1] for row in query.with_entities(
|
||
Question.id, Question.difficulty).all() if row[1]}
|
||
|
||
answered = db.query(AttemptAnswer.question_id, AttemptAnswer.is_correct, QuizAttempt.completed_at).join(
|
||
QuizAttempt, QuizAttempt.id == AttemptAnswer.attempt_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)),
|
||
AttemptAnswer.question_id.in_([row[0] for row in self.rows]),
|
||
).order_by(QuizAttempt.completed_at.desc(), QuizAttempt.id.desc()).all()
|
||
|
||
#: question id → (was correct, when, age in days) of the latest answer.
|
||
self.latest: dict[int, tuple[bool, object, float]] = {}
|
||
evidence: dict = defaultdict(lambda: [0.0, 0.0])
|
||
for question_id, was_correct, when in answered:
|
||
age = max(0.0, (self.now - when).total_seconds() / 86400.0)
|
||
self.latest.setdefault(question_id, (bool(was_correct), when, age))
|
||
# Looked up, not scanned. This was a linear search through every
|
||
# candidate for every answer — the slowest part of building a
|
||
# session.
|
||
category = self.category_of.get(question_id)
|
||
if category is None:
|
||
continue
|
||
counts = evidence[category]
|
||
counts[0] += recency_weight(age)
|
||
if was_correct:
|
||
counts[1] += recency_weight(age)
|
||
self._evidence = evidence
|
||
self.recall = {question_id: recall_probability(was_correct, age)
|
||
for question_id, (was_correct, _, age) in self.latest.items()}
|
||
|
||
self.weights = blueprint_weights(db, user)
|
||
# The middle of what the board publishes, for a topic it does not
|
||
# mention. A zero would make unmapped material unreachable; the highest
|
||
# would make it the priority. Neither is a claim the blueprint supports.
|
||
self._neutral_weight = statistics.median(self.weights.values()) if self.weights else 1.0
|
||
|
||
self.unseen = [row for row in self.rows if row[0] not in self.latest]
|
||
self.seen = [row for row in self.rows if row[0] in self.latest]
|
||
self.due = [row for row in self.seen if self.recall[row[0]] < DUE_RECALL]
|
||
|
||
def accuracy(self, category) -> float:
|
||
"""Share of this topic answered correctly, recent answers counting most.
|
||
|
||
Pulled towards `NEUTRAL_RECALL` by `PRIOR_ANSWERS`, so one lucky answer
|
||
does not settle a topic and a topic left alone drifts back to unknown.
|
||
"""
|
||
total, correct = self._evidence.get(category, (0.0, 0.0))
|
||
return (correct + PRIOR_ANSWERS * NEUTRAL_RECALL) / (total + PRIOR_ANSWERS)
|
||
|
||
def evidence_weight(self, categories) -> float:
|
||
"""Answers' worth of evidence behind these topics, after decay.
|
||
|
||
How much the accuracy beside it is worth. Two answers from the spring
|
||
come to a fifth of one answer from yesterday, and a claim about the
|
||
learner should not be made on the strength of them.
|
||
"""
|
||
return sum(self._evidence.get(category, (0.0, 0.0))[0] for category in categories)
|
||
|
||
def combined_accuracy(self, categories) -> float | None:
|
||
"""Accuracy over several topics at once, or None with nothing to go on.
|
||
|
||
Evidence is pooled before the ratio is taken, rather than the topics'
|
||
accuracies being averaged: a discipline whose two hundred cardiology
|
||
answers went badly and whose one rheumatology answer went well is not
|
||
halfway between the two.
|
||
"""
|
||
total = self.evidence_weight(categories)
|
||
if not total:
|
||
return None
|
||
correct = sum(self._evidence.get(category, (0.0, 0.0))[1] for category in categories)
|
||
return (correct + PRIOR_ANSWERS * NEUTRAL_RECALL) / (total + PRIOR_ANSWERS)
|
||
|
||
def weight(self, category) -> float:
|
||
"""The topic's share of the real paper, or a neutral stand-in."""
|
||
return self.weights.get(category, self._neutral_weight) if self.weights else 1.0
|
||
|
||
def impact(self, category) -> float:
|
||
"""How much a question here could move the score."""
|
||
return (1 - self.accuracy(category)) * self.weight(category)
|
||
|
||
def difficulty_fit(self, row) -> float:
|
||
"""How well this question's level suits where the learner is on its topic.
|
||
|
||
This is the half of "adaptive" that was missing: until now difficulty
|
||
was a filter a learner could set and nothing the session did on its
|
||
own, so somebody at 30% on a topic and somebody at 90% were asked the
|
||
same questions in the same order.
|
||
|
||
Measured per topic rather than overall, because a learner is not one
|
||
level: strong on growth and weak on arrhythmias is the normal case, and
|
||
an average across the two describes nobody.
|
||
"""
|
||
level = self.difficulty_of.get(row[0])
|
||
if level is None:
|
||
return UNRATED_FIT
|
||
wanted = target_difficulty(self.accuracy(row[1]))
|
||
try:
|
||
distance = abs(DIFFICULTY_ORDER.index(level) - DIFFICULTY_ORDER.index(wanted))
|
||
except ValueError:
|
||
return UNRATED_FIT
|
||
return DIFFICULTY_FIT[min(distance, len(DIFFICULTY_FIT) - 1)]
|
||
|
||
def _unseen_value(self, row) -> float:
|
||
return self.impact(row[1])
|
||
|
||
def _review_value(self, row) -> float:
|
||
return (1 - self.recall[row[0]]) * self.weight(row[1])
|
||
|
||
def value(self, row) -> float:
|
||
"""What one candidate is worth, whichever pool it came from.
|
||
|
||
Topic first, level second. Which topic to spend a question on is the
|
||
decision that moves a score; how hard that question should be is how
|
||
the session meets the learner where they are inside it.
|
||
"""
|
||
base = self._review_value(row) if row[0] in self.recall else self._unseen_value(row)
|
||
return base * self.difficulty_fit(row)
|
||
|
||
def review_budget(self, count: int) -> int:
|
||
"""Slots this session gives to questions already seen."""
|
||
return min(len(self.due), round(MAX_REVIEW_SHARE * count)) if count > 0 else 0
|
||
|
||
def select(self, count: int) -> list[int]:
|
||
"""The questions, in the order they will be asked."""
|
||
if count <= 0:
|
||
return []
|
||
damping: dict = defaultdict(lambda: 1.0)
|
||
taken: list[int] = []
|
||
spent: set[int] = set()
|
||
|
||
def draw(pool, budget):
|
||
candidates = [row for row in pool if row[0] not in spent]
|
||
picked = 0
|
||
while candidates and picked < budget:
|
||
best, best_score = None, None
|
||
for row in candidates:
|
||
score = self.value(row) * damping[row[1]]
|
||
if best_score is None or score > best_score:
|
||
best, best_score = row, score
|
||
candidates.remove(best)
|
||
taken.append(best[0])
|
||
spent.add(best[0])
|
||
picked += 1
|
||
damping[best[1]] *= CATEGORY_DAMPING
|
||
|
||
review = self.review_budget(count)
|
||
draw(self.unseen, count - review)
|
||
draw(self.due, review)
|
||
# Whatever the two budgets could not fill. A bank with nothing unseen
|
||
# left, or nothing due, still owes the learner the length they asked
|
||
# for.
|
||
draw(self.rows, count - len(taken))
|
||
return taken
|
||
|
||
|
||
def adaptive_select(db, user, count, category_ids, state, difficulty, now=None):
|
||
"""Adaptive selection: the questions most likely to raise the learner's score.
|
||
|
||
The rules live on `CandidateRanking`, which the prepared session reads to
|
||
explain itself. This is the same selection reached from the Adaptive toggle
|
||
on the manual builder.
|
||
"""
|
||
return CandidateRanking(db, user, category_ids, state, difficulty, now).select(count)
|
||
|
||
|
||
def require_objective(db, user) -> None:
|
||
"""Refuse to build a session for somebody who has not said what they study.
|
||
|
||
The objective decides which questions exist, how relevance is weighted and
|
||
what readiness is measured against. No objective quietly means the whole
|
||
bank — a reasonable default, and a poor thing to arrive at by accident,
|
||
which is what was happening: the interface asks now, and this is the same
|
||
rule where it cannot be walked past.
|
||
|
||
Only where there is something to choose. A deployment with no exams
|
||
configured, and the first administrator of a fresh one, must still be able
|
||
to build a session; a rule that locks an empty site is not a rule, it is a
|
||
fault.
|
||
"""
|
||
if getattr(user, "active_exam_id", None):
|
||
return
|
||
from app.models.exam import Exam
|
||
|
||
if not db.query(Exam.id).filter(Exam.is_active == 1).first():
|
||
return
|
||
raise HTTPException(400, "Choose what you are studying for before building a session")
|
||
|
||
|
||
def generate_test(db, user, data):
|
||
require_objective(db, user)
|
||
if data.algorithm == "blueprint":
|
||
return _blueprint_test(db, user, data)
|
||
if data.algorithm == "adaptive":
|
||
ids = adaptive_select(db, user, data.count, data.category_ids, data.state, data.difficulty)
|
||
if len(ids) < data.count:
|
||
raise HTTPException(400, f"Only {len(ids)} questions available; requested {data.count}")
|
||
return create_saved_test(db, user, data, ids)
|
||
query = filtered_bank_query(db, user, data.category_ids, data.state,
|
||
data.difficulty, data.article_ids, data.tag_ids, data.system_ids)
|
||
if data.explicit_ids:
|
||
query = query.filter(Question.id.in_(list(dict.fromkeys(data.explicit_ids))))
|
||
ids = [row[0] for row in query.with_entities(Question.id).all()]
|
||
if data.expected_count is not None and data.expected_count != len(ids):
|
||
raise HTTPException(409, "Available count changed. Refresh the count and try again")
|
||
if len(ids) < data.count:
|
||
raise HTTPException(400, f"Only {len(ids)} questions available; requested {data.count}")
|
||
return create_saved_test(db, user, data, random.sample(ids, data.count))
|
||
|
||
|
||
def _blueprint_test(db, user, data):
|
||
"""A paper shaped like the objective the learner is studying for.
|
||
|
||
Forty questions drawn evenly across a bank is forty coin flips; the same
|
||
forty drawn to the board's published weights is a rehearsal. What each
|
||
domain was owed and what it could give is returned with the test, so a
|
||
thin corner of the bank is visible rather than quietly changing the shape.
|
||
"""
|
||
from app.services import exam_blueprint
|
||
|
||
exam_id = getattr(user, "active_exam_id", None)
|
||
if not exam_id:
|
||
raise HTTPException(400, "Choose a study objective first — a blueprint belongs to an exam")
|
||
if not exam_blueprint.domains(db, exam_id):
|
||
raise HTTPException(400, "That objective has no blueprint yet")
|
||
|
||
ids, report = exam_blueprint.sample(
|
||
db, exam_id, data.count, predicate=bank_question_predicate(user) & general_question_predicate())
|
||
if not ids:
|
||
raise HTTPException(400, "No questions match this objective's blueprint")
|
||
quiz = create_saved_test(db, user, data, ids)
|
||
# Attached rather than merged, so a caller that does not know about
|
||
# blueprints is unaffected.
|
||
try:
|
||
quiz.blueprint_report = report
|
||
except Exception: # pragma: no cover — a plain schema object
|
||
pass
|
||
return quiz
|