| 1 | #!/usr/bin/env python3 |
| 2 | """Validate crawl-and-publish rerun modes before any publishing side effects.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import argparse |
| 7 | import json |
| 8 | import re |
| 9 | from dataclasses import dataclass |
| 10 | from pathlib import Path |
| 11 | |
| 12 | RUN_MODES = {"normal", "dry-run", "restore", "force-replace", "candidate-only"} |
| 13 | SOURCE_REFRESH_POLICIES = {"reuse-same-day", "refresh-missing-stale", "force-refresh"} |
| 14 | WEEK_PATTERN = re.compile(r"^[0-9]{4}-W[0-9]{2}$") |
| 15 | RUN_ID_PATTERN = re.compile(r"^[0-9]+$") |
| 16 | |
| 17 | |
| 18 | @dataclass(frozen=True) |
| 19 | class ModeDecision: |
| 20 | run_mode: str |
| 21 | source_refresh_policy: str |
| 22 | action: str |
| 23 | publish_allowed: bool |
| 24 | crawl_allowed: bool |
| 25 | reasons: list[str] |
| 26 | |
| 27 | |
| 28 | def validate_modes( |
| 29 | *, |
| 30 | run_mode: str, |
| 31 | source_refresh_policy: str, |
| 32 | rebuild_week: str = "", |
| 33 | source_run_id: str = "", |
| 34 | publish_release: bool = False, |
| 35 | ) -> ModeDecision: |
| 36 | reasons: list[str] = [] |
| 37 | if run_mode not in RUN_MODES: |
| 38 | reasons.append(f"invalid run_mode: {run_mode}") |
| 39 | if source_refresh_policy not in SOURCE_REFRESH_POLICIES: |
| 40 | reasons.append(f"invalid source_refresh_policy: {source_refresh_policy}") |
| 41 | if rebuild_week and run_mode != "restore": |
| 42 | reasons.append("rebuild_week is a restore operation and requires run_mode=restore") |
| 43 | if rebuild_week and not WEEK_PATTERN.fullmatch(rebuild_week): |
| 44 | reasons.append("rebuild_week must use YYYY-WNN format") |
| 45 | if run_mode == "restore" and not rebuild_week: |
| 46 | reasons.append("run_mode=restore requires rebuild_week=YYYY-WNN") |
| 47 | if run_mode == "restore" and not source_run_id: |
| 48 | reasons.append("run_mode=restore requires source_run_id") |
| 49 | if source_run_id and run_mode != "restore": |
| 50 | reasons.append("source_run_id is only allowed with run_mode=restore") |
| 51 | if source_run_id and not RUN_ID_PATTERN.fullmatch(source_run_id): |
| 52 | reasons.append("source_run_id must be a numeric GitHub Actions workflow run ID") |
| 53 | if run_mode in {"dry-run", "candidate-only"} and publish_release: |
| 54 | reasons.append(f"publish_release is not allowed with run_mode={run_mode}") |
| 55 | if run_mode == "restore" and source_refresh_policy == "force-refresh": |
| 56 | reasons.append( |
| 57 | "run_mode=restore hydrates publish artifacts and cannot force-refresh sources" |
| 58 | ) |
| 59 | |
| 60 | publish_allowed = run_mode not in {"dry-run", "candidate-only"} |
| 61 | crawl_allowed = not rebuild_week |
| 62 | if run_mode == "dry-run": |
| 63 | action = "analyze candidate only; never commit or deploy" |
| 64 | elif run_mode == "candidate-only": |
| 65 | action = "produce candidate artifacts only; never promote" |
| 66 | elif run_mode == "restore": |
| 67 | action = f"restore published artifacts for {rebuild_week} and regenerate through guarded promotion" |
| 68 | elif run_mode == "force-replace": |
| 69 | action = "explicit replacement run; promotion still requires all gates" |
| 70 | else: |
| 71 | action = "normal guarded crawl, analysis, publish, and deploy" |
| 72 | |
| 73 | return ModeDecision( |
| 74 | run_mode, source_refresh_policy, action, publish_allowed, crawl_allowed, reasons |
| 75 | ) |
| 76 | |
| 77 | |
| 78 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 79 | parser = argparse.ArgumentParser(description=__doc__) |
| 80 | parser.add_argument("--run-mode", default="normal") |
| 81 | parser.add_argument("--source-refresh-policy", default="reuse-same-day") |
| 82 | parser.add_argument("--rebuild-week", default="") |
| 83 | parser.add_argument("--source-run-id", default="") |
| 84 | parser.add_argument("--publish-release", action="store_true") |
| 85 | parser.add_argument("--summary-json", type=Path) |
| 86 | return parser.parse_args(argv) |
| 87 | |
| 88 | |
| 89 | def main(argv: list[str] | None = None) -> int: |
| 90 | args = parse_args(argv) |
| 91 | decision = validate_modes( |
| 92 | run_mode=args.run_mode, |
| 93 | source_refresh_policy=args.source_refresh_policy, |
| 94 | rebuild_week=args.rebuild_week.strip(), |
| 95 | source_run_id=args.source_run_id.strip(), |
| 96 | publish_release=args.publish_release, |
| 97 | ) |
| 98 | payload = { |
| 99 | "run_mode": decision.run_mode, |
| 100 | "source_refresh_policy": decision.source_refresh_policy, |
| 101 | "publish_allowed": decision.publish_allowed, |
| 102 | "crawl_allowed": decision.crawl_allowed, |
| 103 | "source_run_id": args.source_run_id.strip(), |
| 104 | "action": decision.action, |
| 105 | "valid": not decision.reasons, |
| 106 | "reasons": decision.reasons, |
| 107 | } |
| 108 | if args.summary_json: |
| 109 | args.summary_json.parent.mkdir(parents=True, exist_ok=True) |
| 110 | args.summary_json.write_text( |
| 111 | json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" |
| 112 | ) |
| 113 | print(json.dumps(payload, sort_keys=True)) |
| 114 | if decision.reasons: |
| 115 | for reason in decision.reasons: |
| 116 | print(f"::error::{reason}") |
| 117 | return 1 |
| 118 | return 0 |
| 119 | |
| 120 | |
| 121 | if __name__ == "__main__": |
| 122 | raise SystemExit(main()) |