diff --git a/backend/scripts/mdm_pass.py b/backend/scripts/mdm_pass.py new file mode 100644 index 0000000..47cc380 --- /dev/null +++ b/backend/scripts/mdm_pass.py @@ -0,0 +1,236 @@ +"""Give the clinical view a shape: the order a decision is actually made in. + +The Short and Long views have a structure — In short; Definition, Epidemiology, +Etiology, Clinical features, Diagnostics, Treatment. The Clinical view has none. +413 sections across 323 articles carry 242 different titles, most of them one +free-form block called "Management" or "Approach at the bedside" holding +everything from the presenting complaint to discharge advice. + +The shape is medical decision making: what walks in and where it can branch, +how it is confirmed, what is done, and how it ends when that is worth saying. + + Clinical paths the presentations, and the branch points they lead to + Diagnosis how it is confirmed, and what is excluded + Management what is done, in the order it is done + Prognosis and outcome only where it adds something + +Unlike the prose pass, this moves text *between* sections, so nothing can be +checked section by section. Every check is over the whole article's clinical +view at once: the same cross-references, the same numbers, the same overall +length. Content may be reordered and resplit freely; it may not appear or +vanish. + + python -m scripts.mdm_pass export --out /tmp/mdm --batches 6 + python -m scripts.mdm_pass import --in /tmp/mdm [--apply] + python -m scripts.mdm_pass check +""" +import argparse +import json +import re +import sys +import uuid +from pathlib import Path + +from sqlalchemy.orm.attributes import flag_modified + +from app.database import SessionLocal +from app.models.article import Article + +#: The clinical view's sections, in the order a decision is made. +MDM_SECTIONS = ["Clinical paths", "Diagnosis", "Management", "Prognosis and outcome"] +#: Only the last is optional; an article with no paths or no management has not +#: been restructured, it has been truncated. +REQUIRED = MDM_SECTIONS[:3] + +MARKER = re.compile(r"\[\[(\d+)\|") +NUMBER = re.compile(r"\d+(?:\.\d+)?") +MIN_RATIO, MAX_RATIO = 0.75, 1.35 + + +def slugify(title: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") + + +def clinical_of(article) -> list[dict]: + return [s for s in (article.sections or []) if (s.get("variant") or "long") == "clinical"] + + +def conforms(article) -> bool: + """Whether this article's clinical view is already in MDM shape.""" + titles = [s.get("title") for s in clinical_of(article)] + if not titles: + return False + if any(title not in MDM_SECTIONS for title in titles): + return False + if titles != sorted(titles, key=MDM_SECTIONS.index): + return False + return all(required in titles for required in REQUIRED) + + +def validate(before: list[dict], after: list[dict]) -> list[str]: + """Reasons this restructure must not be applied. Empty means it is safe.""" + problems = [] + titles = [s.get("title") for s in after] + if not titles: + return ["no sections"] + unknown = [t for t in titles if t not in MDM_SECTIONS] + if unknown: + problems.append(f"section titles outside the shape: {unknown}") + if len(set(titles)) != len(titles): + problems.append("a section title used twice") + known = [t for t in titles if t in MDM_SECTIONS] + if known != sorted(known, key=MDM_SECTIONS.index): + problems.append(f"sections out of order: {known}") + missing = [r for r in REQUIRED if r not in titles] + if missing: + problems.append(f"missing: {missing}") + + old_text = "\n".join(s.get("content") or "" for s in before) + new_text = "\n".join(s.get("content") or "" for s in after) + if not new_text.strip(): + return ["empty"] + + want, got = sorted(MARKER.findall(old_text)), sorted(MARKER.findall(new_text)) + if want != got: + problems.append(f"cross-references changed (lost {sorted(set(want) - set(got)) or '-'}, " + f"added {sorted(set(got) - set(want)) or '-'})") + want_n, got_n = sorted(NUMBER.findall(old_text)), sorted(NUMBER.findall(new_text)) + if want_n != got_n: + problems.append(f"numbers changed (lost {sorted(set(want_n) - set(got_n)) or '-'}, " + f"added {sorted(set(got_n) - set(want_n)) or '-'})") + ratio = len(new_text) / max(len(old_text), 1) + if not MIN_RATIO <= ratio <= MAX_RATIO: + problems.append(f"length {ratio:.2f}x of the original") + return problems + + +def cmd_export(args) -> int: + out = Path(args.out) + out.mkdir(parents=True, exist_ok=True) + db = SessionLocal() + try: + rows = [] + for article in db.query(Article).order_by(Article.id).all(): + sections = clinical_of(article) + if not sections or conforms(article): + continue + rows.append({ + "article_id": article.id, + "article_title": article.title, + "sections": [{"title": s.get("title"), "content": s.get("content") or ""} + for s in sections], + }) + finally: + db.close() + if not rows: + print("Every clinical view is already in shape.") + return 0 + + per = max(1, -(-len(rows) // args.batches)) + for index in range(args.batches): + chunk = rows[index * per:(index + 1) * per] + if not chunk: + break + path = out / f"mdm-{index + 1:02d}.json" + path.write_text(json.dumps(chunk, indent=2, ensure_ascii=False)) + words = sum(len(s["content"].split()) for row in chunk for s in row["sections"]) + print(f" {path} {len(chunk):3d} articles, ~{words} words") + print(f"\n{len(rows)} articles to restructure.") + print("Rewrite each article's `sections` into the shape and save as mdm-NN.done.json.") + return 0 + + +def cmd_import(args) -> int: + folder = Path(args.folder) + done = sorted(folder.glob("*.done.json")) + if not done: + print(f"No *.done.json in {folder}") + return 1 + + db = SessionLocal() + try: + articles = {a.id: a for a in db.query(Article).all()} + applied = skipped = 0 + for path in done: + for entry in json.loads(path.read_text()): + article = articles.get(entry["article_id"]) + if not article: + continue + before = clinical_of(article) + after = entry["sections"] + problems = validate(before, after) + if problems: + print(f" ! {entry['article_title']}: {'; '.join(problems)}") + skipped += 1 + continue + + # Keep every other variant exactly as it is; replace the + # clinical view wholesale, since its sections are now different + # objects rather than edits of the old ones. + kept = [s for s in (article.sections or []) if (s.get("variant") or "long") != "clinical"] + rebuilt = [{ + "id": uuid.uuid4().hex, + "slug": slugify(section["title"]), + "title": section["title"], + "content": section["content"], + "parent_id": None, + "variant": "clinical", + } for section in after] + article.sections = kept + rebuilt + flag_modified(article, "sections") + applied += 1 + print(f"\n applied: {applied} refused: {skipped}") + if args.apply: + db.commit() + print(" committed.") + else: + print("\n Re-run with --apply to write.") + finally: + db.close() + return 0 + + +def cmd_check(args) -> int: + db = SessionLocal() + try: + articles = db.query(Article).all() + with_clinical = [a for a in articles if clinical_of(a)] + good = [a for a in with_clinical if conforms(a)] + print(f" articles with a clinical view : {len(with_clinical)} of {len(articles)}") + print(f" in medical-decision-making shape: {len(good)}") + print(f" still free-form : {len(with_clinical) - len(good)}") + if len(good) < len(with_clinical): + from collections import Counter + titles = Counter(s.get("title") for a in with_clinical if not conforms(a) + for s in clinical_of(a)) + print("\n commonest titles still in use:") + for title, count in titles.most_common(8): + print(f" {count:4d} {title}") + finally: + db.close() + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="cmd", required=True) + + export = sub.add_parser("export") + export.add_argument("--out", required=True) + export.add_argument("--batches", type=int, default=6) + export.set_defaults(func=cmd_export) + + imp = sub.add_parser("import") + imp.add_argument("--in", dest="folder", required=True) + imp.add_argument("--apply", action="store_true") + imp.set_defaults(func=cmd_import) + + check = sub.add_parser("check") + check.set_defaults(func=cmd_check) + + args = parser.parse_args() + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main())