feat: mdm_pass — give the clinical view the shape of a decision

The Short and Long views have a structure; the Clinical view has none.
413 sections across 323 articles carry 242 different titles, most of
them one free-form block called 'Management' holding everything from the
presenting complaint to discharge advice.

The shape is medical decision making: Clinical paths, Diagnosis,
Management, and Prognosis and outcome where it adds something.

This pass moves text between sections, so nothing can be checked section
by section the way the prose pass was. Every check is over the whole
article's clinical view at once — the same cross-references, the same
numbers, the same overall length — plus the shape itself: known titles,
in order, none missing, none twice. Content may be reordered and resplit
freely; it may not appear or vanish.

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 13:11:30 +02:00
parent 44af549dbc
commit db2bbf0638

236
backend/scripts/mdm_pass.py Normal file
View file

@ -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())