diff --git a/backend/app/routers/study_plans.py b/backend/app/routers/study_plans.py index 4246503..7f15314 100644 --- a/backend/app/routers/study_plans.py +++ b/backend/app/routers/study_plans.py @@ -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)): diff --git a/backend/app/services/exam_blueprint.py b/backend/app/services/exam_blueprint.py index 768b785..ce6eef4 100644 --- a/backend/app/services/exam_blueprint.py +++ b/backend/app/services/exam_blueprint.py @@ -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 diff --git a/backend/tests/test_blueprint_plans.py b/backend/tests/test_blueprint_plans.py new file mode 100644 index 0000000..d37fe2f --- /dev/null +++ b/backend/tests/test_blueprint_plans.py @@ -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()