9b7ab72038
Comparing marker and number *sets* reported 'lost -, added -' whenever a token merely appeared a different number of times — printed on the line explaining why the article was refused, which read as a contradiction. It counts repeats now: 'cross-references changed (duplicated 396x1)', which is what actually happened to Cystic Fibrosis. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
293 lines
11 KiB
Python
293 lines
11 KiB
Python
"""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):
|
|
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())
|