Spaces:
Running
Running
File size: 4,797 Bytes
d4d951a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """Generate the Scribe and Interpreter term artifacts from shared source data."""
from __future__ import annotations
import argparse
import csv
import io
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / "shared" / "carepath_shared" / "terms" / "medical_terms.json"
SCRIBE_TARGET = ROOT / "data" / "medical_lexicon.json"
INTERPRETER_TARGET = ROOT / "interpreter" / "app" / "glossary" / "data" / "seed_glossary.csv"
def _terms() -> list[dict[str, object]]:
payload = json.loads(SOURCE.read_text(encoding="utf-8"))
terms = payload.get("terms")
if not isinstance(terms, list):
raise ValueError(f"{SOURCE} must contain a terms list")
return terms
def _target(term: dict[str, object], name: str) -> dict[str, object] | None:
targets = term.get("targets")
if not isinstance(targets, dict):
raise ValueError(f"{term.get('term_en')!r} has no target mapping")
target = targets.get(name)
if target is None:
return None
if not isinstance(target, dict):
raise ValueError(f"{term.get('term_en')!r} has an invalid {name} mapping")
return target
def _ordered_terms(terms: list[dict[str, object]], name: str) -> list[tuple[dict[str, object], dict[str, object]]]:
selected = [(term, target) for term in terms if (target := _target(term, name)) is not None]
orders = [target.get("order") for _, target in selected]
if sorted(orders) != list(range(len(selected))):
raise ValueError(f"{name} target orders must be contiguous and unique")
return sorted(selected, key=lambda item: int(item[1]["order"]))
def _scribe_artifact(terms: list[dict[str, object]]) -> str:
lines = ["{", ' "terms": [']
rows = []
for term, target in _ordered_terms(terms, "scribe"):
aliases = term.get("aliases")
if (
not isinstance(term.get("term_en"), str)
or not isinstance(target.get("category"), str)
or not isinstance(target.get("term_vi"), str)
or not isinstance(aliases, list)
or not all(isinstance(alias, str) for alias in aliases)
):
raise ValueError(f"{term.get('term_en')!r} has invalid Scribe fields")
rows.append((term["term_en"], target["category"], target["term_vi"], aliases))
for index, (term_en, category, term_vi, aliases) in enumerate(rows):
comma = "," if index + 1 < len(rows) else ""
lines.extend(
(
" {",
f' "term": {json.dumps(term_en, ensure_ascii=False)},',
f' "category": {json.dumps(category, ensure_ascii=False)},',
f' "vietnamese": {json.dumps(term_vi, ensure_ascii=False)},',
f' "aliases": {json.dumps(aliases, ensure_ascii=False)}',
f" }}{comma}",
)
)
return "\n".join([*lines, " ]", "}", "", ""])
def _interpreter_artifact(terms: list[dict[str, object]]) -> str:
output = io.StringIO(newline="")
writer = csv.DictWriter(
output,
fieldnames=["term_vi", "term_en", "kind", "lasa_group"],
lineterminator="\n",
)
writer.writeheader()
for term, _target_data in _ordered_terms(terms, "interpreter"):
flags = term.get("risk_flags")
if (
not all(isinstance(term.get(field), str) for field in ("term_vi", "term_en", "kind"))
or not isinstance(flags, dict)
or not isinstance(flags.get("lasa_group", ""), str)
):
raise ValueError(f"{term.get('term_en')!r} has invalid risk flags")
writer.writerow(
{
"term_vi": term["term_vi"],
"term_en": term["term_en"],
"kind": term["kind"],
"lasa_group": flags.get("lasa_group", ""),
}
)
return output.getvalue()
def artifacts() -> dict[Path, str]:
terms = _terms()
return {SCRIBE_TARGET: _scribe_artifact(terms), INTERPRETER_TARGET: _interpreter_artifact(terms)}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--check", action="store_true", help="fail if generated artifacts differ")
args = parser.parse_args()
generated = artifacts()
changed = [path for path, content in generated.items() if path.read_text(encoding="utf-8") != content]
if args.check:
if changed:
raise SystemExit("Generated term artifacts differ: " + ", ".join(map(str, changed)))
print("Generated term artifacts are current.")
return
for path, content in generated.items():
path.write_text(content, encoding="utf-8")
print(f"Wrote {path.relative_to(ROOT)}")
if __name__ == "__main__":
main()
|