pdf-quiz-generator/backend/scripts/seed_abp_blueprint.py

"""Seed the ABP General Pediatrics content outline as a blueprint.

Source: the American Board of Pediatrics, "General Pediatrics Content Outline",
effective 15 October 2024 (minor revisions December 2024). Only the structure
is taken — domain numbering, headings and the published exam weights, which are
the board's statement of what a paper contains. No exam material is reproduced.

The mapping below is the part that is ours. The board's outline is arranged for
examining and our taxonomy is arranged for studying, and neither should be bent
to fit the other, so this says which of our categories feed each domain. It is
a first pass by name; an administrator edits it from the exam screen, and
anything left unmapped is reported rather than hidden.

    docker compose cp backend/scripts/seed_abp_blueprint.py backend:/app/seed.py
    docker compose exec backend python /app/seed.py --exam pediatrics-boards [--apply]
"""
import argparse
import json
import os
import sys
from pathlib import Path

sys.path.insert(0, "/app")

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.exam import BlueprintCategoryLink, Exam, ExamBlueprint
from app.models.question_category import QuestionCategory

OUTLINE = Path(__file__).with_name("abp-2024-outline.json")

#: ABP domain code -> the root categories that feed it, by name. Everything
#: beneath a named category counts, so only the roots are listed.
MAPPING: dict[str, list[str]] = {
    "1": ["Primary Care & Prevention"],
    "2": ["Neonatology"],
    "3": ["Adolescent Medicine"],
    "4": ["Infectious Disease", "Sepsis"],
    "5": ["Psychiatry & Psychology", "Developmental & Behavioral"],
    # Poverty, family structure, school, grief. No root of ours holds these;
    # left for an administrator rather than forced into a poor fit.
    "6": [],
    "7": ["Emergency Medicine", "Critical Care", "Toxicology"],
    # Nor this one. Reported, not hidden.
    "8": [],
    "9": ["Orthopedics & Sports"],
    "10": ["Ophthalmology", "Otolaryngology"],
    "11": ["Cardiology"],
    "12": ["Pulmonology"],
    "13": ["Gastroenterology & Nutrition"],
    "14": ["Neurology"],
    "15": ["Dermatology"],
    "16": ["Hematology-Oncology"],
    "17": ["Allergy/Immunology"],
    "18": ["Endocrinology"],
    "19": ["Nephrology & Urology"],
    # The board splits genitourinary from nephrology; we do not, so both point
    # at the same root and the weights add up to the pair's real share.
    "20": ["Nephrology & Urology", "Surgery"],
    "21": ["Genetics & Metabolism"],
    "22": ["Rheumatology"],
    "23": ["Professional Topics"],
    "24": ["Professional Topics"],
}


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--exam", default="pediatrics-boards")
    parser.add_argument("--apply", action="store_true", help="write; otherwise report only")
    args = parser.parse_args()

    engine = create_engine(os.environ["DATABASE_URL"])
    db = sessionmaker(bind=engine)()

    exam = db.query(Exam).filter(Exam.slug == args.exam).first()
    if not exam:
        print(f"No exam with slug {args.exam!r}")
        return 1

    outline = json.loads(OUTLINE.read_text())
    roots = {c.name: c.id for c in db.query(QuestionCategory).filter(
        QuestionCategory.parent_id.is_(None)).all()}

    unknown = sorted({name for names in MAPPING.values() for name in names} - set(roots))
    if unknown:
        print("Mapping names no such root category:", ", ".join(unknown))
        return 1

    existing = {row.code: row for row in db.query(ExamBlueprint).filter(
        ExamBlueprint.exam_id == exam.id).all()}

    written = mapped = 0
    unmapped = []
    for order, domain in enumerate(outline):
        code = domain["code"]
        line = existing.get(code)
        if line is None:
            line = ExamBlueprint(exam_id=exam.id, code=code, title=domain["title"],
                                 weight=domain["weight"], sort_order=order)
            if args.apply:
                db.add(line)
                db.flush()
            written += 1
        else:
            line.title, line.weight, line.sort_order = domain["title"], domain["weight"], order

        for sub_order, sub in enumerate(domain["subdomains"]):
            if sub["code"] in existing:
                continue
            if args.apply:
                db.add(ExamBlueprint(exam_id=exam.id, parent_id=line.id, code=sub["code"],
                                     title=sub["title"], weight=None, sort_order=sub_order))
            written += 1

        names = MAPPING.get(code, [])
        if not names:
            unmapped.append(f"{code}. {domain['title']} ({domain['weight']}%)")
            continue
        if args.apply and line.id:
            have = {row.category_id for row in db.query(BlueprintCategoryLink).filter(
                BlueprintCategoryLink.blueprint_id == line.id).all()}
            for name in names:
                if roots[name] not in have:
                    db.add(BlueprintCategoryLink(blueprint_id=line.id, category_id=roots[name]))
        mapped += 1

    if args.apply:
        db.commit()

    print(f"{'Wrote' if args.apply else 'Would write'} {written} blueprint lines "
          f"for {exam.name}; {mapped} of {len(outline)} domains mapped to categories.")
    if unmapped:
        print("\nNo category maps to these — an administrator should choose, or they")
        print("will contribute nothing to a weighted paper:")
        for row in unmapped:
            print("  ", row)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())