"""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 collections import Counter 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+)?") #: The "3." that opens an ordered-list item. It is a position in a list, not a #: fact, and it necessarily changes when a list is split across sections — so #: it is stripped before the numbers are compared. Requiring it to survive made #: the first pass keep the original numbering, which left sections opening on #: "2." and "4." because the list they came from had been divided. LIST_MARKER = re.compile(r"^(\s*)\d+([.)])(\s)", re.M) def facts_only(text: str) -> str: """The text with ordered-list numbering removed, for comparing numbers.""" return LIST_MARKER.sub(r"\1\2\3", text) def renumber(text: str) -> str: """Number every ordered-list item in this section from 1. A list split across two sections leaves the second starting wherever the first stopped — "2. Blood cultures" opening the Diagnosis section. Counting is per indentation level, so a nested list is numbered independently of its parent, and nothing else resets it: a blank line between items is a loose list, not a new one, and prose in between does not restart the count. """ counters: dict[int, int] = {} out = [] for line in text.split("\n"): match = re.match(r"^(\s*)(\d+)([.)])(\s.*)$", line) if not match: out.append(line) continue indent, _was, dot, rest = match.groups() depth = len(indent) for deeper in [level for level in counters if level > depth]: del counters[deeper] counters[depth] = counters.get(depth, 0) + 1 out.append(f"{indent}{counters[depth]}{dot}{rest}") return "\n".join(out) 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 _multiset_problem(what: str, before: list[str], after: list[str]) -> list[str]: """Describe how two multisets differ, counting repeats. Comparing sets alone reported "lost -, added -" when a token simply appeared a different number of times — which reads as nothing having changed, on a line explaining a refusal. """ want, got = Counter(before), Counter(after) if want == got: return [] lost = {k: want[k] - got.get(k, 0) for k in want if want[k] > got.get(k, 0)} gained = {k: got[k] - want.get(k, 0) for k in got if got[k] > want.get(k, 0)} parts = [] if lost: parts.append("lost " + ", ".join(f"{k}x{n}" for k, n in sorted(lost.items())[:6])) if gained: parts.append("duplicated " + ", ".join(f"{k}x{n}" for k, n in sorted(gained.items())[:6])) return [f"{what} changed ({'; '.join(parts)})"] 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"] problems += _multiset_problem("cross-references", MARKER.findall(old_text), MARKER.findall(new_text)) problems += _multiset_problem("numbers", NUMBER.findall(facts_only(old_text)), NUMBER.findall(facts_only(new_text))) 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"], # Each section's lists start at 1, whatever they started at # in the block they were carved out of. "content": renumber(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): 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}") # A section that says nothing is worse than a missing one: the shape # promises an answer and then does not give it. Reported so an editor # can find them, not refused — a one-line prognosis is often the whole # honest answer. # "No therapy is required" is a real answer; "No protocols are # available" is the absence of one. The difference is whether the # sentence is about the medicine or about the source. placeholder = re.compile( r"^(no|none)\b[^.]{0,60}\b(available|specified|provided|documented|given)\b", re.I) thin = [(a.title, s["title"], (s.get("content") or "").strip()) for a in good for s in clinical_of(a) if placeholder.match((s.get("content") or "").strip())] if thin: print(f"\n sections with nothing in them ({len(thin)}) — for an editor:") for title, section, body in thin[:12]: print(f" {title} / {section}: {body[:60]}") 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())