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

"""Group symptom tags under the organ system they present in.

Symptoms were a flat list of ~4,300 keywords, so the picker was an alphabetical
wall. AMBOSS groups them by body region — "Head and neck > Eye > Blindness" — and
that is what makes the list navigable.

Only the symptoms that match a system's pattern are moved. Everything else stays
top-level rather than being forced into an approximate bucket, and nothing is
renamed, merged or deleted: this only sets a parent.

    docker compose exec backend python -m scripts.nest_symptoms_by_system
    docker compose exec backend python -m scripts.nest_symptoms_by_system --apply
"""
import re
import sys
from collections import defaultdict

from sqlalchemy import text as sa_text

from app.database import SessionLocal

# Only worth a branch if several symptoms land in it.
MIN_CHILDREN = 3

# Organ system -> what a symptom in it looks like. Ordered: the first match wins,
# so put the specific patterns above the general ones.
SYSTEM_PATTERNS = [
    ("Nervous System & Special Senses",
     r"seizure|headache|migraine|ataxia|paralysis|palsy|neuropath|hypotoni|hypertoni|"
     r"tremor|dizz|vertigo|syncope|coma|conscious|meningism|blind|\bvision\b|visual|"
     r"strabismus|nystagmus|deaf|hearing|tinnitus|aphasia|dysarthr|weakness|numbness"),
    ("Cardiovascular System",
     r"murmur|palpitation|chest pain|cyanosis|hypertens|hypotens|tachycardi|bradycardi|"
     r"edema|oedema|shock|syncope|pulse|perfusion"),
    ("Respiratory System",
     r"cough|wheez|stridor|dyspnea|dyspnoea|tachypnea|apnea|apnoea|respiratory distress|"
     r"hemoptysis|haemoptysis|hypox|retraction|grunting"),
    ("Gastrointestinal System",
     r"vomit|diarrhea|diarrhoea|constipation|abdominal pain|abdominal distention|"
     r"abdominal distension|jaundice|hematemesis|melena|dysphagia|reflux|colic|"
     r"hepatomegal|splenomegal|ascites|encopresis"),
    ("Renal & Urinary System",
     r"hematuria|haematuria|proteinuria|dysuria|oliguria|polyuria|enuresis|"
     r"urinary frequency|urinary retention|flank pain"),
    ("Musculoskeletal System",
     r"limp|joint pain|arthralgia|arthritis|back pain|scoliosis|fracture|"
     r"limb pain|gait|swelling of the joint|muscle pain|myalgia"),
    ("Skin & Subcutaneous Tissue",
     r"rash|pruritus|itch|urticaria|petechia|purpura|blister|vesicle|pustule|"
     r"alopecia|hair loss|pallor|erythema|lesion"),
    ("Blood & Lymphoreticular System",
     r"anemia|anaemia|bleeding|bruis|lymphadenopath|pallor|petechia|thrombocytopen"),
    ("Endocrine System",
     r"polydipsia|polyuria|weight loss|weight gain|obesity|short stature|tall stature|"
     r"precocious|delayed puberty|goiter|goitre|hypoglycem|hyperglycem"),
    ("Behavioral Health",
     r"anxiety|depress|irritab|hyperactiv|inattent|aggress|self.harm|suicid|"
     r"insomnia|sleep disturbance|tantrum|behaviou?r"),
    ("Immune System",
     r"allerg|anaphylax|recurrent infection|immunodefic|angioedema"),
    ("Human Development",
     r"developmental delay|failure to thrive|growth failure|milestone|regression|"
     r"feeding difficult|poor feeding"),
    ("Multisystem Processes & Disorders",
     r"\bfever\b|fatigue|lethargy|malaise|sepsis|weight loss|dehydration|"
     r"failure to thrive|night sweats"),
]


def main():
    apply_changes = "--apply" in sys.argv
    db = SessionLocal()
    try:
        systems = {row[1].lower(): row[0] for row in db.execute(sa_text(
            "SELECT id, name FROM question_tags WHERE type = 'system'")).fetchall()}
        if not systems:
            print("  No 'system' tags yet; run the migration first.")
            return 1

        symptoms = db.execute(sa_text("""
            SELECT t.id, t.name, COUNT(l.question_id) AS uses
            FROM question_tags t
            LEFT JOIN question_tag_links l ON l.tag_id = t.id
            WHERE t.type = 'keyword' AND t.parent_id IS NULL
            GROUP BY t.id, t.name
            ORDER BY uses DESC, t.name
        """)).fetchall()
        print(f"  ungrouped symptoms: {len(symptoms)}")

        compiled = [(name, re.compile(pattern, re.I)) for name, pattern in SYSTEM_PATTERNS]
        planned: dict[str, list[tuple[int, str, int]]] = defaultdict(list)
        for tag_id, name, uses in symptoms:
            for system_name, matcher in compiled:
                if system_name.lower() in systems and matcher.search(name):
                    planned[system_name].append((tag_id, name, uses))
                    break

        # A branch holding one or two entries is noise, not navigation.
        planned = {k: v for k, v in planned.items() if len(v) >= MIN_CHILDREN}

        if apply_changes:
            for system_name, members in planned.items():
                parent = systems[system_name.lower()]
                for tag_id, _name, _uses in members:
                    db.execute(sa_text("UPDATE question_tags SET parent_id = :p WHERE id = :i"),
                               {"p": parent, "i": tag_id})
            db.commit()

        moved = sum(len(v) for v in planned.values())
        print("APPLIED" if apply_changes else "DRY RUN")
        print(f"  grouped under a system : {moved}")
        print(f"  left top-level         : {len(symptoms) - moved}\n")
        for system_name in sorted(planned, key=lambda k: -len(planned[k])):
            members = planned[system_name]
            shown = ", ".join(name for _i, name, _u in members[:4])
            more = f" … +{len(members) - 4}" if len(members) > 4 else ""
            print(f"  {system_name} ({len(members)}): {shown}{more}")
        if not apply_changes:
            print("\n  Re-run with --apply to group them.")
    finally:
        db.close()


if __name__ == "__main__":
    sys.exit(main())