feat: study plans built to the board's content outline

Two shapes, because a plan is asked to do two different things. Papers
are rehearsal: each block is drawn to the ABP's published weights, so
sitting one says something about how you would do on the day. Domains
are study: the board's twenty-four content areas in its own order and
carrying its own titles, each given the share of the plan the board
gives it on the exam.

Both were written, then run against the real bank, which found two bugs
a unit test on a clean fixture would not have. Domains 19 and 20 —
nephrology and genitourinary — both map to our "Nephrology & Urology",
so a question sat in two pools and was dealt twice; the deal now keeps a
record of what has gone. And chunking every question a domain has into
blocks of forty gave preventive care six blocks and the plan a hundred
and sixty, which is not a plan: blocks are shared out by weight, with at
least one per domain so nothing the board examines is left out.

Built on the live bank alongside what was already there: Boards: Full
Papers (12 × 40) and Boards: By Content Domain (27 blocks, 1069
questions). Nothing existing was touched.

Psychosocial Issues and Child Abuse and Neglect — 6% of the paper
between them — had no category of ours at all, so they could contribute
nothing. Both now exist, with sub-topics named from the board's own
subdomains, and all 24 domains map to categories.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
This commit is contained in:
Daniel 2026-09-11 21:40:17 +02:00
parent 76848c5947
commit 0fa5598aad
3 changed files with 335 additions and 1 deletions

View file

