Skills › Scientific Agent Skills › iso-standards-readiness › check_capa.py
check_capa.py
240 lines · skills/iso-standards-readiness/scripts/check_capa.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check CAPA record structure, evidence, and effectiveness gates."""
4from __future__ import annotations
6from typing import Any
8from _common import (
9 Review,
10 finish,
11 guarded_main,
12 load_json,
13 require_root_object,
14 standard_parser,
15)
17CAPA_STATUSES = {
18 "open",
19 "investigating",
20 "actions-in-progress",
21 "effectiveness-pending",
22 "closed",
23 "cancelled",
24}
25EFFECTIVENESS_RESULTS = {"pending", "effective", "ineffective"}
26RISK_DECISIONS = {"escalate", "control-and-monitor", "no-escalation"}
29def _text_array(
30 review: Review,
31 value: Any,
32 path: str,
33 *,
34 min_items: int = 1,
35) -> list[str]:
36 raw = review.list(value, path, min_items=min_items)
37 if raw is None:
38 return []
39 result: list[str] = []
40 for index, item in enumerate(raw):
41 if not isinstance(item, str) or not item.strip():
42 review.add(
43 "TEXT_REQUIRED",
44 f"{path}[{index}]",
45 "requires non-placeholder text",
46 )
47 else:
48 result.append(item.strip())
49 return result
52def validate(data: dict[str, Any]) -> tuple[Review, dict[str, int]]:
53 review = Review()
54 metadata = review.object(data.get("metadata"), "metadata")
55 if metadata is not None:
56 review.controlled_item(metadata, "metadata", require_approved=True)
57 review.text(metadata, "register_id", "metadata", max_chars=120)
58 review.date(metadata, "review_date", "metadata")
60 capas = review.list(data.get("capas"), "capas", min_items=1) or []
61 closed = 0
62 for index, value in enumerate(capas):
63 path = f"capas[{index}]"
64 capa = review.object(value, path)
65 if capa is None:
66 continue
67 review.text(capa, "id", path, max_chars=120)
68 review.text(capa, "owner", path, max_chars=200)
69 status = review.choice(capa, "status", CAPA_STATUSES, path)
70 review.text(capa, "source_event", path, max_chars=1_000)
71 review.text(capa, "problem_statement", path, max_chars=2_000)
72 review.text(capa, "scope", path, max_chars=2_000)
73 review.text(capa, "correction_or_containment", path, max_chars=2_000)
74 review.evidence(capa, path)
75 review.source_refs(capa, path)
76 review.approval(capa, path, require_approved=status in {"closed", "cancelled"})
78 risk_path = f"{path}.risk_assessment"
79 risk = review.object(capa.get("risk_assessment"), risk_path)
80 if risk is not None:
81 review.text(risk, "patient_and_user_impact", risk_path, max_chars=2_000)
82 review.text(risk, "product_and_process_impact", risk_path, max_chars=2_000)
83 review.text(risk, "reportability_review", risk_path, max_chars=2_000)
84 review.choice(risk, "decision", RISK_DECISIONS, risk_path)
85 review.text(risk, "decision_rationale", risk_path, max_chars=2_000)
86 review.text(risk, "owner", risk_path, max_chars=200)
87 review.evidence(risk, risk_path)
88 review.approval(risk, risk_path, require_approved=True)
90 investigation_path = f"{path}.investigation"
91 investigation = review.object(capa.get("investigation"), investigation_path)
92 if investigation is not None:
93 review.text(investigation, "method", investigation_path, max_chars=500)
94 review.text(
95 investigation,
96 "root_cause_or_justified_conclusion",
97 investigation_path,
98 max_chars=3_000,
99 )
100 review.text(
101 investigation,
102 "systemic_extent_review",
103 investigation_path,
104 max_chars=2_000,
105 )
106 review.text(investigation, "owner", investigation_path, max_chars=200)
107 review.evidence(investigation, investigation_path)
108 review.approval(
109 investigation,
110 investigation_path,
111 require_approved=status not in {"open", "investigating"},
112 )
114 actions = review.list(capa.get("actions"), f"{path}.actions", min_items=1) or []
115 all_implemented = True
116 for action_index, action_value in enumerate(actions):
117 action_path = f"{path}.actions[{action_index}]"
118 action = review.object(action_value, action_path)
119 if action is None:
120 all_implemented = False
121 continue
122 review.text(action, "id", action_path, max_chars=120)
123 review.text(action, "description", action_path, max_chars=2_000)
124 review.text(action, "owner", action_path, max_chars=200)
125 review.date(action, "due_date", action_path)
126 implemented = action.get("implemented_date")
127 if implemented in (None, ""):
128 all_implemented = False
129 if status in {"effectiveness-pending", "closed"}:
130 review.add(
131 "ACTION_NOT_IMPLEMENTED",
132 f"{action_path}.implemented_date",
133 "implemented date is required at this CAPA status",
134 )
135 else:
136 review.date(action, "implemented_date", action_path)
137 review.evidence(action, action_path)
138 review.approval(action, action_path, require_approved=True)
139 review.unique_ids(actions, f"{path}.actions")
141 _text_array(
142 review,
143 capa.get("change_control_refs"),
144 f"{path}.change_control_refs",
145 min_items=1,
146 )
148 effect_path = f"{path}.effectiveness"
149 effectiveness = review.object(capa.get("effectiveness"), effect_path)
150 result = None
151 if effectiveness is not None:
152 review.text(effectiveness, "plan", effect_path, max_chars=2_000)
153 review.text(
154 effectiveness,
155 "objective_acceptance_criteria",
156 effect_path,
157 max_chars=2_000,
158 )
159 review.text(effectiveness, "baseline", effect_path, max_chars=1_000)
160 review.text(
161 effectiveness,
162 "sample_or_observation_window",
163 effect_path,
164 max_chars=1_000,
165 )
166 review.date(effectiveness, "due_date", effect_path)
167 review.text(effectiveness, "owner", effect_path, max_chars=200)
168 review.text(
169 effectiveness,
170 "independent_reviewer",
171 effect_path,
172 max_chars=200,
173 )
174 result = review.choice(
175 effectiveness,
176 "result",
177 EFFECTIVENESS_RESULTS,
178 effect_path,
179 )
180 if result != "pending":
181 review.text(
182 effectiveness,
183 "conclusion",
184 effect_path,
185 max_chars=2_000,
186 )
187 review.date(effectiveness, "review_date", effect_path)
188 review.evidence(effectiveness, effect_path)
189 review.approval(
190 effectiveness,
191 effect_path,
192 require_approved=True,
193 )
195 if status == "closed":
196 closed += 1
197 if not all_implemented:
198 review.add(
199 "CLOSURE_BLOCKED",
200 path,
201 "closed CAPA has unimplemented actions",
202 "blocker",
203 )
204 if result != "effective":
205 review.add(
206 "CLOSURE_BLOCKED",
207 f"{effect_path}.result",
208 "closed CAPA requires approved effective result",
209 "blocker",
210 )
211 review.date(capa, "closure_date", path)
212 review.text(capa, "closure_summary", path, max_chars=2_000)
213 elif result == "ineffective":
214 review.add(
215 "REOPEN_REQUIRED",
216 f"{effect_path}.result",
217 "ineffective action must not be closed and needs documented follow-up",
218 "blocker",
219 )
221 if status == "cancelled":
222 review.text(capa, "cancellation_rationale", path, max_chars=2_000)
224 review.unique_ids(capas, "capas")
225 return review, {"capas": len(capas), "closed_capas": closed}
228def main() -> int:
229 parser = standard_parser(
230 "Check CAPA records and fail closed on missing effectiveness evidence.",
231 "Path to the local CAPA JSON file",
232 )
233 args = parser.parse_args()
234 data = require_root_object(load_json(args.input))
235 review, metrics = validate(data)
236 return finish("check_capa", review, args, metrics=metrics)
239if __name__ == "__main__":
240 guarded_main(main)