Skills › Scientific Agent Skills › clinical-reports › check_deidentification.py
check_deidentification.py
304 lines · skills/clinical-reports/scripts/check_deidentification.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate de-identification process documentation without scanning patient text."""
4from __future__ import annotations
6import argparse
7import sys
8from typing import Any
10sys.dont_write_bytecode = True
12from _common import ( # noqa: E402
13 ValidationError,
14 error_report,
15 load_json_object,
16 require_bool,
17 require_data_class,
18 require_exact_keys,
19 require_string,
20 write_json_report,
21)
23TOOL = "check_deidentification"
24SAFE_HARBOR_KEYS = (
25 "names",
26 "geographic_subdivisions",
27 "dates_and_ages",
28 "telephone_numbers",
29 "fax_numbers",
30 "email_addresses",
31 "social_security_numbers",
32 "medical_record_numbers",
33 "health_plan_numbers",
34 "account_numbers",
35 "certificate_and_license_numbers",
36 "vehicle_identifiers",
37 "device_identifiers",
38 "urls",
39 "ip_addresses",
40 "biometric_identifiers",
41 "full_face_images",
42 "other_unique_characteristics_or_codes",
43)
44RESIDUAL_KEYS = (
45 "free_text",
46 "small_cells_and_rare_cases",
47 "images_and_metadata",
48 "linked_data_and_quasi_identifiers",
49)
50METHODS = {
51 "safe_harbor",
52 "expert_determination",
53 "not_applicable_synthetic_or_aggregate",
54}
55CLEAR_STATUSES = {"cleared_by_authorized_reviewer", "not_present_by_design"}
56RESIDUAL_CLEAR_STATUSES = {
57 "reviewed_no_unresolved_issue",
58 "not_applicable_with_rationale",
59}
60TOP_LEVEL_FIELDS = {
61 "schema_version",
62 "artifact_kind",
63 "process_status",
64 "safety_notice",
65 "data_scope",
66 "authorized_purpose",
67 "authorization_verified",
68 "local_only_handling_confirmed",
69 "minimum_necessary_reviewed",
70 "method",
71 "safe_harbor_identifiers",
72 "actual_knowledge_review",
73 "expert_determination",
74 "synthetic_or_aggregate_rationale",
75 "residual_risk_review",
76 "review",
77}
80def _required_true(
81 data: dict[str, Any],
82 field: str,
83 errors: list[str],
84) -> None:
85 try:
86 if not require_bool(data.get(field), field):
87 errors.append(f"{field} must be true")
88 except ValidationError as exc:
89 errors.append(str(exc))
92def validate_process(data: dict[str, Any]) -> dict[str, Any]:
93 """Validate recorded process gates without asserting de-identification."""
94 errors: list[str] = []
95 warnings: list[str] = []
96 try:
97 require_exact_keys(data, TOP_LEVEL_FIELDS, "checklist")
98 if data.get("schema_version") != "2.0":
99 raise ValidationError("schema_version must be 2.0")
100 if data.get("process_status") != "BLOCKED_NOT_ASSESSED":
101 raise ValidationError("process_status must remain BLOCKED_NOT_ASSESSED")
102 require_string(data.get("safety_notice"), "safety_notice", max_length=1000)
103 require_exact_keys(
104 data.get("safe_harbor_identifiers"),
105 SAFE_HARBOR_KEYS,
106 "safe_harbor_identifiers",
107 )
108 require_exact_keys(
109 data.get("actual_knowledge_review"),
110 {"completed_by_authorized_privacy_reviewer", "record_reference"},
111 "actual_knowledge_review",
112 )
113 require_exact_keys(
114 data.get("expert_determination"),
115 {
116 "completed_by_qualified_expert",
117 "expert_documentation_reference",
118 "anticipated_recipient_and_conditions_documented",
119 },
120 "expert_determination",
121 )
122 require_exact_keys(
123 data.get("synthetic_or_aggregate_rationale"),
124 {"origin_verified", "record_reference"},
125 "synthetic_or_aggregate_rationale",
126 )
127 require_exact_keys(
128 data.get("residual_risk_review"),
129 RESIDUAL_KEYS,
130 "residual_risk_review",
131 )
132 require_exact_keys(
133 data.get("review"),
134 {
135 "privacy_legal_review",
136 "institutional_release_review",
137 "release_authorized",
138 },
139 "review",
140 )
141 except ValidationError as exc:
142 errors.append(str(exc))
143 if data.get("artifact_kind") != "deidentification_process_checklist":
144 errors.append("artifact_kind must be deidentification_process_checklist")
145 try:
146 data_scope = require_data_class(data.get("data_scope"))
147 require_string(data.get("authorized_purpose"), "authorized_purpose")
148 except ValidationError as exc:
149 errors.append(str(exc))
150 data_scope = ""
152 for field in (
153 "authorization_verified",
154 "local_only_handling_confirmed",
155 "minimum_necessary_reviewed",
156 ):
157 _required_true(data, field, errors)
159 method = data.get("method")
160 if not isinstance(method, str) or method not in METHODS:
161 errors.append(f"method must be one of {sorted(METHODS)}")
162 elif method == "safe_harbor":
163 statuses = data.get("safe_harbor_identifiers")
164 if not isinstance(statuses, dict):
165 errors.append("safe_harbor_identifiers must be an object")
166 else:
167 missing = sorted(set(SAFE_HARBOR_KEYS) - set(statuses))
168 extra = sorted(set(statuses) - set(SAFE_HARBOR_KEYS))
169 if missing:
170 errors.append(f"missing Safe Harbor categories: {missing}")
171 if extra:
172 errors.append(f"unexpected Safe Harbor categories: {extra}")
173 for key in SAFE_HARBOR_KEYS:
174 if (
175 not isinstance(statuses.get(key), str)
176 or statuses.get(key) not in CLEAR_STATUSES
177 ):
178 errors.append(f"safe_harbor_identifiers.{key} is not cleared")
179 actual = data.get("actual_knowledge_review")
180 if not isinstance(actual, dict):
181 errors.append("actual_knowledge_review must be an object")
182 else:
183 try:
184 if not require_bool(
185 actual.get("completed_by_authorized_privacy_reviewer"),
186 "actual_knowledge_review.completed_by_authorized_privacy_reviewer",
187 ):
188 errors.append("authorized actual-knowledge review is required")
189 require_string(
190 actual.get("record_reference"),
191 "actual_knowledge_review.record_reference",
192 )
193 except ValidationError as exc:
194 errors.append(str(exc))
195 elif method == "expert_determination":
196 expert = data.get("expert_determination")
197 if not isinstance(expert, dict):
198 errors.append("expert_determination must be an object")
199 else:
200 try:
201 if not require_bool(
202 expert.get("completed_by_qualified_expert"),
203 "expert_determination.completed_by_qualified_expert",
204 ):
205 errors.append("qualified Expert Determination is required")
206 require_string(
207 expert.get("expert_documentation_reference"),
208 "expert_determination.expert_documentation_reference",
209 )
210 if not require_bool(
211 expert.get("anticipated_recipient_and_conditions_documented"),
212 "expert_determination.anticipated_recipient_and_conditions_documented",
213 ):
214 errors.append("anticipated recipient and conditions must be documented")
215 except ValidationError as exc:
216 errors.append(str(exc))
217 elif method == "not_applicable_synthetic_or_aggregate":
218 if data_scope not in {"synthetic", "aggregate"}:
219 errors.append(
220 "not_applicable_synthetic_or_aggregate requires synthetic or aggregate scope"
221 )
222 rationale = data.get("synthetic_or_aggregate_rationale")
223 if not isinstance(rationale, dict):
224 errors.append("synthetic_or_aggregate_rationale must be an object")
225 else:
226 try:
227 if not require_bool(
228 rationale.get("origin_verified"),
229 "synthetic_or_aggregate_rationale.origin_verified",
230 ):
231 errors.append("synthetic or aggregate origin must be verified")
232 require_string(
233 rationale.get("record_reference"),
234 "synthetic_or_aggregate_rationale.record_reference",
235 )
236 except ValidationError as exc:
237 errors.append(str(exc))
239 residual = data.get("residual_risk_review")
240 if not isinstance(residual, dict):
241 errors.append("residual_risk_review must be an object")
242 else:
243 for key in RESIDUAL_KEYS:
244 if (
245 not isinstance(residual.get(key), str)
246 or residual.get(key) not in RESIDUAL_CLEAR_STATUSES
247 ):
248 errors.append(f"residual_risk_review.{key} is unresolved")
250 review = data.get("review")
251 if not isinstance(review, dict):
252 errors.append("review must be an object")
253 else:
254 for field in ("privacy_legal_review", "institutional_release_review"):
255 if review.get(field) != "completed":
256 warnings.append(f"review.{field} remains required")
257 if review.get("release_authorized") is not False:
258 errors.append("review.release_authorized must remain false")
260 return {
261 "tool": TOOL,
262 "status": "BLOCKED" if errors else "PROCESS_DOCUMENTED_REVIEW_REQUIRED",
263 "method": method,
264 "errors": errors,
265 "warnings": warnings,
266 "limitations": [
267 "No patient text or files were scanned.",
268 "Does not establish de-identification, no-actual-knowledge, Expert Determination, or HIPAA compliance.",
269 ],
270 "review_required": True,
271 "authorizes_release_or_submission": False,
272 }
275def build_parser() -> argparse.ArgumentParser:
276 parser = argparse.ArgumentParser(
277 description=(
278 "Check bounded JSON de-identification process documentation. "
279 "Never labels data HIPAA-compliant or guaranteed de-identified."
280 )
281 )
282 parser.add_argument("input_file", help="Local process checklist (.json)")
283 parser.add_argument("-o", "--output", help="Optional local JSON report")
284 parser.add_argument("--overwrite", action="store_true")
285 return parser
288def main(argv: list[str] | None = None) -> int:
289 args = build_parser().parse_args(argv)
290 try:
291 _, data = load_json_object(args.input_file)
292 report = validate_process(data)
293 except (OSError, ValidationError) as exc:
294 report = error_report(TOOL, exc)
295 try:
296 write_json_report(report, args.output, overwrite=args.overwrite)
297 except (OSError, ValidationError) as exc:
298 print(str(exc), file=sys.stderr)
299 return 2
300 return 0 if report["status"] == "PROCESS_DOCUMENTED_REVIEW_REQUIRED" else 1
303if __name__ == "__main__":
304 raise SystemExit(main())