Skills › Scientific Agent Skills › clinical-reports › terminology_validator.py
terminology_validator.py
277 lines · skills/clinical-reports/scripts/terminology_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check terminology-manifest schema and optional caller-supplied local dictionary."""
4from __future__ import annotations
6import argparse
7import re
8import sys
9from typing import Any
11sys.dont_write_bytecode = True
13from _common import ( # noqa: E402
14 ValidationError,
15 error_report,
16 load_json_object,
17 parse_iso_date,
18 require_bool,
19 require_data_class,
20 require_exact_keys,
21 require_identifier,
22 require_string,
23 write_json_report,
24)
26TOOL = "terminology_validator"
27CODE_PATTERNS = {
28 "MedDRA": re.compile(r"^\d{8}$"),
29 "LOINC": re.compile(r"^\d{1,7}-\d$"),
30 "SNOMED_CT": re.compile(r"^[1-9]\d{5,17}$"),
31 "ICD_10_CM": re.compile(r"^[A-Z][0-9A-Z]{2}(?:\.[0-9A-Z]{1,4})?$"),
32 "UCUM": re.compile(r"^[A-Za-z0-9%.\[\]{}()/*^+'_-]{1,80}$"),
33}
34CODING_STATUSES = {"verified_by_qualified_reviewer", "unverified"}
35TOP_LEVEL_FIELDS = {
36 "schema_version",
37 "artifact_kind",
38 "manifest_status",
39 "safety_notice",
40 "data_classification",
41 "authorized_purpose",
42 "authorization_verified",
43 "provenance_manifest",
44 "entries",
45}
46ENTRY_FIELDS = {
47 "system",
48 "system_uri",
49 "code",
50 "display",
51 "version",
52 "language",
53 "source_fact_id",
54 "coding_status",
55 "verified_by_role",
56 "verified_at",
57}
60def _validate_entry(entry: Any, index: int) -> tuple[str, str, str, str, list[str]]:
61 entry = require_exact_keys(entry, ENTRY_FIELDS, f"entries[{index}]")
62 system = require_string(entry.get("system"), f"entries[{index}].system", max_length=32)
63 if system not in CODE_PATTERNS:
64 raise ValidationError(
65 f"entries[{index}].system must be one of {sorted(CODE_PATTERNS)}"
66 )
67 require_string(
68 entry.get("system_uri"),
69 f"entries[{index}].system_uri",
70 max_length=200,
71 )
72 code = require_string(entry.get("code"), f"entries[{index}].code", max_length=80)
73 if not CODE_PATTERNS[system].fullmatch(code):
74 raise ValidationError(f"entries[{index}].code fails {system} syntax")
75 display = require_string(
76 entry.get("display"),
77 f"entries[{index}].display",
78 max_length=500,
79 )
80 version = require_string(
81 entry.get("version"),
82 f"entries[{index}].version",
83 max_length=80,
84 )
85 require_string(entry.get("language"), f"entries[{index}].language", max_length=32)
86 require_identifier(
87 entry.get("source_fact_id"),
88 f"entries[{index}].source_fact_id",
89 )
90 status = entry.get("coding_status")
91 if not isinstance(status, str) or status not in CODING_STATUSES:
92 raise ValidationError(f"entries[{index}].coding_status is invalid")
93 warnings: list[str] = []
94 if status == "verified_by_qualified_reviewer":
95 require_string(
96 entry.get("verified_by_role"),
97 f"entries[{index}].verified_by_role",
98 max_length=100,
99 )
100 parse_iso_date(entry.get("verified_at"), f"entries[{index}].verified_at")
101 else:
102 warnings.append(f"entries[{index}] remains unverified")
103 if system == "MedDRA" and not re.fullmatch(r"\d{1,2}\.\d", version):
104 warnings.append(f"entries[{index}] MedDRA version has an unusual format")
105 return system, version, code, display, warnings
108def _load_dictionary(raw_path: str) -> dict[tuple[str, str, str], str]:
109 _, data = load_json_object(raw_path)
110 require_exact_keys(
111 data,
112 {"schema_version", "artifact_kind", "entries"},
113 "dictionary",
114 )
115 if data.get("schema_version") != "2.0":
116 raise ValidationError("dictionary schema_version must be 2.0")
117 if data.get("artifact_kind") != "terminology_dictionary":
118 raise ValidationError("dictionary artifact_kind must be terminology_dictionary")
119 entries = data.get("entries")
120 if not isinstance(entries, list) or not entries:
121 raise ValidationError("dictionary entries must be a non-empty array")
122 if len(entries) > 20_000:
123 raise ValidationError("dictionary may contain at most 20,000 entries")
124 dictionary: dict[tuple[str, str, str], str] = {}
125 for index, entry in enumerate(entries):
126 entry = require_exact_keys(
127 entry,
128 {"system", "version", "code", "display"},
129 f"dictionary entries[{index}]",
130 )
131 system = require_string(
132 entry.get("system"),
133 f"dictionary entries[{index}].system",
134 max_length=32,
135 )
136 version = require_string(
137 entry.get("version"),
138 f"dictionary entries[{index}].version",
139 max_length=80,
140 )
141 code = require_string(
142 entry.get("code"),
143 f"dictionary entries[{index}].code",
144 max_length=80,
145 )
146 display = require_string(
147 entry.get("display"),
148 f"dictionary entries[{index}].display",
149 max_length=500,
150 )
151 key = (system, version, code)
152 if key in dictionary and dictionary[key] != display:
153 raise ValidationError(f"dictionary contains conflicting duplicate at index {index}")
154 dictionary[key] = display
155 return dictionary
158def validate_terminology_manifest(
159 data: dict[str, Any],
160 dictionary: dict[tuple[str, str, str], str] | None = None,
161) -> dict[str, Any]:
162 """Validate schema and optional exact tuple matches."""
163 errors: list[str] = []
164 warnings: list[str] = []
165 try:
166 require_exact_keys(data, TOP_LEVEL_FIELDS, "manifest")
167 if data.get("schema_version") != "2.0":
168 raise ValidationError("schema_version must be 2.0")
169 if data.get("manifest_status") != "BLOCKED_INCOMPLETE":
170 raise ValidationError("manifest_status must remain BLOCKED_INCOMPLETE")
171 require_string(data.get("safety_notice"), "safety_notice", max_length=1000)
172 require_string(data.get("authorized_purpose"), "authorized_purpose")
173 if not require_bool(data.get("authorization_verified"), "authorization_verified"):
174 raise ValidationError("authorization_verified must be true")
175 require_string(data.get("provenance_manifest"), "provenance_manifest")
176 except ValidationError as exc:
177 errors.append(str(exc))
178 if data.get("artifact_kind") != "terminology_manifest":
179 errors.append("artifact_kind must be terminology_manifest")
180 try:
181 require_data_class(data.get("data_classification"))
182 except ValidationError as exc:
183 errors.append(str(exc))
185 entries = data.get("entries")
186 if not isinstance(entries, list) or not entries:
187 errors.append("entries must be a non-empty array")
188 entries = []
189 if len(entries) > 5_000:
190 errors.append("entries may contain at most 5,000 items")
192 seen: set[tuple[str, str, str]] = set()
193 verified_count = 0
194 dictionary_match_count = 0
195 for index, entry in enumerate(entries[:5_000]):
196 try:
197 system, version, code, display, entry_warnings = _validate_entry(
198 entry,
199 index,
200 )
201 warnings.extend(entry_warnings)
202 key = (system, version, code)
203 if key in seen:
204 errors.append(f"entries[{index}] duplicates a prior system/version/code")
205 seen.add(key)
206 if entry.get("coding_status") == "verified_by_qualified_reviewer":
207 verified_count += 1
208 if dictionary is not None:
209 expected_display = dictionary.get(key)
210 if expected_display is None:
211 errors.append(f"entries[{index}] is absent from the supplied dictionary")
212 elif expected_display != display:
213 errors.append(
214 f"entries[{index}] display does not match the supplied dictionary"
215 )
216 else:
217 dictionary_match_count += 1
218 except ValidationError as exc:
219 errors.append(str(exc))
221 if dictionary is None:
222 warnings.append(
223 "No local dictionary supplied; code existence, currency, and display were not checked."
224 )
225 success_status = "SCHEMA_VALID_SYNTAX_ONLY_REVIEW_REQUIRED"
226 else:
227 success_status = "SCHEMA_AND_LOCAL_DICTIONARY_MATCH_REVIEW_REQUIRED"
229 return {
230 "tool": TOOL,
231 "status": "BLOCKED" if errors else success_status,
232 "entry_count": len(entries),
233 "qualified_reviewer_verified_count": verified_count,
234 "local_dictionary_match_count": dictionary_match_count,
235 "errors": errors,
236 "warnings": warnings,
237 "limitations": [
238 "Syntax or local-dictionary matching does not establish clinical correctness.",
239 "Caller is responsible for dictionary authority, version, license, and completeness.",
240 ],
241 "review_required": True,
242 "authorizes_clinical_use_or_submission": False,
243 }
246def build_parser() -> argparse.ArgumentParser:
247 parser = argparse.ArgumentParser(
248 description=(
249 "Check a bounded terminology JSON manifest. Optional dictionary comparison "
250 "uses a caller-supplied local JSON file and never calls a network service."
251 )
252 )
253 parser.add_argument("input_file", help="Local terminology manifest (.json)")
254 parser.add_argument("--dictionary", help="Optional authorized local dictionary (.json)")
255 parser.add_argument("-o", "--output", help="Optional local JSON report")
256 parser.add_argument("--overwrite", action="store_true")
257 return parser
260def main(argv: list[str] | None = None) -> int:
261 args = build_parser().parse_args(argv)
262 try:
263 _, data = load_json_object(args.input_file)
264 dictionary = _load_dictionary(args.dictionary) if args.dictionary else None
265 report = validate_terminology_manifest(data, dictionary)
266 except (OSError, ValidationError) as exc:
267 report = error_report(TOOL, exc)
268 try:
269 write_json_report(report, args.output, overwrite=args.overwrite)
270 except (OSError, ValidationError) as exc:
271 print(str(exc), file=sys.stderr)
272 return 2
273 return 0 if report["status"].startswith("SCHEMA_") else 1
276if __name__ == "__main__":
277 raise SystemExit(main())