Skills › Scientific Agent Skills › clinical-decision-support › _common.py
_common.py
227 lines · skills/clinical-decision-support/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared bounded local-file utilities for research-only CDS helpers."""
4from __future__ import annotations
6import json
7import math
8from dataclasses import dataclass, field
9from pathlib import Path
10from typing import Any, Iterable
12MAX_INPUT_BYTES = 1_000_000
13MAX_TEXT_LENGTH = 4_000
14MAX_SOURCES = 200
16PERSON_LEVEL_KEYS = {
17 "address",
18 "date_of_birth",
19 "dob",
20 "email",
21 "full_name",
22 "individual_record",
23 "individual_records",
24 "medical_record_number",
25 "mrn",
26 "note_text",
27 "person_name",
28 "patient",
29 "patient_data",
30 "patient_id",
31 "patient_record",
32 "patient_records",
33 "phone",
34 "raw_data",
35 "raw_rows",
36 "record_id",
37 "social_security_number",
38 "ssn",
39}
42class InputError(ValueError):
43 """Raised for unsafe, malformed, or out-of-bounds input."""
46@dataclass
47class IssueLog:
48 """Deterministic validation findings."""
50 errors: list[str] = field(default_factory=list)
51 warnings: list[str] = field(default_factory=list)
52 info: list[str] = field(default_factory=list)
54 @property
55 def ok(self) -> bool:
56 return not self.errors
58 def as_dict(self) -> dict[str, Any]:
59 return {
60 "status": "pass" if self.ok else "fail",
61 "errors": self.errors,
62 "warnings": self.warnings,
63 "info": self.info,
64 "disclaimer": (
65 "Not for patient care or live clinical use. "
66 "Structural checks only; not a clinical, privacy, regulatory, "
67 "legal, or compliance determination."
68 ),
69 }
72def _reject_nonlocal_path(raw: str) -> None:
73 lowered = raw.strip().lower()
74 if not lowered:
75 raise InputError("Path must not be empty")
76 if "\x00" in raw or "://" in lowered or lowered.startswith("\\\\"):
77 raise InputError("Only local filesystem paths are allowed")
80def local_input_path(raw: str, suffixes: Iterable[str] | None = None) -> Path:
81 """Resolve a bounded, regular, non-symlink local input file."""
83 _reject_nonlocal_path(raw)
84 path = Path(raw).expanduser()
85 if path.is_symlink():
86 raise InputError("Symlink inputs are not allowed")
87 resolved = path.resolve()
88 if not resolved.is_file():
89 raise InputError(f"Input is not a regular file: {path}")
90 if suffixes and resolved.suffix.lower() not in {s.lower() for s in suffixes}:
91 raise InputError(f"Unsupported input suffix: {resolved.suffix}")
92 size = resolved.stat().st_size
93 if size > MAX_INPUT_BYTES:
94 raise InputError(f"Input exceeds {MAX_INPUT_BYTES} bytes")
95 return resolved
98def local_output_path(raw: str, suffixes: Iterable[str] | None = None) -> Path:
99 """Resolve a local output whose parent already exists."""
101 _reject_nonlocal_path(raw)
102 path = Path(raw).expanduser()
103 if path.is_symlink():
104 raise InputError("Symlink outputs are not allowed")
105 resolved = path.resolve()
106 if not resolved.parent.is_dir():
107 raise InputError("Output parent directory must already exist")
108 if suffixes and resolved.suffix.lower() not in {s.lower() for s in suffixes}:
109 raise InputError(f"Unsupported output suffix: {resolved.suffix}")
110 return resolved
113def load_json_object(raw_path: str) -> dict[str, Any]:
114 """Load one bounded UTF-8 JSON object."""
116 path = local_input_path(raw_path, {".json"})
117 try:
118 value = json.loads(path.read_text(encoding="utf-8"))
119 if not isinstance(value, dict):
120 raise InputError("Top-level JSON value must be an object")
121 ensure_no_person_level_keys(value)
122 except (UnicodeDecodeError, ValueError, RecursionError) as exc:
123 raise InputError(f"Invalid UTF-8 JSON: {exc}") from exc
124 return value
127def ensure_no_person_level_keys(value: Any, location: str = "$") -> None:
128 """Reject common person-level fields anywhere in a JSON document."""
130 if isinstance(value, dict):
131 for key, nested in value.items():
132 normalized = str(key).strip().lower()
133 if normalized in PERSON_LEVEL_KEYS:
134 raise InputError(
135 f"Person-level or raw-data key is prohibited at {location}.{key}"
136 )
137 ensure_no_person_level_keys(nested, f"{location}.{key}")
138 elif isinstance(value, list):
139 for index, nested in enumerate(value):
140 ensure_no_person_level_keys(nested, f"{location}[{index}]")
143def write_json(raw_path: str, payload: dict[str, Any]) -> None:
144 path = local_output_path(raw_path, {".json"})
145 text = json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
146 path.write_text(text, encoding="utf-8")
149def write_text(raw_path: str, text: str, suffixes: Iterable[str]) -> None:
150 path = local_output_path(raw_path, suffixes)
151 path.write_text(text, encoding="utf-8")
154def require_nonempty_text(
155 value: Any, field_name: str, *, max_length: int = MAX_TEXT_LENGTH
156) -> str:
157 if not isinstance(value, str) or not value.strip():
158 raise InputError(f"{field_name} must be non-empty text")
159 text = value.strip()
160 if len(text) > max_length:
161 raise InputError(f"{field_name} exceeds {max_length} characters")
162 return text
165def require_list(value: Any, field_name: str, *, maximum: int) -> list[Any]:
166 if not isinstance(value, list):
167 raise InputError(f"{field_name} must be a list")
168 if len(value) > maximum:
169 raise InputError(f"{field_name} exceeds {maximum} entries")
170 return value
173def finite_number(value: Any, field_name: str) -> float:
174 if isinstance(value, bool) or not isinstance(value, (int, float)):
175 raise InputError(f"{field_name} must be numeric")
176 try:
177 result = float(value)
178 except OverflowError as exc:
179 raise InputError(f"{field_name} must be representable as a finite number") from exc
180 if not math.isfinite(result):
181 raise InputError(f"{field_name} must be finite")
182 return result
185def nonnegative_int(value: Any, field_name: str) -> int:
186 if isinstance(value, bool) or not isinstance(value, int) or value < 0:
187 raise InputError(f"{field_name} must be a non-negative integer")
188 return value
191def source_ids(document: dict[str, Any]) -> set[str]:
192 sources = require_list(document.get("sources"), "sources", maximum=MAX_SOURCES)
193 identifiers: set[str] = set()
194 for index, source in enumerate(sources):
195 if not isinstance(source, dict):
196 raise InputError(f"sources[{index}] must be an object")
197 identifier = require_nonempty_text(
198 source.get("id"), f"sources[{index}].id", max_length=100
199 )
200 require_nonempty_text(
201 source.get("citation"), f"sources[{index}].citation", max_length=1_000
202 )
203 if identifier in identifiers:
204 raise InputError(f"Duplicate source id: {identifier}")
205 identifiers.add(identifier)
206 return identifiers
209def validate_references(
210 references: Any, known_sources: set[str], field_name: str
211) -> list[str]:
212 values = require_list(references, field_name, maximum=50)
213 normalized: list[str] = []
214 for index, value in enumerate(values):
215 identifier = require_nonempty_text(
216 value, f"{field_name}[{index}]", max_length=100
217 )
218 if identifier not in known_sources:
219 raise InputError(f"{field_name} references unknown source id: {identifier}")
220 normalized.append(identifier)
221 if not normalized:
222 raise InputError(f"{field_name} must cite at least one source")
223 return normalized
226def print_report(report: dict[str, Any]) -> None:
227 print(json.dumps(report, indent=2, sort_keys=True, ensure_ascii=False))