Skills › Scientific Agent Skills › peer-review › select_reporting_guidelines.py
select_reporting_guidelines.py
383 lines · skills/peer-review/scripts/select_reporting_guidelines.py at 5f2bc03
1#!/usr/bin/env python3
2"""Select bundled reporting guidance and audit checklist coverage locally."""
4from __future__ import annotations
6import argparse
7import re
8from collections import Counter
9from pathlib import Path
10from typing import Any
12from _common import (
13 ValidationError,
14 error_exit,
15 issue,
16 read_csv_records,
17 read_json,
18 require_enum,
19 require_exact_keys,
20 require_identifier,
21 require_identifier_list,
22 require_object,
23 require_text,
24 require_url,
25 write_json_report,
26)
28ASSET_CATALOG = Path(__file__).resolve().parents[1] / "assets" / "reporting_guidelines.json"
29COVERAGE_FIELDS = (
30 "guideline_id",
31 "item_id",
32 "status",
33 "location",
34 "rationale",
35)
36COVERAGE_STATUSES = {
37 "reported",
38 "partly_reported",
39 "not_reported",
40 "not_applicable",
41 "not_assessed",
42}
43CATALOG_STATUSES = {"current", "legacy_current_qualified"}
44CATEGORIES = {"reporting_guideline", "domain_metadata_standard"}
45REPORT_KINDS = {"results", "protocol", "abstract", "data_release"}
46ITEM_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$")
49def load_catalog(path: Path = ASSET_CATALOG) -> dict[str, Any]:
50 payload = require_object(read_json(path), "catalog")
51 require_exact_keys(
52 payload,
53 required={"schema_version", "reviewed_on", "notice", "guidelines"},
54 context="catalog",
55 )
56 require_enum(payload["schema_version"], {"2.0"}, "catalog.schema_version")
57 require_text(payload["reviewed_on"], "catalog.reviewed_on", maximum=10)
58 require_text(payload["notice"], "catalog.notice", maximum=2_000)
59 guidelines = payload["guidelines"]
60 if not isinstance(guidelines, list) or not guidelines:
61 raise ValidationError("catalog.guidelines must be a non-empty array")
63 parsed: list[dict[str, Any]] = []
64 seen: set[str] = set()
65 for index, raw_entry in enumerate(guidelines):
66 context = f"catalog.guidelines[{index}]"
67 entry = require_object(raw_entry, context)
68 require_exact_keys(
69 entry,
70 required={
71 "id",
72 "name",
73 "version",
74 "status",
75 "category",
76 "study_types",
77 "report_kinds",
78 "required_features",
79 "domains",
80 "main_item_count",
81 "url",
82 "notes",
83 },
84 context=context,
85 )
86 guideline_id = require_identifier(entry["id"], f"{context}.id")
87 if guideline_id in seen:
88 raise ValidationError(f"catalog contains duplicate ID: {guideline_id}")
89 seen.add(guideline_id)
90 item_count = entry["main_item_count"]
91 if item_count is not None:
92 if type(item_count) is not int or not 1 <= item_count <= 200:
93 raise ValidationError(
94 f"{context}.main_item_count must be null or an integer 1-200"
95 )
96 parsed.append(
97 {
98 "id": guideline_id,
99 "name": require_text(entry["name"], f"{context}.name", maximum=200),
100 "version": require_text(
101 entry["version"], f"{context}.version", maximum=100
102 ),
103 "status": require_enum(
104 entry["status"], CATALOG_STATUSES, f"{context}.status"
105 ),
106 "category": require_enum(
107 entry["category"], CATEGORIES, f"{context}.category"
108 ),
109 "study_types": require_identifier_list(
110 entry["study_types"], f"{context}.study_types", minimum=1
111 ),
112 "report_kinds": [
113 require_enum(item, REPORT_KINDS, f"{context}.report_kinds")
114 for item in entry["report_kinds"]
115 ],
116 "required_features": require_identifier_list(
117 entry["required_features"], f"{context}.required_features"
118 ),
119 "domains": require_identifier_list(
120 entry["domains"], f"{context}.domains", minimum=1
121 ),
122 "main_item_count": item_count,
123 "url": require_url(entry["url"], f"{context}.url"),
124 "notes": require_text(
125 entry["notes"], f"{context}.notes", maximum=1_000
126 ),
127 }
128 )
129 return {
130 "schema_version": payload["schema_version"],
131 "reviewed_on": payload["reviewed_on"],
132 "notice": payload["notice"],
133 "guidelines": parsed,
134 }
137def load_profile(payload: Any) -> dict[str, Any]:
138 profile = require_object(payload, "profile")
139 require_exact_keys(
140 profile,
141 required={
142 "schema_version",
143 "profile_id",
144 "study_types",
145 "report_kind",
146 "features",
147 "domains",
148 },
149 context="profile",
150 )
151 return {
152 "schema_version": require_enum(
153 profile["schema_version"], {"2.0"}, "profile.schema_version"
154 ),
155 "profile_id": require_identifier(profile["profile_id"], "profile.profile_id"),
156 "study_types": require_identifier_list(
157 profile["study_types"], "profile.study_types", minimum=1
158 ),
159 "report_kind": require_enum(
160 profile["report_kind"], REPORT_KINDS, "profile.report_kind"
161 ),
162 "features": require_identifier_list(profile["features"], "profile.features"),
163 "domains": require_identifier_list(
164 profile["domains"], "profile.domains", minimum=1
165 ),
166 }
169def select_guidelines(
170 profile: dict[str, Any], catalog: dict[str, Any]
171) -> list[dict[str, Any]]:
172 study_types = set(profile["study_types"])
173 features = set(profile["features"])
174 domains = set(profile["domains"])
175 selected: list[dict[str, Any]] = []
176 for entry in catalog["guidelines"]:
177 if not study_types.intersection(entry["study_types"]):
178 continue
179 if profile["report_kind"] not in entry["report_kinds"]:
180 continue
181 if not set(entry["required_features"]).issubset(features):
182 continue
183 if "all" not in entry["domains"] and not domains.intersection(entry["domains"]):
184 continue
185 selected.append(entry)
186 return sorted(
187 selected,
188 key=lambda item: (
189 item["category"] != "reporting_guideline",
190 item["id"],
191 ),
192 )
195def load_coverage(raw_path: str | Path) -> list[dict[str, str]]:
196 rows = read_csv_records(
197 raw_path,
198 required_fields=COVERAGE_FIELDS,
199 allowed_fields=COVERAGE_FIELDS,
200 )
201 seen: set[tuple[str, str]] = set()
202 parsed: list[dict[str, str]] = []
203 for index, row in enumerate(rows, start=2):
204 guideline_id = require_identifier(
205 row["guideline_id"], f"coverage row {index}.guideline_id"
206 )
207 item_id = require_text(
208 row["item_id"], f"coverage row {index}.item_id", maximum=64
209 )
210 if not ITEM_ID_RE.fullmatch(item_id):
211 raise ValidationError(
212 f"coverage row {index}.item_id has an invalid identifier format"
213 )
214 key = (guideline_id, item_id)
215 if key in seen:
216 raise ValidationError(
217 f"coverage contains duplicate guideline/item pair: "
218 f"{guideline_id}/{item_id}"
219 )
220 seen.add(key)
221 status = require_enum(
222 row["status"], COVERAGE_STATUSES, f"coverage row {index}.status"
223 )
224 location = require_text(
225 row["location"],
226 f"coverage row {index}.location",
227 allow_empty=True,
228 maximum=500,
229 )
230 rationale = require_text(
231 row["rationale"],
232 f"coverage row {index}.rationale",
233 allow_empty=True,
234 maximum=1_000,
235 )
236 parsed.append(
237 {
238 "guideline_id": guideline_id,
239 "item_id": item_id,
240 "status": status,
241 "location": location,
242 "rationale": rationale,
243 }
244 )
245 return parsed
248def assess(
249 profile: dict[str, Any],
250 catalog: dict[str, Any],
251 coverage_rows: list[dict[str, str]] | None = None,
252) -> dict[str, Any]:
253 selected = select_guidelines(profile, catalog)
254 selected_by_id = {entry["id"]: entry for entry in selected}
255 errors: list[dict[str, str]] = []
256 warnings: list[dict[str, str]] = []
257 if not selected:
258 warnings.append(issue("NO_BUNDLED_GUIDELINE_MATCH", "profile.study_types"))
260 rows_by_guideline: dict[str, list[dict[str, str]]] = {}
261 if coverage_rows is not None:
262 for row in coverage_rows:
263 rows_by_guideline.setdefault(row["guideline_id"], []).append(row)
264 subject = f"{row['guideline_id']}:{row['item_id']}"
265 if row["status"] in {"reported", "partly_reported"} and not row["location"]:
266 errors.append(issue("COVERAGE_LOCATION_REQUIRED", subject))
267 if row["status"] == "not_applicable" and not row["rationale"]:
268 errors.append(issue("NOT_APPLICABLE_RATIONALE_REQUIRED", subject))
269 if row["guideline_id"] not in selected_by_id:
270 warnings.append(issue("COVERAGE_FOR_UNSELECTED_GUIDELINE", subject))
272 coverage: list[dict[str, Any]] = []
273 for entry in selected:
274 rows = rows_by_guideline.get(entry["id"], [])
275 statuses = Counter(row["status"] for row in rows)
276 observed_ids = {row["item_id"] for row in rows}
277 expected_count = entry["main_item_count"]
278 unrecorded: list[str] = []
279 unexpected: list[str] = []
280 record_complete: bool | None = None
281 if expected_count is not None:
282 expected_ids = {str(number) for number in range(1, expected_count + 1)}
283 unrecorded = sorted(
284 expected_ids - observed_ids, key=lambda value: int(value)
285 )
286 unexpected = sorted(observed_ids - expected_ids)
287 record_complete = not unrecorded and not unexpected
288 if unexpected:
289 warnings.append(
290 issue("NON_MAIN_ITEM_IDS_RECORDED", entry["id"])
291 )
292 elif rows:
293 record_complete = None
294 warnings.append(
295 issue("OFFICIAL_ITEM_SET_MUST_BE_VERIFIED", entry["id"])
296 )
298 gap_ids = sorted(
299 row["item_id"]
300 for row in rows
301 if row["status"]
302 in {"partly_reported", "not_reported", "not_assessed"}
303 )
304 coverage.append(
305 {
306 "guideline_id": entry["id"],
307 "main_item_count": expected_count,
308 "recorded_item_count": len(rows),
309 "coverage_record_complete": record_complete,
310 "status_counts": dict(sorted(statuses.items())),
311 "unrecorded_main_item_ids": unrecorded,
312 "unexpected_item_ids": unexpected,
313 "reporting_gap_item_ids": gap_ids,
314 }
315 )
317 return {
318 "schema_version": "2.0",
319 "profile_id": profile["profile_id"],
320 "valid": not errors,
321 "status": "VALID" if not errors else "INVALID_COVERAGE_RECORD",
322 "errors": errors,
323 "warnings": warnings,
324 "selected_guidelines": [
325 {
326 "id": entry["id"],
327 "name": entry["name"],
328 "version": entry["version"],
329 "status": entry["status"],
330 "category": entry["category"],
331 "url": entry["url"],
332 "notes": entry["notes"],
333 }
334 for entry in selected
335 ],
336 "coverage": coverage if coverage_rows is not None else None,
337 "catalog_reviewed_on": catalog["reviewed_on"],
338 "notice": (
339 "Guideline selection and checklist coverage concern reporting "
340 "completeness only. They are not scores and do not establish study "
341 "quality, validity, conduct, or manuscript merit. Check the official "
342 "guideline, applicable extensions, and target venue policy."
343 ),
344 }
347def build_parser() -> argparse.ArgumentParser:
348 parser = argparse.ArgumentParser(
349 description=(
350 "Select current bundled reporting guidance and optionally audit a "
351 "local checklist coverage CSV. No network calls are made."
352 )
353 )
354 parser.add_argument("profile", help="Local study profile JSON")
355 parser.add_argument(
356 "--coverage",
357 help=(
358 "Optional coverage CSV using aggregate main item IDs (for example, "
359 "1 through 30 for CONSORT 2025)"
360 ),
361 )
362 parser.add_argument("-o", "--output", help="Optional local JSON report")
363 parser.add_argument(
364 "--force", action="store_true", help="Replace an existing output file"
365 )
366 return parser
369def main() -> int:
370 args = build_parser().parse_args()
371 try:
372 profile = load_profile(read_json(args.profile))
373 catalog = load_catalog()
374 coverage = load_coverage(args.coverage) if args.coverage else None
375 report = assess(profile, catalog, coverage)
376 write_json_report(report, args.output, force=args.force)
377 return 0 if report["valid"] else 1
378 except ValidationError as exc:
379 return error_exit(exc)
382if __name__ == "__main__":
383 raise SystemExit(main())