Skills › Scientific Agent Skills › clinical-reports › validate_trial_report.py
validate_trial_report.py
534 lines · skills/clinical-reports/scripts/validate_trial_report.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate structured ICH E3, CONSORT 2025, or SPIRIT 2025 coverage."""
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_identifier,
20 require_string,
21 write_json_report,
22)
24TOOL = "validate_trial_report"
25ALLOWED_STATUSES = {
26 "verified_present",
27 "not_applicable_with_rationale",
28 "missing",
29 "conflict",
30}
31E3_SECTIONS = (
32 "title_page",
33 "synopsis",
34 "table_of_contents",
35 "abbreviations_and_definitions",
36 "ethics",
37 "investigators_and_administration",
38 "introduction",
39 "study_objectives",
40 "investigational_plan",
41 "study_patients",
42 "efficacy_evaluation",
43 "safety_evaluation",
44 "discussion_and_conclusions",
45 "tables_figures_graphs_not_in_text",
46 "reference_list",
47 "appendices",
48)
49COMMON_TOP_FIELDS = {
50 "schema_version",
51 "artifact_kind",
52 "draft_status",
53 "safety_notice",
54 "data_classification",
55 "authorized_purpose",
56 "authorization_verified",
57 "guidance",
58 "provenance_manifest",
59 "review",
60}
61ROUTES = {
62 "clinical_study_report_draft": {
63 "container": "sections",
64 "keys": E3_SECTIONS,
65 "guidance": "ICH E3 Step 4 1995-11-30 with Q&A R1 2012-07-06",
66 "review_fields": (
67 "medical_review",
68 "statistical_review",
69 "safety_review",
70 "privacy_legal_review",
71 "quality_review",
72 "regulatory_review",
73 ),
74 "authorization_field": "submission_authorized",
75 "draft_status": "BLOCKED_INCOMPLETE_DRAFT_NOT_FOR_FILING_OR_SUBMISSION",
76 "top_fields": COMMON_TOP_FIELDS | {"study_metadata", "sections"},
77 "metadata_name": "study_metadata",
78 "item_fields": {"status", "source_fact_ids", "rationale"},
79 },
80 "randomized_trial_results_draft": {
81 "container": "checklist_items",
82 "keys": tuple(f"C{number:02d}" for number in range(1, 31)),
83 "guidance": "CONSORT 2025",
84 "review_fields": (
85 "accountable_author_review",
86 "methodologist_review",
87 "statistical_review",
88 "safety_review",
89 ),
90 "authorization_field": "publication_authorized",
91 "draft_status": (
92 "BLOCKED_INCOMPLETE_DRAFT_NOT_FOR_PUBLICATION_OR_SUBMISSION"
93 ),
94 "top_fields": COMMON_TOP_FIELDS
95 | {
96 "study_metadata",
97 "checklist_items",
98 "not_applicable_rationales",
99 "participant_flow_source_fact_ids",
100 },
101 "metadata_name": "study_metadata",
102 "item_fields": {"status", "source_fact_ids", "official_item_locator"},
103 },
104 "randomized_trial_protocol_reporting_manifest": {
105 "container": "checklist_items",
106 "keys": tuple(f"S{number:02d}" for number in range(1, 35)),
107 "guidance": "SPIRIT 2025",
108 "review_fields": (
109 "investigator_sponsor_review",
110 "methodologist_review",
111 "statistical_review",
112 "safety_review",
113 "ethics_regulatory_review",
114 ),
115 "authorization_field": "protocol_approved",
116 "draft_status": (
117 "BLOCKED_INCOMPLETE_REPORTING_CHECKLIST_NOT_A_PROTOCOL_OR_ETHICS_APPROVAL"
118 ),
119 "top_fields": COMMON_TOP_FIELDS
120 | {
121 "protocol_metadata",
122 "checklist_items",
123 "not_applicable_rationales",
124 "participant_timeline_source_fact_ids",
125 },
126 "metadata_name": "protocol_metadata",
127 "item_fields": {"status", "source_fact_ids", "official_item_locator"},
128 },
129}
132def _validate_guidance(
133 artifact_kind: str,
134 guidance: Any,
135 errors: list[str],
136) -> None:
137 if not isinstance(guidance, dict):
138 errors.append("guidance must be an object")
139 return
140 if artifact_kind == "clinical_study_report_draft":
141 try:
142 guidance = require_exact_keys(
143 guidance,
144 {
145 "base",
146 "base_version",
147 "qa_version",
148 "gcp_version_considered",
149 "regional_adoption_verified",
150 },
151 "guidance",
152 )
153 except ValidationError as exc:
154 errors.append(str(exc))
155 return
156 if guidance.get("base") != "ICH E3":
157 errors.append("CSR guidance.base must be ICH E3")
158 if guidance.get("base_version") != "Step 4 1995-11-30":
159 errors.append("CSR base_version is not the supported ICH E3 version")
160 if guidance.get("qa_version") != "R1 2012-07-06":
161 errors.append("CSR qa_version is not the supported E3 Q&A version")
162 if (
163 guidance.get("gcp_version_considered")
164 != "ICH E6(R3) consolidated 2026-06-16"
165 ):
166 errors.append("CSR must record consideration of current ICH E6(R3)")
167 if guidance.get("regional_adoption_verified") is not True:
168 errors.append("CSR regional adoption/version review is required")
169 elif artifact_kind == "randomized_trial_results_draft":
170 try:
171 guidance = require_exact_keys(
172 guidance,
173 {
174 "base",
175 "item_count",
176 "statement_and_explanation_checked",
177 "applicable_extensions",
178 "applicable_extensions_reviewed",
179 "extension_conflicts_resolved_by_methodologist",
180 },
181 "guidance",
182 )
183 except ValidationError as exc:
184 errors.append(str(exc))
185 return
186 if guidance.get("base") != "CONSORT 2025" or guidance.get("item_count") != 30:
187 errors.append("results manifest must declare CONSORT 2025 with 30 items")
188 if guidance.get("statement_and_explanation_checked") is not True:
189 errors.append("CONSORT statement and explanation must be checked")
190 if guidance.get("applicable_extensions_reviewed") is not True:
191 errors.append("applicable CONSORT extensions must be reviewed")
192 extensions = guidance.get("applicable_extensions")
193 if not isinstance(extensions, list):
194 errors.append("guidance.applicable_extensions must be an array")
195 else:
196 for index, extension in enumerate(extensions):
197 try:
198 require_string(
199 extension,
200 f"guidance.applicable_extensions[{index}]",
201 max_length=200,
202 )
203 except ValidationError as exc:
204 errors.append(str(exc))
205 try:
206 conflicts_resolved = require_bool(
207 guidance.get("extension_conflicts_resolved_by_methodologist"),
208 "guidance.extension_conflicts_resolved_by_methodologist",
209 )
210 if extensions and not conflicts_resolved:
211 errors.append("extension conflicts require methodologist resolution")
212 except ValidationError as exc:
213 errors.append(str(exc))
214 elif artifact_kind == "randomized_trial_protocol_reporting_manifest":
215 try:
216 guidance = require_exact_keys(
217 guidance,
218 {
219 "base",
220 "item_count",
221 "statement_and_explanation_checked",
222 "applicable_extensions",
223 "applicable_extensions_reviewed",
224 "extension_conflicts_resolved_by_methodologist",
225 },
226 "guidance",
227 )
228 except ValidationError as exc:
229 errors.append(str(exc))
230 return
231 if guidance.get("base") != "SPIRIT 2025" or guidance.get("item_count") != 34:
232 errors.append("protocol manifest must declare SPIRIT 2025 with 34 items")
233 if guidance.get("statement_and_explanation_checked") is not True:
234 errors.append("SPIRIT statement and explanation must be checked")
235 if guidance.get("applicable_extensions_reviewed") is not True:
236 errors.append("applicable SPIRIT extensions must be reviewed")
237 extensions = guidance.get("applicable_extensions")
238 if not isinstance(extensions, list):
239 errors.append("guidance.applicable_extensions must be an array")
240 else:
241 for index, extension in enumerate(extensions):
242 try:
243 require_string(
244 extension,
245 f"guidance.applicable_extensions[{index}]",
246 max_length=200,
247 )
248 except ValidationError as exc:
249 errors.append(str(exc))
250 try:
251 conflicts_resolved = require_bool(
252 guidance.get("extension_conflicts_resolved_by_methodologist"),
253 "guidance.extension_conflicts_resolved_by_methodologist",
254 )
255 if extensions and not conflicts_resolved:
256 errors.append("extension conflicts require methodologist resolution")
257 except ValidationError as exc:
258 errors.append(str(exc))
261def _validate_metadata(
262 artifact_kind: str,
263 value: Any,
264 errors: list[str],
265 warnings: list[str],
266) -> None:
267 if artifact_kind == "clinical_study_report_draft":
268 fields = {
269 "protocol_reference",
270 "sap_reference",
271 "data_cut_reference",
272 "analysis_output_reference",
273 "coding_dictionary_versions",
274 }
275 try:
276 metadata = require_exact_keys(value, fields, "study_metadata")
277 for field in fields - {"coding_dictionary_versions"}:
278 require_string(metadata.get(field), f"study_metadata.{field}")
279 versions = metadata.get("coding_dictionary_versions")
280 if not isinstance(versions, list) or not versions:
281 raise ValidationError(
282 "study_metadata.coding_dictionary_versions must be non-empty"
283 )
284 for index, version in enumerate(versions):
285 require_string(
286 version,
287 f"study_metadata.coding_dictionary_versions[{index}]",
288 max_length=100,
289 )
290 except ValidationError as exc:
291 errors.append(str(exc))
292 elif artifact_kind == "randomized_trial_results_draft":
293 fields = {
294 "protocol_reference",
295 "sap_reference",
296 "registry_reference",
297 "data_cut_reference",
298 "analysis_output_reference",
299 }
300 try:
301 metadata = require_exact_keys(value, fields, "study_metadata")
302 for field in fields:
303 require_string(metadata.get(field), f"study_metadata.{field}")
304 except ValidationError as exc:
305 errors.append(str(exc))
306 else:
307 fields = {
308 "authorized_protocol_reference",
309 "protocol_version",
310 "registry_reference",
311 "ethics_record_reference",
312 }
313 try:
314 metadata = require_exact_keys(value, fields, "protocol_metadata")
315 for field in fields - {"ethics_record_reference"}:
316 require_string(metadata.get(field), f"protocol_metadata.{field}")
317 if metadata.get("ethics_record_reference") is None:
318 warnings.append("protocol_metadata.ethics_record_reference remains pending")
319 else:
320 require_string(
321 metadata.get("ethics_record_reference"),
322 "protocol_metadata.ethics_record_reference",
323 )
324 except ValidationError as exc:
325 errors.append(str(exc))
328def _validated_fact_ids(value: Any, field: str) -> list[str]:
329 if not isinstance(value, list):
330 raise ValidationError(f"{field} must be an array")
331 return [require_identifier(item, field) for item in value]
334def validate_trial_manifest(data: dict[str, Any]) -> dict[str, Any]:
335 """Return structural findings without evaluating trial content."""
336 errors: list[str] = []
337 warnings: list[str] = []
338 artifact_kind = data.get("artifact_kind")
339 route = ROUTES.get(artifact_kind) if isinstance(artifact_kind, str) else None
340 if route is None:
341 return {
342 "tool": TOOL,
343 "status": "BLOCKED",
344 "errors": [f"unsupported artifact_kind: {artifact_kind!r}"],
345 "warnings": [],
346 "review_required": True,
347 "authorizes_clinical_use_or_submission": False,
348 }
350 try:
351 require_exact_keys(data, route["top_fields"], "manifest")
352 if data.get("schema_version") != "2.0":
353 raise ValidationError("schema_version must be 2.0")
354 if data.get("draft_status") != route["draft_status"]:
355 raise ValidationError(
356 "draft_status must preserve the blocked non-use/non-submission warning"
357 )
358 require_string(data.get("safety_notice"), "safety_notice", max_length=1000)
359 require_data_class(data.get("data_classification"))
360 require_string(data.get("authorized_purpose"), "authorized_purpose")
361 if not require_bool(data.get("authorization_verified"), "authorization_verified"):
362 errors.append("authorization_verified must be true")
363 require_string(data.get("provenance_manifest"), "provenance_manifest")
364 except ValidationError as exc:
365 errors.append(str(exc))
367 _validate_guidance(str(artifact_kind), data.get("guidance"), errors)
368 _validate_metadata(
369 str(artifact_kind),
370 data.get(route["metadata_name"]),
371 errors,
372 warnings,
373 )
375 container_name = str(route["container"])
376 expected_keys = tuple(route["keys"])
377 container = data.get(container_name)
378 if not isinstance(container, dict):
379 errors.append(f"{container_name} must be an object")
380 container = {}
381 missing_keys = sorted(set(expected_keys) - set(container))
382 extra_keys = sorted(set(container) - set(expected_keys))
383 if missing_keys:
384 errors.append(f"missing {container_name} keys: {missing_keys}")
385 if extra_keys:
386 errors.append(f"unexpected {container_name} keys: {extra_keys}")
388 coverage: dict[str, str] = {}
389 rationale_map = data.get("not_applicable_rationales", {})
390 if container_name == "checklist_items" and not isinstance(rationale_map, dict):
391 errors.append("not_applicable_rationales must be an object")
392 rationale_map = {}
393 for key in expected_keys:
394 item = container.get(key)
395 if not isinstance(item, dict):
396 errors.append(f"{container_name}.{key} must be an object")
397 continue
398 try:
399 item = require_exact_keys(
400 item,
401 route["item_fields"],
402 f"{container_name}.{key}",
403 )
404 except ValidationError as exc:
405 errors.append(str(exc))
406 continue
407 status = item.get("status")
408 coverage[key] = str(status)
409 if not isinstance(status, str) or status not in ALLOWED_STATUSES:
410 errors.append(f"{container_name}.{key}.status is invalid")
411 continue
412 try:
413 facts = _validated_fact_ids(
414 item.get("source_fact_ids"),
415 f"{container_name}.{key}.source_fact_ids",
416 )
417 except ValidationError as exc:
418 errors.append(str(exc))
419 facts = []
420 if status == "verified_present":
421 if not facts:
422 errors.append(f"{container_name}.{key} requires a verified source fact")
423 if container_name == "checklist_items":
424 try:
425 require_string(
426 item.get("official_item_locator"),
427 f"{container_name}.{key}.official_item_locator",
428 )
429 except ValidationError as exc:
430 errors.append(str(exc))
431 elif status == "not_applicable_with_rationale":
432 try:
433 require_string(
434 (
435 item.get("rationale")
436 if container_name == "sections"
437 else rationale_map.get(key)
438 ),
439 (
440 f"{container_name}.{key}.rationale"
441 if container_name == "sections"
442 else f"not_applicable_rationales.{key}"
443 ),
444 max_length=500,
445 )
446 except ValidationError as exc:
447 errors.append(str(exc))
448 else:
449 errors.append(f"{container_name}.{key} is {status}")
451 if artifact_kind == "randomized_trial_results_draft":
452 try:
453 flow_facts = _validated_fact_ids(
454 data.get("participant_flow_source_fact_ids"),
455 "participant_flow_source_fact_ids",
456 )
457 if not flow_facts:
458 errors.append("participant flow requires verified source facts")
459 except ValidationError as exc:
460 errors.append(str(exc))
461 elif artifact_kind == "randomized_trial_protocol_reporting_manifest":
462 try:
463 timeline_facts = _validated_fact_ids(
464 data.get("participant_timeline_source_fact_ids"),
465 "participant_timeline_source_fact_ids",
466 )
467 if not timeline_facts:
468 errors.append("participant timeline requires verified source facts")
469 except ValidationError as exc:
470 errors.append(str(exc))
472 review = data.get("review")
473 try:
474 review = require_exact_keys(
475 review,
476 {*route["review_fields"], route["authorization_field"]},
477 "review",
478 )
479 for field in route["review_fields"]:
480 if review.get(field) != "completed":
481 warnings.append(f"review.{field} remains required")
482 authorization_field = str(route["authorization_field"])
483 if review.get(authorization_field) is not False:
484 errors.append(f"review.{authorization_field} must remain false")
485 except ValidationError as exc:
486 errors.append(str(exc))
488 return {
489 "tool": TOOL,
490 "artifact_kind": artifact_kind,
491 "guidance": route["guidance"],
492 "status": "BLOCKED" if errors else "STRUCTURE_COMPLETE_REVIEW_REQUIRED",
493 "coverage": coverage,
494 "errors": errors,
495 "warnings": warnings,
496 "limitations": [
497 "Checks item keys, statuses, and fact-ID shape; does not check cross-artifact linkage.",
498 "Does not validate conduct, analyses, clinical content, compliance, or submission format.",
499 ],
500 "review_required": True,
501 "authorizes_clinical_use_or_submission": False,
502 }
505def build_parser() -> argparse.ArgumentParser:
506 parser = argparse.ArgumentParser(
507 description=(
508 "Check a bounded JSON ICH E3, CONSORT 2025, or SPIRIT 2025 "
509 "coverage manifest; never claims compliance."
510 )
511 )
512 parser.add_argument("input_file", help="Local trial-report manifest (.json)")
513 parser.add_argument("-o", "--output", help="Optional local JSON report path")
514 parser.add_argument("--overwrite", action="store_true")
515 return parser
518def main(argv: list[str] | None = None) -> int:
519 args = build_parser().parse_args(argv)
520 try:
521 _, data = load_json_object(args.input_file)
522 report = validate_trial_manifest(data)
523 except (OSError, ValidationError) as exc:
524 report = error_report(TOOL, exc)
525 try:
526 write_json_report(report, args.output, overwrite=args.overwrite)
527 except (OSError, ValidationError) as exc:
528 print(str(exc), file=sys.stderr)
529 return 2
530 return 0 if report["status"] == "STRUCTURE_COMPLETE_REVIEW_REQUIRED" else 1
533if __name__ == "__main__":
534 raise SystemExit(main())