#!/usr/bin/env python3 """Validate author/institution-supplied metadata before a formal v1 release.""" from __future__ import annotations import argparse import json import re import tempfile from datetime import datetime from pathlib import Path from typing import Any DOI_PATTERN = re.compile(r"^10\.\d{4,9}/\S+$", re.IGNORECASE) EMAIL_PATTERN = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") PENDING_MARKERS = { "", "pending", "tbd", "todo", "unknown", "not_minted", "doi_pending", "null", "none", } def meaningful(value: Any) -> bool: return ( isinstance(value, str) and value.strip().lower() not in PENDING_MARKERS ) def nested(payload: dict[str, Any], *keys: str) -> Any: value: Any = payload for key in keys: if not isinstance(value, dict): return None value = value.get(key) return value def iso_datetime(value: Any) -> bool: if not meaningful(value): return False try: datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return False return True def validate( payload: dict[str, Any], manifest: dict[str, Any] ) -> list[str]: errors: list[str] = [] expected_release = manifest["version_axes"]["resource_release"] expected_layers = manifest.get("release_layers") or [ { "layer": "analysis_core", "archive": manifest["canonical_archive"], } ] if payload.get("resource_release") != expected_release: errors.append("resource_release does not match the release manifest") deposited = nested(payload, "deposit", "archives") if not isinstance(deposited, list): errors.append("deposit.archives is required") else: deposited_by_layer = { row.get("layer"): row for row in deposited if isinstance(row, dict) and isinstance(row.get("layer"), str) } for expected_layer in expected_layers: layer = expected_layer["layer"] expected_archive = expected_layer["archive"] deposited_archive = deposited_by_layer.get(layer) if not isinstance(deposited_archive, dict): errors.append(f"deposit.archives is missing layer {layer}") continue for field in ("filename", "size_bytes", "sha256"): if deposited_archive.get(field) != expected_archive.get(field): errors.append( f"deposit.archives[{layer}].{field} does not match the release manifest" ) doi = nested(payload, "deposit", "doi") if not meaningful(doi) or not DOI_PATTERN.fullmatch(str(doi).strip()): errors.append("deposit.doi must be a minted DOI without a URL prefix") if not meaningful(nested(payload, "deposit", "repository")): errors.append("deposit.repository is required") for section in ("data_licence", "code_licence"): identifier = nested(payload, section, "identifier") status = nested(payload, section, "approval_status") if not meaningful(identifier): errors.append(f"{section}.identifier is required") if status != "institution_approved": errors.append( f"{section}.approval_status must be institution_approved" ) for field in ("name", "institution", "email"): if not meaningful(nested(payload, "maintainer", field)): errors.append(f"maintainer.{field} is required") email = nested(payload, "maintainer", "email") if meaningful(email) and not EMAIL_PATTERN.fullmatch(str(email).strip()): errors.append("maintainer.email is not a valid email address") if nested(payload, "maintainer", "approval_status") != "author_approved": errors.append( "maintainer.approval_status must be author_approved" ) authorization = nested(payload, "formal_release_authorization") if not isinstance(authorization, dict): errors.append("formal_release_authorization is required") else: if authorization.get("authorized") is not True: errors.append( "formal_release_authorization.authorized must be true" ) if not meaningful(authorization.get("approved_by")): errors.append( "formal_release_authorization.approved_by is required" ) if not iso_datetime(authorization.get("approved_at")): errors.append( "formal_release_authorization.approved_at must be ISO-8601" ) return errors def run_self_test() -> None: manifest = { "version_axes": {"resource_release": "v0.9.2-beta"}, "canonical_archive": { "filename": "archive.tar.gz", "size_bytes": 42, "sha256": "a" * 64, }, "release_layers": [ { "layer": "analysis_core", "archive": { "filename": "archive.tar.gz", "size_bytes": 42, "sha256": "a" * 64, }, }, { "layer": "evidence_atlas_overlay", "archive": { "filename": "overlay.tar.gz", "size_bytes": 17, "sha256": "b" * 64, }, }, ], } valid = { "resource_release": "v0.9.2-beta", "deposit": { "repository": "Example repository", "doi": "10.1234/example.record", "archives": [ { "layer": row["layer"], **row["archive"], } for row in manifest["release_layers"] ], }, "data_licence": { "identifier": "Example-Data-License", "approval_status": "institution_approved", }, "code_licence": { "identifier": "Example-Code-License", "approval_status": "institution_approved", }, "maintainer": { "name": "Example Maintainer", "institution": "Example Institution", "email": "maintainer@example.org", "approval_status": "author_approved", }, "formal_release_authorization": { "authorized": True, "approved_by": "Example Approver", "approved_at": "2026-07-25T00:00:00+00:00", }, } invalid = json.loads(json.dumps(valid)) invalid["deposit"]["doi"] = None invalid["formal_release_authorization"]["authorized"] = False assert validate(valid, manifest) == [] failures = validate(invalid, manifest) assert any("deposit.doi" in error for error in failures) assert any("authorized must be true" in error for error in failures) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "roundtrip.json" path.write_text(json.dumps(valid)) assert json.loads(path.read_text()) == valid print("v1 external-gate validator self-test: 2/2 passed") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("metadata", type=Path, nargs="?") parser.add_argument("--manifest", type=Path) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: run_self_test() return 0 if args.metadata is None or args.manifest is None: parser.error("metadata and --manifest are required") payload = json.loads(args.metadata.read_text()) manifest = json.loads(args.manifest.read_text()) errors = validate(payload, manifest) if errors: print("FORMAL_V1_NOT_READY") for error in errors: print(f"- {error}") return 2 print("FORMAL_V1_EXTERNAL_GATES_COMPLETE") return 0 if __name__ == "__main__": raise SystemExit(main())