Skills › Scientific Agent Skills › clinical-decision-support › deidentification_checklist.py
deidentification_checklist.py
260 lines · skills/clinical-decision-support/scripts/deidentification_checklist.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check documentation of a de-identification process without reading health data."""
4from __future__ import annotations
6import argparse
7import sys
8from typing import Any
10from _common import (
11 InputError,
12 IssueLog,
13 load_json_object,
14 print_report,
15 require_list,
16 require_nonempty_text,
17 source_ids,
18 validate_references,
19 write_json,
20)
22SAFE_HARBOR_IDS = {
23 "names",
24 "geography",
25 "dates_and_ages",
26 "telephone_numbers",
27 "fax_numbers",
28 "email_addresses",
29 "social_security_numbers",
30 "medical_record_numbers",
31 "health_plan_numbers",
32 "account_numbers",
33 "certificate_license_numbers",
34 "vehicle_identifiers",
35 "device_identifiers",
36 "web_urls",
37 "ip_addresses",
38 "biometric_identifiers",
39 "full_face_images",
40 "other_unique_identifiers",
41}
42ALLOWED_STATUS = {
43 "not_present",
44 "removed",
45 "generalized",
46 "expert_reviewed",
47 "unresolved",
48}
49PLACEHOLDER_MARKERS = ("REPLACE_", "REQUIRES_", "YYYY-MM-DD")
52def _find_placeholders(value: Any, location: str = "$") -> list[str]:
53 findings: list[str] = []
54 if isinstance(value, dict):
55 for key, nested in value.items():
56 findings.extend(_find_placeholders(nested, f"{location}.{key}"))
57 elif isinstance(value, list):
58 for index, nested in enumerate(value):
59 findings.extend(_find_placeholders(nested, f"{location}[{index}]"))
60 elif isinstance(value, str) and any(marker in value for marker in PLACEHOLDER_MARKERS):
61 findings.append(location)
62 return findings
65def check_documentation(document: dict[str, Any]) -> tuple[IssueLog, dict[str, Any]]:
66 log = IssueLog()
67 summary: dict[str, Any] = {
68 "method": None,
69 "documentation_complete": False,
70 "deidentification_determined": False,
71 "hipaa_compliance_determined": False,
72 "unresolved_categories": [],
73 }
74 try:
75 require_nonempty_text(document.get("schema_version"), "schema_version")
76 require_nonempty_text(document.get("checklist_id"), "checklist_id")
77 require_nonempty_text(document.get("title"), "title")
78 known_sources = source_ids(document)
79 validate_references(document.get("source_ids"), known_sources, "source_ids")
81 method = require_nonempty_text(document.get("method"), "method")
82 summary["method"] = method
83 if method not in {"safe_harbor", "expert_determination", "undecided"}:
84 log.errors.append("method is unsupported")
85 if method == "undecided":
86 log.errors.append("A qualified human must select and document a method")
88 boundary = document.get("data_boundary")
89 if not isinstance(boundary, dict):
90 raise InputError("data_boundary must be an object")
91 if boundary.get("raw_data_supplied") is not False:
92 log.errors.append("data_boundary.raw_data_supplied must be false")
93 if boundary.get("metadata_only") is not True:
94 log.errors.append("data_boundary.metadata_only must be true")
95 for field in ("data_context", "recipient_context", "release_context"):
96 require_nonempty_text(boundary.get(field), f"data_boundary.{field}")
98 entries = require_list(
99 document.get("identifier_categories"),
100 "identifier_categories",
101 maximum=18,
102 )
103 observed_ids: set[str] = set()
104 for index, entry in enumerate(entries):
105 field = f"identifier_categories[{index}]"
106 if not isinstance(entry, dict):
107 raise InputError(f"{field} must be an object")
108 identifier = require_nonempty_text(entry.get("id"), f"{field}.id")
109 if identifier not in SAFE_HARBOR_IDS:
110 log.errors.append(f"{field}.id is not a Safe Harbor category")
111 if identifier in observed_ids:
112 log.errors.append(f"Duplicate identifier category: {identifier}")
113 observed_ids.add(identifier)
114 status = require_nonempty_text(entry.get("status"), f"{field}.status")
115 if status not in ALLOWED_STATUS:
116 log.errors.append(f"{field}.status is unsupported")
117 evidence = require_nonempty_text(entry.get("evidence"), f"{field}.evidence")
118 if status == "unresolved":
119 summary["unresolved_categories"].append(identifier)
120 if evidence.lower().startswith(("example:", "value:")):
121 log.errors.append(
122 f"{field}.evidence must not contain an example identifier or value"
123 )
124 missing_ids = sorted(SAFE_HARBOR_IDS - observed_ids)
125 extra_ids = sorted(observed_ids - SAFE_HARBOR_IDS)
126 if missing_ids:
127 log.errors.append("Missing Safe Harbor categories: " + ", ".join(missing_ids))
128 if extra_ids:
129 log.errors.append("Unexpected categories: " + ", ".join(extra_ids))
131 safe_harbor = document.get("safe_harbor_review")
132 if not isinstance(safe_harbor, dict):
133 raise InputError("safe_harbor_review must be an object")
134 for field in (
135 "actual_knowledge",
136 "free_text",
137 "derived_fields",
138 "date_age_zip_rules",
139 ):
140 require_nonempty_text(safe_harbor.get(field), f"safe_harbor_review.{field}")
141 if method == "safe_harbor":
142 if summary["unresolved_categories"]:
143 log.errors.append("Safe Harbor review has unresolved identifier categories")
144 if safe_harbor.get("completed") is not True:
145 log.errors.append("safe_harbor_review.completed must be true")
146 require_nonempty_text(
147 safe_harbor.get("reviewer_role"), "safe_harbor_review.reviewer_role"
148 )
149 require_nonempty_text(
150 safe_harbor.get("review_date"), "safe_harbor_review.review_date"
151 )
153 expert = document.get("expert_determination_review")
154 if not isinstance(expert, dict):
155 raise InputError("expert_determination_review must be an object")
156 if method == "expert_determination":
157 for field in (
158 "expert_qualification_reference",
159 "method_document_reference",
160 "risk_threshold_rationale",
161 "residual_risk",
162 "validity_period",
163 "change_triggers",
164 "review_date",
165 ):
166 require_nonempty_text(
167 expert.get(field), f"expert_determination_review.{field}"
168 )
169 if expert.get("completed") is not True:
170 log.errors.append(
171 "expert_determination_review.completed must be true"
172 )
174 residual = document.get("residual_risk_review")
175 if not isinstance(residual, dict):
176 raise InputError("residual_risk_review must be an object")
177 for field in (
178 "linkage",
179 "differencing",
180 "rare_combinations",
181 "longitudinal_patterns",
182 "geography",
183 "genomics",
184 "prior_releases",
185 ):
186 require_nonempty_text(residual.get(field), f"residual_risk_review.{field}")
187 if residual.get("completed") is not True:
188 log.errors.append("residual_risk_review.completed must be true")
190 review = document.get("human_review")
191 if not isinstance(review, dict):
192 raise InputError("human_review must be an object")
193 if review.get("required") is not True:
194 log.errors.append("human_review.required must be true")
195 require_nonempty_text(review.get("role"), "human_review.role")
196 require_nonempty_text(
197 review.get("approval_boundary"), "human_review.approval_boundary"
198 )
199 if review.get("completed") is not True:
200 log.errors.append("human_review.completed must be true")
202 governance = document.get("governance")
203 if not isinstance(governance, dict):
204 raise InputError("governance must be an object")
205 for field in (
206 "version",
207 "owner",
208 "change_summary",
209 "auditability",
210 "monitoring",
211 ):
212 require_nonempty_text(governance.get(field), f"governance.{field}")
214 if document.get("compliance_claims") is not False:
215 log.errors.append("compliance_claims must be false")
216 if review.get("completed") is True:
217 placeholders = _find_placeholders(document)
218 if placeholders:
219 log.errors.append(
220 "Unresolved template placeholders remain: "
221 + ", ".join(placeholders[:10])
222 )
223 except InputError as exc:
224 log.errors.append(str(exc))
226 summary["documentation_complete"] = log.ok
227 if log.ok:
228 log.info.append(
229 "Process documentation is complete; no de-identification or compliance determination was made"
230 )
231 return log, summary
234def main() -> int:
235 parser = argparse.ArgumentParser(
236 description=(
237 "Check metadata documenting Safe Harbor or Expert Determination work. "
238 "The script never reads a dataset and never determines compliance."
239 )
240 )
241 parser.add_argument("input", help="Local checklist JSON")
242 parser.add_argument("-o", "--output", help="Optional local JSON report")
243 args = parser.parse_args()
245 try:
246 document = load_json_object(args.input)
247 log, summary = check_documentation(document)
248 report = log.as_dict()
249 report["deidentification_review"] = summary
250 if args.output:
251 write_json(args.output, report)
252 print_report(report)
253 except InputError as exc:
254 print_report(IssueLog(errors=[str(exc)]).as_dict())
255 return 2
256 return 0 if report["status"] == "pass" else 1
259if __name__ == "__main__":
260 sys.exit(main())