Skills › Scientific Agent Skills › treatment-plans › privacy_process_check.py
privacy_process_check.py
213 lines · skills/treatment-plans/scripts/privacy_process_check.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check documented privacy process without determining compliance."""
4from __future__ import annotations
6import argparse
7import sys
9from _common import (
10 Issue,
11 ValidationError,
12 direct_identifier_key_paths,
13 error_report,
14 load_package,
15 print_report,
16 report_payload,
17 validate_package_structure,
18)
21def check_privacy_process(documents: dict[str, dict]) -> dict:
22 issues = validate_package_structure(documents)
23 if issues:
24 return report_payload(
25 "privacy_process",
26 issues,
27 counts={"structural_issues": len(issues)},
28 )
30 handoff = documents["intended_use_handoff_record"]
31 classification = handoff["data_classification"]
32 privacy = handoff["privacy_process"]
34 for document_type, document in documents.items():
35 if document["data_classification"] != classification:
36 issues.append(
37 Issue(
38 "DATA_CLASSIFICATION_MISMATCH",
39 f"{document_type}:$.data_classification",
40 )
41 )
42 if document["subject_ref"] != handoff["subject_ref"]:
43 issues.append(
44 Issue(
45 "SUBJECT_REFERENCE_MISMATCH",
46 f"{document_type}:$.subject_ref",
47 )
48 )
49 for path in direct_identifier_key_paths(document):
50 issues.append(
51 Issue(
52 "DIRECT_IDENTIFIER_FIELD_NAME_PROHIBITED",
53 f"{document_type}:{path}",
54 )
55 )
57 if privacy["data_classification"] != classification:
58 issues.append(
59 Issue(
60 "PRIVACY_CLASSIFICATION_MISMATCH",
61 (
62 "intended_use_handoff_record:"
63 "$.privacy_process.data_classification"
64 ),
65 )
66 )
68 required_true = (
69 "local_authorization_confirmed",
70 "minimum_necessary_confirmed",
71 "no_external_tools_confirmed",
72 "no_prompt_log_example_copy_confirmed",
73 )
74 for field in required_true:
75 if privacy[field] is not True:
76 issues.append(
77 Issue(
78 "PRIVACY_ATTESTATION_REQUIRED",
79 f"intended_use_handoff_record:$.privacy_process.{field}",
80 )
81 )
83 for field in (
84 "authorized_environment_reference",
85 "retention_policy_reference",
86 ):
87 if not privacy[field].strip():
88 issues.append(
89 Issue(
90 "PRIVACY_REFERENCE_REQUIRED",
91 f"intended_use_handoff_record:$.privacy_process.{field}",
92 )
93 )
95 if privacy["data_disposition_status"] == "pending":
96 issues.append(
97 Issue(
98 "DATA_DISPOSITION_NOT_DOCUMENTED",
99 (
100 "intended_use_handoff_record:"
101 "$.privacy_process.data_disposition_status"
102 ),
103 )
104 )
106 if classification == "synthetic":
107 if not handoff["subject_ref"].startswith("SYNTHETIC-"):
108 issues.append(
109 Issue(
110 "SYNTHETIC_REFERENCE_NOT_MARKED",
111 "intended_use_handoff_record:$.subject_ref",
112 )
113 )
114 if privacy["privacy_review_status"] not in {
115 "not_applicable_synthetic",
116 "completed_by_qualified_reviewer",
117 }:
118 issues.append(
119 Issue(
120 "SYNTHETIC_PRIVACY_STATUS_PENDING",
121 (
122 "intended_use_handoff_record:"
123 "$.privacy_process.privacy_review_status"
124 ),
125 )
126 )
127 else:
128 if (
129 privacy["privacy_review_status"]
130 != "completed_by_qualified_reviewer"
131 ):
132 issues.append(
133 Issue(
134 "QUALIFIED_PRIVACY_REVIEW_REQUIRED",
135 (
136 "intended_use_handoff_record:"
137 "$.privacy_process.privacy_review_status"
138 ),
139 )
140 )
141 if not privacy["privacy_reviewer_role"].strip():
142 issues.append(
143 Issue(
144 "PRIVACY_REVIEWER_ROLE_REQUIRED",
145 (
146 "intended_use_handoff_record:"
147 "$.privacy_process.privacy_reviewer_role"
148 ),
149 )
150 )
151 if privacy["reviewed_at"] is None:
152 issues.append(
153 Issue(
154 "PRIVACY_REVIEW_TIME_REQUIRED",
155 (
156 "intended_use_handoff_record:"
157 "$.privacy_process.reviewed_at"
158 ),
159 )
160 )
161 issues.append(
162 Issue(
163 "PATIENT_DERIVED_FREE_TEXT_REQUIRES_LOCAL_REVIEW",
164 "$",
165 "warning",
166 )
167 )
169 issues.sort(key=lambda item: (item.path, item.code))
170 return report_payload(
171 "privacy_process",
172 issues,
173 counts={
174 "documents_checked": len(documents),
175 "direct_identifier_field_paths": sum(
176 len(direct_identifier_key_paths(document))
177 for document in documents.values()
178 ),
179 },
180 extra={
181 "deidentification_determined": False,
182 "hipaa_compliance_determined": False,
183 "external_processing_authorized": False,
184 "values_scanned_or_echoed": False,
185 },
186 )
189def build_parser() -> argparse.ArgumentParser:
190 parser = argparse.ArgumentParser(
191 description=(
192 "Check local authorization, minimization, no-external-tool, "
193 "qualified-review, retention, and direct-field declarations. "
194 "This is not a de-identification or compliance determination."
195 )
196 )
197 parser.add_argument("package", help="Complete local package directory")
198 return parser
201def main() -> int:
202 args = build_parser().parse_args()
203 try:
204 documents, _ = load_package(args.package)
205 report = check_privacy_process(documents)
206 except ValidationError as exc:
207 report = error_report("privacy_process", exc)
208 print_report(report)
209 return 1 if report["status"] == "fail" else 0
212if __name__ == "__main__":
213 sys.exit(main())