Skills › Scientific Agent Skills › scientific-writing › validate_manifest.py
validate_manifest.py
460 lines · skills/scientific-writing/scripts/validate_manifest.py at 5f2bc03
1"""Validate bounded manuscript and source manifests without network access."""
3from __future__ import annotations
5import argparse
6import re
7from typing import Any
9from _common import (
10 InputError,
11 Issue,
12 emit_report,
13 is_nonempty_string,
14 is_iso_date,
15 is_placeholder,
16 issue,
17 read_json,
18 require_list,
19 require_object,
20 run,
21)
23TOOL = "validate_manifest"
24ID_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._-]{1,63}$")
25EVIDENCE_ID_RE = re.compile(r"^E[0-9]{3,8}$")
26SOURCE_TYPES = {
27 "journal_article",
28 "book",
29 "chapter",
30 "conference_paper",
31 "dataset",
32 "software",
33 "preprint",
34 "report",
35 "policy",
36 "guideline",
37 "registry",
38 "webpage",
39 "other",
40}
41CONFIDENTIALITY = {"public", "restricted", "sensitive", "proprietary"}
42DRAFT_STATUSES = {"draft", "internal_review", "submission_candidate", "submitted"}
43STATEMENT_STATUSES = {"not_applicable", "missing", "draft", "verified"}
44MANUSCRIPT_FIELDS = {
45 "schema_version",
46 "document_id",
47 "study_design",
48 "draft_status",
49 "submission_ready",
50 "registries",
51 "reporting_guidelines",
52 "human_verification",
53 "confidentiality_review",
54 "required_statements",
55}
56SOURCE_FIELDS = {
57 "evidence_id",
58 "source_type",
59 "title",
60 "authors",
61 "year",
62 "identifiers",
63 "locator",
64 "confidentiality",
65 "verification",
66}
67IDENTIFIER_FIELDS = {"doi", "pmid", "pmcid", "isbn", "url"}
68VERIFICATION_FIELDS = {"status", "source_opened", "verified_by", "verified_on"}
71def _check_unknown_fields(
72 obj: dict[str, Any],
73 allowed: set[str],
74 *,
75 location: str,
76 issues: list[Issue],
77) -> None:
78 for key in sorted(set(obj) - allowed):
79 issues.append(
80 issue("error", "UNKNOWN_SCHEMA_FIELD", location=location, item_id=key)
81 )
84def _missing_or_placeholder(
85 obj: dict[str, Any],
86 key: str,
87 *,
88 location: str,
89 issues: list[Issue],
90) -> None:
91 value = obj.get(key)
92 if not is_nonempty_string(value) or is_placeholder(value):
93 issues.append(
94 issue("error", "MISSING_VERIFIED_VALUE", location=location, item_id=key)
95 )
98def _validate_human_gate(
99 value: Any,
100 *,
101 location: str,
102 submission_ready: bool,
103 issues: list[Issue],
104) -> None:
105 gate = require_object(value, location)
106 completed = gate.get("completed")
107 if not isinstance(completed, bool):
108 issues.append(issue("error", "GATE_COMPLETED_NOT_BOOLEAN", location=location))
109 return
110 if completed:
111 _missing_or_placeholder(gate, "verified_by", location=location, issues=issues)
112 verified_on = gate.get("verified_on")
113 if not is_iso_date(verified_on):
114 issues.append(
115 issue("error", "INVALID_VERIFICATION_DATE", location=location)
116 )
117 elif submission_ready:
118 issues.append(issue("error", "SUBMISSION_GATE_INCOMPLETE", location=location))
121def validate_manuscript_manifest(data: dict[str, Any]) -> list[Issue]:
122 issues: list[Issue] = []
123 _check_unknown_fields(
124 data,
125 MANUSCRIPT_FIELDS,
126 location="root",
127 issues=issues,
128 )
129 if data.get("schema_version") != "1.0":
130 issues.append(
131 issue("error", "UNSUPPORTED_SCHEMA_VERSION", location="schema_version")
132 )
134 document_id = data.get("document_id")
135 if not isinstance(document_id, str) or not ID_RE.fullmatch(document_id):
136 issues.append(issue("error", "INVALID_DOCUMENT_ID", location="document_id"))
137 _missing_or_placeholder(data, "study_design", location="root", issues=issues)
139 draft_status = data.get("draft_status")
140 if draft_status not in DRAFT_STATUSES:
141 issues.append(issue("error", "INVALID_DRAFT_STATUS", location="draft_status"))
142 submission_ready = data.get("submission_ready")
143 if not isinstance(submission_ready, bool):
144 issues.append(
145 issue("error", "SUBMISSION_READY_NOT_BOOLEAN", location="submission_ready")
146 )
147 submission_ready = False
149 paths = require_object(data.get("registries"), "registries")
150 for key in (
151 "source_manifest",
152 "claim_evidence",
153 "consistency_manifest",
154 "authorship_manifest",
155 "reporting_coverage",
156 ):
157 _missing_or_placeholder(paths, key, location="registries", issues=issues)
159 guidelines = require_list(data.get("reporting_guidelines"), "reporting_guidelines")
160 if submission_ready and not guidelines:
161 issues.append(
162 issue(
163 "error",
164 "NO_REPORTING_GUIDELINE_RECORDED",
165 location="reporting_guidelines",
166 )
167 )
168 for index, guideline_id in enumerate(guidelines):
169 if not isinstance(guideline_id, str) or not ID_RE.fullmatch(guideline_id):
170 issues.append(
171 issue(
172 "error",
173 "INVALID_GUIDELINE_ID",
174 location=f"reporting_guidelines[{index}]",
175 )
176 )
178 _validate_human_gate(
179 data.get("human_verification"),
180 location="human_verification",
181 submission_ready=bool(submission_ready),
182 issues=issues,
183 )
184 confidentiality = require_object(
185 data.get("confidentiality_review"), "confidentiality_review"
186 )
187 _validate_human_gate(
188 confidentiality,
189 location="confidentiality_review",
190 submission_ready=bool(submission_ready),
191 issues=issues,
192 )
193 if not isinstance(confidentiality.get("policy_checked"), bool):
194 issues.append(
195 issue(
196 "error", "POLICY_CHECKED_NOT_BOOLEAN", location="confidentiality_review"
197 )
198 )
199 elif submission_ready and not confidentiality["policy_checked"]:
200 issues.append(
201 issue(
202 "error", "POLICY_REVIEW_INCOMPLETE", location="confidentiality_review"
203 )
204 )
205 if not isinstance(confidentiality.get("external_services_authorized"), bool):
206 issues.append(
207 issue(
208 "error",
209 "EXTERNAL_AUTHORIZATION_NOT_BOOLEAN",
210 location="confidentiality_review",
211 )
212 )
214 statements = require_object(data.get("required_statements"), "required_statements")
215 for key in (
216 "ethics",
217 "consent",
218 "funding",
219 "conflicts",
220 "data_availability",
221 "code_availability",
222 "author_contributions",
223 "ai_disclosure",
224 ):
225 status = statements.get(key)
226 if status not in STATEMENT_STATUSES:
227 issues.append(issue("error", "INVALID_STATEMENT_STATUS", location=key))
228 elif submission_ready and status in {"missing", "draft"}:
229 issues.append(issue("error", "UNVERIFIED_REQUIRED_STATEMENT", location=key))
231 if submission_ready and draft_status not in {"submission_candidate", "submitted"}:
232 issues.append(issue("error", "READY_STATUS_MISMATCH", location="draft_status"))
233 return issues
236def validate_source_manifest(
237 data: dict[str, Any],
238 *,
239 require_verified: bool,
240) -> list[Issue]:
241 issues: list[Issue] = []
242 _check_unknown_fields(
243 data,
244 {"schema_version", "sources"},
245 location="root",
246 issues=issues,
247 )
248 if data.get("schema_version") != "1.0":
249 issues.append(
250 issue("error", "UNSUPPORTED_SCHEMA_VERSION", location="schema_version")
251 )
252 sources = require_list(data.get("sources"), "sources")
253 seen_ids: set[str] = set()
254 for index, raw_source in enumerate(sources):
255 location = f"sources[{index}]"
256 source = require_object(raw_source, location)
257 _check_unknown_fields(
258 source,
259 SOURCE_FIELDS,
260 location=location,
261 issues=issues,
262 )
263 evidence_id = source.get("evidence_id")
264 if not isinstance(evidence_id, str) or not EVIDENCE_ID_RE.fullmatch(
265 evidence_id
266 ):
267 issues.append(issue("error", "INVALID_EVIDENCE_ID", location=location))
268 evidence_id = None
269 elif evidence_id in seen_ids:
270 issues.append(
271 issue(
272 "error",
273 "DUPLICATE_EVIDENCE_ID",
274 location=location,
275 item_id=evidence_id,
276 )
277 )
278 else:
279 seen_ids.add(evidence_id)
281 if source.get("source_type") not in SOURCE_TYPES:
282 issues.append(
283 issue(
284 "error",
285 "INVALID_SOURCE_TYPE",
286 location=location,
287 item_id=evidence_id,
288 )
289 )
290 _missing_or_placeholder(source, "title", location=location, issues=issues)
291 authors = require_list(source.get("authors"), f"{location}.authors")
292 for author in authors:
293 if not is_nonempty_string(author) or is_placeholder(author):
294 issues.append(
295 issue(
296 "error",
297 "INVALID_SOURCE_AUTHOR",
298 location=location,
299 item_id=evidence_id,
300 )
301 )
303 year = source.get("year")
304 if year is not None and (
305 not isinstance(year, int)
306 or isinstance(year, bool)
307 or not 1000 <= year <= 2100
308 ):
309 issues.append(
310 issue(
311 "error",
312 "INVALID_SOURCE_YEAR",
313 location=location,
314 item_id=evidence_id,
315 )
316 )
318 identifiers = require_object(
319 source.get("identifiers"), f"{location}.identifiers"
320 )
321 _check_unknown_fields(
322 identifiers,
323 IDENTIFIER_FIELDS,
324 location=f"{location}.identifiers",
325 issues=issues,
326 )
327 if not any(is_nonempty_string(value) for value in identifiers.values()):
328 issues.append(
329 issue(
330 "warning",
331 "NO_SOURCE_IDENTIFIER",
332 location=location,
333 item_id=evidence_id,
334 )
335 )
336 if not is_nonempty_string(source.get("locator")) or is_placeholder(
337 source.get("locator")
338 ):
339 issues.append(
340 issue(
341 "error",
342 "MISSING_SOURCE_LOCATOR",
343 location=location,
344 item_id=evidence_id,
345 )
346 )
348 confidentiality = source.get("confidentiality")
349 if confidentiality not in CONFIDENTIALITY:
350 issues.append(
351 issue(
352 "error",
353 "INVALID_CONFIDENTIALITY_CLASS",
354 location=location,
355 item_id=evidence_id,
356 )
357 )
359 verification = require_object(
360 source.get("verification"), f"{location}.verification"
361 )
362 _check_unknown_fields(
363 verification,
364 VERIFICATION_FIELDS,
365 location=f"{location}.verification",
366 issues=issues,
367 )
368 status = verification.get("status")
369 if status not in {"unverified", "verified", "rejected"}:
370 issues.append(
371 issue(
372 "error",
373 "INVALID_VERIFICATION_STATUS",
374 location=location,
375 item_id=evidence_id,
376 )
377 )
378 if status == "verified":
379 _missing_or_placeholder(
380 verification,
381 "verified_by",
382 location=location,
383 issues=issues,
384 )
385 verified_on = verification.get("verified_on")
386 if not is_iso_date(verified_on):
387 issues.append(
388 issue(
389 "error",
390 "INVALID_VERIFICATION_DATE",
391 location=location,
392 item_id=evidence_id,
393 )
394 )
395 if verification.get("source_opened") is not True:
396 issues.append(
397 issue(
398 "error",
399 "SOURCE_NOT_OPENED_FOR_VERIFICATION",
400 location=location,
401 item_id=evidence_id,
402 )
403 )
404 elif require_verified:
405 issues.append(
406 issue(
407 "error",
408 "SOURCE_NOT_VERIFIED",
409 location=location,
410 item_id=evidence_id,
411 )
412 )
413 return issues
416def detect_kind(data: dict[str, Any]) -> str:
417 if "sources" in data:
418 return "source"
419 if "document_id" in data:
420 return "manuscript"
421 raise InputError("unable to infer manifest kind")
424def build_parser() -> argparse.ArgumentParser:
425 parser = argparse.ArgumentParser(
426 description=(
427 "Validate a local JSON manuscript or source manifest. "
428 "No network calls or identifier resolution are performed."
429 )
430 )
431 parser.add_argument("manifest", help="UTF-8 JSON manifest (maximum 5 MB)")
432 parser.add_argument(
433 "--kind",
434 choices=("auto", "manuscript", "source"),
435 default="auto",
436 help="manifest schema to validate (default: infer from keys)",
437 )
438 parser.add_argument(
439 "--require-verified",
440 action="store_true",
441 help="for source manifests, fail if any source is not verified",
442 )
443 return parser
446def cli() -> int:
447 args = build_parser().parse_args()
448 data = require_object(read_json(args.manifest))
449 kind = detect_kind(data) if args.kind == "auto" else args.kind
450 if kind == "manuscript":
451 issues = validate_manuscript_manifest(data)
452 records = 1
453 else:
454 issues = validate_source_manifest(data, require_verified=args.require_verified)
455 records = len(require_list(data.get("sources"), "sources"))
456 return emit_report(TOOL, issues, summary={"kind": kind, "records": records})
459if __name__ == "__main__":
460 run(TOOL, cli)