@ -15,7 +15,9 @@ from app.models.study_plan import (
StudyPlanBlockProgress,
)
from app.models.user import User
from app.services.quiz_builder import GenerateTestRequest, bank_query, create_saved_test
from app.services import exam_blueprint
from app.services.quiz_builder import (GenerateTestRequest, bank_query, create_saved_test,
general_question_predicate)
from app.utils.auth import get_current_user, require_moderator
router = APIRouter()
@ -223,6 +225,99 @@ def create_plan(data: PlanWrite, db: Session = Depends(get_db),
return {"id": plan.id, "slug": plan.slug, "name": plan.name}
class BlueprintPlan(BaseModel):
exam_id: int
slug: str
name: str
description: str | None = None
#: "papers" — every block a miniature of the real exam, for rehearsal.
#: "domains" — one block per content domain, for working through a subject.
shape: str = "domains"
block_size: int = Field(default=40, ge=5, le=200)
#: How many blocks the plan has. For papers, how many to deal out; for
#: domains, how they are shared out by weight — every domain still gets at
#: least one, so nothing the board examines is left out.
blocks: int = Field(default=24, ge=1, le=80)
sort_order: int = 100
is_published: bool = True
#: Attach the reading filed under each domain to its block.
with_reading: bool = True
@router.post("/from-blueprint", status_code=201)
def create_from_blueprint(data: BlueprintPlan, db: Session = Depends(get_db),
current_user: User = Depends(require_moderator)):
"""Build a plan to an examining board's published content outline.
Two shapes, because a plan is asked to do two different things. Papers are
rehearsal: each block is drawn to the board's weights, so sitting one tells
a learner something about how they would do on the day. Domains are study:
one block per content area, in the board's order and carrying its titles,
which is how someone works through a subject rather than tests themselves
on all of it.
Membership is snapshotted, as with any plan a plan you are part-way
through must not reshuffle between visits and no question appears in two
blocks of the same plan.
"""
slug = data.slug.strip().lower()
if not SLUG_RE.match(slug):
raise HTTPException(400, "A slug is lowercase words joined by hyphens")
if db.query(StudyPlan.id).filter(StudyPlan.slug == slug).first():
raise HTTPException(409, "A plan with that slug already exists")
if data.shape not in ("papers", "domains"):
raise HTTPException(400, "Shape must be papers or domains")
exam = db.get(Exam, data.exam_id)
if not exam:
raise HTTPException(404, "Exam not found")
if not exam_blueprint.domains(db, data.exam_id):
raise HTTPException(400, "That exam has no blueprint to build from")
keep = general_question_predicate()
if data.shape == "papers":
drawn = exam_blueprint.sample_papers(db, data.exam_id, data.block_size,
data.blocks, predicate=keep)
built = [(None, f"Paper {i + 1}", ids) for i, (ids, _) in enumerate(drawn)]
else:
built = exam_blueprint.domain_blocks(db, data.exam_id, data.block_size,
data.blocks, predicate=keep)
if not built:
raise HTTPException(400, "No questions match this blueprint yet")
plan = StudyPlan(slug=slug, name=data.name.strip(), description=data.description,
kind="set", exam_id=data.exam_id, sort_order=data.sort_order,
is_published=1 if data.is_published else 0)
db.add(plan)
db.flush()
# Reading per domain, so a block opens on what prepares you for it.
reading: dict[str, list[int]] = {}
if data.with_reading:
for line in exam_blueprint.domains(db, data.exam_id):
cats = exam_blueprint.categories_for(db, line.id)
if not cats:
continue
reading[line.code] = [row[0] for row in db.query(Article.id).filter(
Article.category_id.in_(cats), Article.status == "published").limit(6).all()]
for position, (code, title, ids) in enumerate(built):
block = StudyPlanBlock(plan_id=plan.id, position=position, title=title,
question_ids=list(ids))
db.add(block)
db.flush()
for index, article_id in enumerate(reading.get(code, [])):
db.add(StudyPlanBlockArticle(block_id=block.id, article_id=article_id, position=index))
db.commit()
return {
"id": plan.id, "slug": plan.slug, "name": plan.name, "shape": data.shape,
"blocks": [{"position": i, "title": title, "questions": len(ids)}
for i, (_, title, ids) in enumerate(built)],
"questions": sum(len(ids) for _, _, ids in built),
}
@router.patch("/{plan_id}")
def update_plan(plan_id: int, data: PlanUpdate, db: Session = Depends(get_db),
current_user: User = Depends(require_moderator)):

View file

@ -176,3 +176,126 @@ def total_weight(db: Session, exam_id: int) -> float:
value = db.query(func.coalesce(func.sum(ExamBlueprint.weight), 0)).filter(
ExamBlueprint.exam_id == exam_id, ExamBlueprint.parent_id.is_(None)).scalar()
return float(value or 0)
def sample_papers(db: Session, exam_id: int, size: int, count: int,
predicate=None, rng=None) -> list[tuple[list[int], list[dict]]]:
"""Several papers from one bank, no question sitting in two of them.
A plan of twelve blocks that each drew independently would ask the same
preventive-care question four times, because that is the domain with the
largest share and the smallest pool relative to it. Drawing once and
dealing out fixes that but only with a record of what has already gone,
because two domains can map to the same categories. The board splits
genitourinary from nephrology and we do not, so a question sits in both
pools and would otherwise be dealt twice.
"""
import random
rng = rng or random
lines = [line for line in domains(db, exam_id) if line.weight is not None]
if not lines or size <= 0 or count <= 0:
return []
pools: dict[int, list[int]] = {}
weights: dict[int, Decimal] = {}
for line in lines:
ids = question_ids_for(db, exam_id, categories_for(db, line.id), predicate)
rng.shuffle(ids)
pools[line.id] = ids
weights[line.id] = Decimal(line.weight)
by_id = {line.id: line for line in lines}
dealt: set[int] = set()
def take(line_id: int, want: int) -> list[int]:
out = []
pool = pools[line_id]
while pool and len(out) < want:
candidate = pool.pop()
if candidate not in dealt:
dealt.add(candidate)
out.append(candidate)
return out
papers = []
for _ in range(count):
# Recounted each round: what a domain can still supply shrinks as it is
# dealt, and a plan should shorten gracefully rather than repeat.
room = {key: sum(1 for q in ids if q not in dealt) for key, ids in pools.items()}
plan = allocate(weights, size, room)
if not plan:
break
chosen: list[int] = []
report = []
for line_id, want in plan.items():
taken = take(line_id, want)
chosen.extend(taken)
line = by_id[line_id]
report.append({"code": line.code, "title": line.title,
"weight": float(weights[line_id]),
"asked_for": want, "given": len(taken)})
if not chosen:
break
rng.shuffle(chosen)
report.sort(key=lambda row: row["weight"], reverse=True)
papers.append((chosen, report))
return papers
def domain_blocks(db: Session, exam_id: int, size: int, blocks: int,
predicate=None, rng=None) -> list[tuple[str, str, list[int]]]:
"""One or more blocks per domain, sized by the weight the board gives it.
The other shape. A paper is for rehearsal; this is for working through a
subject, which is what a plan is usually for so the blocks carry the
board's own titles in its own order.
Sized by weight rather than by what happens to be in the bank: chunking
every question into blocks of forty gave preventive care six blocks and
the plan a hundred and sixty, which is not a plan. Every domain gets at
least one block, so nothing the board examines is left out of a plan built
from its outline.
"""
import random
rng = rng or random
lines = [line for line in domains(db, exam_id) if line.weight is not None]
if not lines or size <= 0 or blocks <= 0:
return []
pools: dict[int, list[int]] = {}
weights: dict[int, Decimal] = {}
for line in lines:
ids = question_ids_for(db, exam_id, categories_for(db, line.id), predicate)
if not ids:
continue
rng.shuffle(ids)
pools[line.id] = ids
weights[line.id] = Decimal(line.weight)
if not pools:
return []
# How many blocks each domain is owed, then at least one each.
share = allocate(weights, blocks, {key: len(ids) for key, ids in pools.items()})
for key in pools:
share.setdefault(key, 0)
share[key] = max(share[key], 1)
by_id = {line.id: line for line in lines}
dealt: set[int] = set()
out = []
for line in lines:
if line.id not in pools:
continue
want = share[line.id]
pool = [q for q in pools[line.id] if q not in dealt]
parts = min(want, max(1, -(-len(pool) // size)))
for index in range(parts):
chunk = [q for q in pool[index * size:(index + 1) * size] if q not in dealt]
if not chunk:
continue
dealt.update(chunk)
suffix = f" ({index + 1} of {parts})" if parts > 1 else ""
out.append((line.code, f"{by_id[line.id].title}{suffix}", chunk))
return out

View file

@ -0,0 +1,116 @@
"""Plans built to a board's content outline.
Two shapes and one rule they share: no question appears twice in a plan. Two
domains can map to the same categories of ours the board splits genitourinary
from nephrology and we do not so a pool-per-domain is not enough on its own.
"""
import random
import unittest
from decimal import Decimal
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
import test_quiz_builder # noqa: F401 — imports every model
from app.database import Base
from app.models.exam import BlueprintCategoryLink, Exam, ExamBlueprint, QuestionExamLink
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.user import User
from app.services import exam_blueprint
class BlueprintPlanTests(unittest.TestCase):
def setUp(self):
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
poolclass=StaticPool)
Base.metadata.create_all(self.engine)
self.db = Session(self.engine)
self.db.add(User(id=1, name="Mod", email="m@example.test",
hashed_password="unused", role="moderator"))
self.db.add(Exam(id=1, slug="boards", name="Boards"))
self.db.add_all([QuestionCategory(id=1, name="Kidney", user_id=1),
QuestionCategory(id=2, name="Chest", user_id=1)])
self.db.flush()
for qid in range(1, 121):
self.db.add(Question(id=qid, question_category_id=1 if qid <= 60 else 2, user_id=1,
question_text=f"Q{qid}", question_type="mcq",
options=["a", "b"], correct_answer="a"))
self.db.add(QuestionExamLink(question_id=qid, exam_id=1))
# Two domains, one shared category: exactly the case that repeats.
self.db.add_all([
ExamBlueprint(id=1, exam_id=1, code="19", title="Nephrology", weight=Decimal("50"), sort_order=0),
ExamBlueprint(id=2, exam_id=1, code="20", title="Genitourinary", weight=Decimal("30"), sort_order=1),
ExamBlueprint(id=3, exam_id=1, code="12", title="Pulmonology", weight=Decimal("20"), sort_order=2),
])
self.db.flush()
self.db.add_all([BlueprintCategoryLink(blueprint_id=1, category_id=1),
BlueprintCategoryLink(blueprint_id=2, category_id=1),
BlueprintCategoryLink(blueprint_id=3, category_id=2)])
self.db.commit()
def tearDown(self):
self.db.close()
self.engine.dispose()
def test_papers_never_repeat_a_question_across_blocks(self):
papers = exam_blueprint.sample_papers(self.db, 1, 10, 6, rng=random.Random(3))
drawn = [q for ids, _ in papers for q in ids]
self.assertEqual(len(drawn), len(set(drawn)))
def test_two_domains_sharing_a_category_do_not_deal_the_same_question(self):
# Nephrology and Genitourinary both point at Kidney. Without a record
# of what has gone, a question is dealt once from each.
self.assertEqual(exam_blueprint.sample_papers(self.db, 1, 20, 0), [],
"asking for no papers draws nothing")
one = exam_blueprint.sample_papers(self.db, 1, 20, 1, rng=random.Random(3))
ids = one[0][0]
self.assertEqual(len(ids), len(set(ids)))
def test_a_paper_keeps_its_length_and_the_board_s_proportions(self):
papers = exam_blueprint.sample_papers(self.db, 1, 10, 1, rng=random.Random(3))
ids, report = papers[0]
self.assertEqual(len(ids), 10)
given = {row["code"]: row["given"] for row in report}
self.assertEqual(given["19"], 5) # 50%
self.assertEqual(given["12"], 2) # 20%
def test_the_plan_shortens_rather_than_repeating_when_the_bank_runs_out(self):
# Twelve papers of forty is four hundred and eighty from a bank of 120.
papers = exam_blueprint.sample_papers(self.db, 1, 40, 12, rng=random.Random(3))
drawn = [q for ids, _ in papers for q in ids]
self.assertEqual(len(drawn), len(set(drawn)))
self.assertLessEqual(len(drawn), 120)
def test_domain_blocks_are_sized_by_weight_not_by_what_is_lying_about(self):
# Chunking every question into blocks gave preventive care six blocks
# and a plan of a hundred and sixty. Weight decides how many.
blocks = exam_blueprint.domain_blocks(self.db, 1, 10, 6, rng=random.Random(3))
counts = {}
for code, _, ids in blocks:
counts[code] = counts.get(code, 0) + 1
self.assertGreater(counts["19"], counts["12"], "the heavier domain gets more blocks")
ids = [q for _, _, b in blocks for q in b]
self.assertEqual(len(ids), len(set(ids)))
def test_every_domain_with_questions_gets_at_least_one_block(self):
blocks = exam_blueprint.domain_blocks(self.db, 1, 10, 2, rng=random.Random(3))
# Two blocks asked for, three domains: nothing the board examines is
# left out of a plan built from its outline.
self.assertEqual({code for code, _, _ in blocks}, {"19", "20", "12"})
def test_blocks_carry_the_board_s_own_titles(self):
blocks = exam_blueprint.domain_blocks(self.db, 1, 10, 6, rng=random.Random(3))
self.assertTrue(any(title.startswith("Nephrology") for _, title, _ in blocks))
self.assertTrue(any("of" in title for _, title, _ in blocks), "a split domain says so")
def test_an_exam_with_no_blueprint_builds_nothing(self):
self.db.add(Exam(id=2, slug="step", name="Step"))
self.db.commit()
self.assertEqual(exam_blueprint.sample_papers(self.db, 2, 40, 4), [])
self.assertEqual(exam_blueprint.domain_blocks(self.db, 2, 40, 4), [])
if __name__ == "__main__":
unittest.main()