| 1 | from __future__ import annotations |
| 2 | |
| 3 | import argparse |
| 4 | import hashlib |
| 5 | import json |
| 6 | import os |
| 7 | import re |
| 8 | from datetime import UTC, date, datetime |
| 9 | from pathlib import Path |
| 10 | from typing import Any |
| 11 | |
| 12 | |
| 13 | class PromotionBlocked(ValueError): |
| 14 | def __init__(self, reasons: list[str]): |
| 15 | self.reasons = reasons |
| 16 | super().__init__("; ".join(reasons)) |
| 17 | |
| 18 | |
| 19 | WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$") |
| 20 | FRONTMATTER_PATTERN = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL) |
| 21 | FALLBACK_MIN_QUALITY_SCORE = 70 |
| 22 | PROMOTION_TRANSACTION_SCHEMA_VERSION = "promotion_transaction_v1" |
| 23 | |
| 24 | |
| 25 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 26 | parser = argparse.ArgumentParser(description="Promote an eligible weekly analysis candidate.") |
| 27 | parser.add_argument( |
| 28 | "--manifest", required=True, type=Path, help="Publish eligibility manifest path." |
| 29 | ) |
| 30 | parser.add_argument("--root", default=".", type=Path, help="Repository/workspace root.") |
| 31 | return parser.parse_args(argv) |
| 32 | |
| 33 | |
| 34 | def _load_manifest(path: Path) -> dict[str, Any]: |
| 35 | try: |
| 36 | manifest = json.loads(path.read_text(encoding="utf-8")) |
| 37 | except FileNotFoundError as exc: |
| 38 | raise PromotionBlocked([f"Missing publish eligibility manifest: {path}"]) from exc |
| 39 | except json.JSONDecodeError as exc: |
| 40 | raise PromotionBlocked([f"Malformed publish eligibility manifest: {exc.msg}"]) from exc |
| 41 | |
| 42 | if not isinstance(manifest, dict): |
| 43 | raise PromotionBlocked(["Publish eligibility manifest must be a JSON object."]) |
| 44 | return manifest |
| 45 | |
| 46 | |
| 47 | def _parse_date(value: Any) -> date | None: |
| 48 | if not isinstance(value, str) or not value.strip(): |
| 49 | return None |
| 50 | candidate = value.strip() |
| 51 | if candidate.endswith("Z"): |
| 52 | candidate = f"{candidate[:-1]}+00:00" |
| 53 | try: |
| 54 | return datetime.fromisoformat(candidate).astimezone(UTC).date() |
| 55 | except ValueError: |
| 56 | try: |
| 57 | return date.fromisoformat(value.strip()) |
| 58 | except ValueError: |
| 59 | return None |
| 60 | |
| 61 | |
| 62 | def _resolve_under_root(root: Path, value: Any, field: str, reasons: list[str]) -> Path | None: |
| 63 | if not isinstance(value, str) or not value.strip(): |
| 64 | reasons.append(f"{field} is required.") |
| 65 | return None |
| 66 | path = Path(value) |
| 67 | if path.is_absolute(): |
| 68 | reasons.append(f"{field} must be relative to the repository root.") |
| 69 | return None |
| 70 | resolved_root = root.resolve() |
| 71 | resolved_path = (resolved_root / path).resolve() |
| 72 | try: |
| 73 | resolved_path.relative_to(resolved_root) |
| 74 | except ValueError: |
| 75 | reasons.append(f"{field} must stay under the repository root.") |
| 76 | return None |
| 77 | return resolved_path |
| 78 | |
| 79 | |
| 80 | def _manifest_candidate_path(manifest: dict[str, Any], legacy_key: str, nested_key: str) -> Any: |
| 81 | candidate = manifest.get("candidate") |
| 82 | if isinstance(candidate, dict) and nested_key in candidate: |
| 83 | return candidate.get(nested_key) |
| 84 | return manifest.get(legacy_key) |
| 85 | |
| 86 | |
| 87 | def _manifest_candidate_content_path(manifest: dict[str, Any]) -> Any: |
| 88 | content_path = _manifest_candidate_path(manifest, "candidate_content_path", "content_path") |
| 89 | if content_path is not None: |
| 90 | return content_path |
| 91 | return _manifest_candidate_path(manifest, "candidate_summary_path", "summary_path") |
| 92 | |
| 93 | |
| 94 | def _manifest_promotion_eligible(manifest: dict[str, Any]) -> bool: |
| 95 | promotion = manifest.get("promotion") |
| 96 | if isinstance(promotion, dict): |
| 97 | return promotion.get("eligible") is True and promotion.get("decision") == "promote" |
| 98 | return manifest.get("promotion_eligible") is True |
| 99 | |
| 100 | |
| 101 | def _manifest_ai_provenance(manifest: dict[str, Any]) -> dict[str, Any] | None: |
| 102 | ai_provenance = manifest.get("ai_provenance") |
| 103 | if isinstance(ai_provenance, dict): |
| 104 | return ai_provenance |
| 105 | analysis = manifest.get("analysis") |
| 106 | if isinstance(analysis, dict): |
| 107 | provenance = ( |
| 108 | analysis.get("provenance") if isinstance(analysis.get("provenance"), dict) else {} |
| 109 | ) |
| 110 | return { |
| 111 | "source": analysis.get("source"), |
| 112 | "model": analysis.get("model"), |
| 113 | "degraded": analysis.get("ai_status") not in {"ai", "no-ai"} |
| 114 | or (analysis.get("ai_status") == "ai" and analysis.get("model_status") != "available"), |
| 115 | "authorship": provenance.get("authorship"), |
| 116 | "fallback_reason": provenance.get("fallback_reason"), |
| 117 | "attempted_ai_paths": provenance.get("attempted_ai_paths"), |
| 118 | } |
| 119 | return None |
| 120 | |
| 121 | |
| 122 | def _manifest_gate_results(manifest: dict[str, Any]) -> dict[str, bool] | None: |
| 123 | gate_results = manifest.get("gate_results") |
| 124 | if isinstance(gate_results, dict): |
| 125 | return {str(key): value is True for key, value in gate_results.items()} |
| 126 | validation = manifest.get("validation") |
| 127 | gate_report = validation.get("gate_report") if isinstance(validation, dict) else None |
| 128 | gates = gate_report.get("gates") if isinstance(gate_report, dict) else None |
| 129 | if isinstance(gates, dict): |
| 130 | return { |
| 131 | str(key): isinstance(value, dict) and value.get("passed") is True |
| 132 | for key, value in gates.items() |
| 133 | } |
| 134 | return None |
| 135 | |
| 136 | |
| 137 | def _manifest_gate_report(manifest: dict[str, Any]) -> dict[str, Any] | None: |
| 138 | validation = manifest.get("validation") |
| 139 | gate_report = validation.get("gate_report") if isinstance(validation, dict) else None |
| 140 | return gate_report if isinstance(gate_report, dict) else None |
| 141 | |
| 142 | |
| 143 | def _manifest_source_artifacts(manifest: dict[str, Any]) -> list[Any] | None: |
| 144 | artifacts = manifest.get("source_artifacts") |
| 145 | return artifacts if isinstance(artifacts, list) else None |
| 146 | |
| 147 | |
| 148 | def _manifest_run_started_at(manifest: dict[str, Any]) -> Any: |
| 149 | return manifest.get("run_started_at") or manifest.get("generated_at") |
| 150 | |
| 151 | |
| 152 | def _sha256_file(path: Path) -> str | None: |
| 153 | if not path.exists() or not path.is_file(): |
| 154 | return None |
| 155 | digest = hashlib.sha256() |
| 156 | with path.open("rb") as handle: |
| 157 | for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| 158 | digest.update(chunk) |
| 159 | return digest.hexdigest() |
| 160 | |
| 161 | |
| 162 | def _relative_to_root(root: Path, path: Path) -> str: |
| 163 | return path.resolve().relative_to(root.resolve()).as_posix() |
| 164 | |
| 165 | |
| 166 | def _artifact_reused_same_day(artifact: dict[str, Any]) -> bool: |
| 167 | if artifact.get("reused_same_day") is True: |
| 168 | return True |
| 169 | same_day_reuse = artifact.get("same_day_reuse") |
| 170 | if isinstance(same_day_reuse, dict): |
| 171 | return str(same_day_reuse.get("status", "")).lower() in { |
| 172 | "reused", |
| 173 | "same_day_reuse", |
| 174 | "same-day-reuse", |
| 175 | } |
| 176 | return str(same_day_reuse or "").lower() in {"reused", "same_day_reuse", "same-day-reuse"} |
| 177 | |
| 178 | |
| 179 | def _frontmatter(path: Path) -> dict[str, Any]: |
| 180 | if not path.exists() or not path.is_file(): |
| 181 | return {} |
| 182 | match = FRONTMATTER_PATTERN.match(path.read_text(encoding="utf-8")) |
| 183 | if not match: |
| 184 | return {} |
| 185 | result: dict[str, Any] = {} |
| 186 | for line in match.group(1).splitlines(): |
| 187 | if ":" not in line or line.startswith((" ", "\t")): |
| 188 | continue |
| 189 | key, value = line.split(":", 1) |
| 190 | scalar = value.strip().strip('"').strip("'") |
| 191 | result[key.strip()] = int(scalar) if scalar.isdigit() else scalar |
| 192 | return result |
| 193 | |
| 194 | |
| 195 | def _is_no_ai_summary(path: Path) -> bool: |
| 196 | if not path.exists(): |
| 197 | return False |
| 198 | text = path.read_text(encoding="utf-8").lower() |
| 199 | return any( |
| 200 | marker in text |
| 201 | for marker in ( |
| 202 | "source: no-ai", |
| 203 | "model: none", |
| 204 | "without ai-powered analysis", |
| 205 | "generated without ai assistance", |
| 206 | "automated data-only summary", |
| 207 | ) |
| 208 | ) |
| 209 | |
| 210 | |
| 211 | def _has_existing_good_ai_article(root: Path, week: str) -> bool: |
| 212 | path = root / "data" / "analyzed" / f"{week}-summary.md" |
| 213 | score = _frontmatter(path).get("quality_score") |
| 214 | return path.exists() and isinstance(score, int) and score >= 60 and not _is_no_ai_summary(path) |
| 215 | |
| 216 | |
| 217 | def _promotion_policy(manifest: dict[str, Any]) -> dict[str, Any]: |
| 218 | policy = manifest.get("promotion_policy") |
| 219 | if isinstance(policy, dict): |
| 220 | return policy |
| 221 | promotion = manifest.get("promotion") |
| 222 | if isinstance(promotion, dict): |
| 223 | audit = manifest.get("audit") if isinstance(manifest.get("audit"), dict) else {} |
| 224 | return { |
| 225 | "mode": promotion.get("policy", "default"), |
| 226 | "reason": promotion.get("reason") or audit.get("reason"), |
| 227 | "actor": audit.get("actor"), |
| 228 | } |
| 229 | return {"mode": "default"} |
| 230 | |
| 231 | |
| 232 | def _validate_manifest( |
| 233 | manifest: dict[str, Any], root: Path, manifest_path: Path |
| 234 | ) -> tuple[str, Path, Path, list[str]]: |
| 235 | reasons: list[str] = [] |
| 236 | |
| 237 | if manifest.get("schema_version") != "publish_eligibility_v1": |
| 238 | reasons.append("schema_version must be publish_eligibility_v1.") |
| 239 | |
| 240 | week = manifest.get("week") |
| 241 | if not isinstance(week, str) or not week.strip(): |
| 242 | reasons.append("week is required.") |
| 243 | week = "unknown-week" |
| 244 | elif not WEEK_PATTERN.fullmatch(week): |
| 245 | reasons.append("week must use YYYY-WNN format.") |
| 246 | |
| 247 | if not _manifest_promotion_eligible(manifest): |
| 248 | reasons.append("promotion_eligible must be true.") |
| 249 | |
| 250 | candidate_summary = _resolve_under_root( |
| 251 | root, |
| 252 | _manifest_candidate_path(manifest, "candidate_summary_path", "summary_path"), |
| 253 | "candidate_summary_path", |
| 254 | reasons, |
| 255 | ) |
| 256 | candidate_content = _resolve_under_root( |
| 257 | root, _manifest_candidate_content_path(manifest), "candidate_content_path", reasons |
| 258 | ) |
| 259 | |
| 260 | policy = _promotion_policy(manifest) |
| 261 | policy_mode = str(policy.get("mode") or "default") |
| 262 | ai_provenance = _manifest_ai_provenance(manifest) |
| 263 | if not isinstance(ai_provenance, dict): |
| 264 | reasons.append("ai_provenance is required.") |
| 265 | else: |
| 266 | source = ai_provenance.get("source") |
| 267 | is_no_ai = source == "no-ai" or ai_provenance.get("authorship") == "no-ai-fallback" |
| 268 | if is_no_ai: |
| 269 | existing_good_ai = _has_existing_good_ai_article(root, str(week)) |
| 270 | if policy_mode == "force-replace": |
| 271 | if not policy.get("reason"): |
| 272 | reasons.append("force-replace requires a reason.") |
| 273 | if not policy.get("actor"): |
| 274 | reasons.append("force-replace requires an actor.") |
| 275 | elif policy_mode == "allow-no-ai-first-publish": |
| 276 | if existing_good_ai: |
| 277 | reasons.append( |
| 278 | "no-AI fallback cannot first-publish over an existing good AI-authored article." |
| 279 | ) |
| 280 | else: |
| 281 | reasons.append("no-AI fallback is ineligible for default promotion.") |
| 282 | if existing_good_ai: |
| 283 | reasons.append( |
| 284 | "no-AI fallback is ineligible to replace an existing good AI-authored article by default." |
| 285 | ) |
| 286 | if candidate_summary is not None: |
| 287 | quality_score = _frontmatter(candidate_summary).get("quality_score") |
| 288 | if not isinstance(quality_score, int) or quality_score < FALLBACK_MIN_QUALITY_SCORE: |
| 289 | reasons.append( |
| 290 | f"no-AI fallback quality_score must be at least {FALLBACK_MIN_QUALITY_SCORE}." |
| 291 | ) |
| 292 | if not ai_provenance.get("fallback_reason"): |
| 293 | reasons.append("no-AI fallback provenance requires fallback_reason.") |
| 294 | if not ai_provenance.get("attempted_ai_paths"): |
| 295 | reasons.append("no-AI fallback provenance requires attempted_ai_paths.") |
| 296 | else: |
| 297 | if policy_mode == "force-replace": |
| 298 | if not policy.get("reason"): |
| 299 | reasons.append("force-replace requires a reason.") |
| 300 | if not policy.get("actor"): |
| 301 | reasons.append("force-replace requires an actor.") |
| 302 | if source in {None, ""}: |
| 303 | reasons.append("AI-authored provenance is required for normal promotion.") |
| 304 | if ai_provenance.get("degraded") is True: |
| 305 | reasons.append("degraded AI provenance is not eligible for normal promotion.") |
| 306 | |
| 307 | gate_report = _manifest_gate_report(manifest) |
| 308 | if gate_report is not None and gate_report.get("passed") is not True: |
| 309 | reasons.append("validation.gate_report.passed must be true.") |
| 310 | |
| 311 | gate_results = _manifest_gate_results(manifest) |
| 312 | if not isinstance(gate_results, dict): |
| 313 | reasons.append("gate_results is required.") |
| 314 | else: |
| 315 | for gate in ( |
| 316 | "structural_schema", |
| 317 | "ai_provenance", |
| 318 | "evidence_citation", |
| 319 | "editorial_quality", |
| 320 | ): |
| 321 | if gate not in gate_results: |
| 322 | reasons.append(f"gate_results must include passing {gate}.") |
| 323 | elif gate_results.get(gate) is not True: |
| 324 | reasons.append(f"{gate} must pass.") |
| 325 | legacy_gates = ("analysis_gate", "editorial_quality_gate", "evidence_freshness_gate") |
| 326 | if any(gate in gate_results for gate in legacy_gates): |
| 327 | for gate in legacy_gates: |
| 328 | if gate_results.get(gate) is not True: |
| 329 | reasons.append(f"{gate} must pass.") |
| 330 | |
| 331 | run_date = _parse_date(_manifest_run_started_at(manifest)) |
| 332 | if run_date is None: |
| 333 | reasons.append("run_started_at/generated_at must be an ISO date or timestamp.") |
| 334 | |
| 335 | source_artifacts = _manifest_source_artifacts(manifest) |
| 336 | if not isinstance(source_artifacts, list) or not source_artifacts: |
| 337 | reasons.append("source_artifacts must include at least one artifact.") |
| 338 | else: |
| 339 | for index, artifact in enumerate(source_artifacts, start=1): |
| 340 | prefix = f"source_artifacts[{index}]" |
| 341 | if not isinstance(artifact, dict): |
| 342 | reasons.append(f"{prefix} must be an object.") |
| 343 | continue |
| 344 | freshness = artifact.get("freshness") |
| 345 | if artifact.get("stale") is True or ( |
| 346 | isinstance(freshness, dict) and freshness.get("status") == "stale" |
| 347 | ): |
| 348 | reasons.append(f"{prefix} is stale.") |
| 349 | generated_date = _parse_date(artifact.get("generated_at") or artifact.get("crawled_at")) |
| 350 | if generated_date is None: |
| 351 | reasons.append(f"{prefix}.generated_at must be an ISO date or timestamp.") |
| 352 | elif ( |
| 353 | run_date is not None |
| 354 | and generated_date != run_date |
| 355 | and not _artifact_reused_same_day(artifact) |
| 356 | ): |
| 357 | reasons.append( |
| 358 | f"{prefix} is not from the current run date or marked as same-day reuse." |
| 359 | ) |
| 360 | artifact_path = _resolve_under_root( |
| 361 | root, artifact.get("path"), f"{prefix}.path", reasons |
| 362 | ) |
| 363 | if artifact_path is not None and not artifact_path.exists(): |
| 364 | reasons.append(f"{prefix}.path does not exist: {artifact.get('path')}") |
| 365 | |
| 366 | if candidate_summary is not None and not candidate_summary.exists(): |
| 367 | reasons.append( |
| 368 | f"candidate_summary_path does not exist: {manifest.get('candidate_summary_path')}" |
| 369 | ) |
| 370 | if candidate_content is not None and not candidate_content.exists(): |
| 371 | reasons.append( |
| 372 | f"candidate_content_path does not exist: {manifest.get('candidate_content_path')}" |
| 373 | ) |
| 374 | |
| 375 | try: |
| 376 | manifest_relative = manifest_path.resolve().relative_to(root.resolve()) |
| 377 | except ValueError: |
| 378 | reasons.append("Publish manifest must be under the repository root.") |
| 379 | manifest_relative = Path() |
| 380 | if manifest_relative.parts[:2] not in {("data", "staging"), ("data", "candidates")}: |
| 381 | reasons.append("Publish manifest must live under data/staging/ or data/candidates/.") |
| 382 | |
| 383 | return str(week), candidate_summary or root, candidate_content or root, reasons |
| 384 | |
| 385 | |
| 386 | def _write_diagnostic( |
| 387 | root: Path, week: str, manifest: dict[str, Any] | None, reasons: list[str] |
| 388 | ) -> Path: |
| 389 | diagnostic_dir = root / "data" / "diagnostics" / "promotion" |
| 390 | diagnostic_dir.mkdir(parents=True, exist_ok=True) |
| 391 | diagnostic_path = diagnostic_dir / f"{week}-blocked.json" |
| 392 | diagnostic_path.write_text( |
| 393 | json.dumps( |
| 394 | {"week": week, "promotion": "blocked", "reasons": reasons, "manifest": manifest}, |
| 395 | indent=2, |
| 396 | sort_keys=True, |
| 397 | ) |
| 398 | + "\n", |
| 399 | encoding="utf-8", |
| 400 | ) |
| 401 | return diagnostic_path |
| 402 | |
| 403 | |
| 404 | def _write_force_audit(root: Path, week: str, manifest: dict[str, Any]) -> Path | None: |
| 405 | policy = _promotion_policy(manifest) |
| 406 | if policy.get("mode") != "force-replace": |
| 407 | return None |
| 408 | diagnostic_dir = root / "data" / "diagnostics" / "promotion" |
| 409 | diagnostic_dir.mkdir(parents=True, exist_ok=True) |
| 410 | audit_path = diagnostic_dir / f"{week}-force-replace-audit.json" |
| 411 | audit_path.write_text( |
| 412 | json.dumps( |
| 413 | { |
| 414 | "week": week, |
| 415 | "mode": "force-replace", |
| 416 | "actor": policy.get("actor"), |
| 417 | "reason": policy.get("reason"), |
| 418 | "source_artifacts": manifest.get("source_artifacts", []), |
| 419 | "manifest": manifest, |
| 420 | }, |
| 421 | indent=2, |
| 422 | sort_keys=True, |
| 423 | ) |
| 424 | + "\n", |
| 425 | encoding="utf-8", |
| 426 | ) |
| 427 | return audit_path |
| 428 | |
| 429 | |
| 430 | def _promotion_transaction_record( |
| 431 | *, |
| 432 | root: Path, |
| 433 | week: str, |
| 434 | manifest_path: Path, |
| 435 | manifest: dict[str, Any], |
| 436 | candidate_summary: Path, |
| 437 | candidate_content: Path, |
| 438 | canonical_summary: Path, |
| 439 | canonical_content: Path, |
| 440 | ) -> dict[str, Any]: |
| 441 | manifest_relative = _relative_to_root(root, manifest_path) |
| 442 | summary_sha = _sha256_file(candidate_summary) |
| 443 | content_sha = _sha256_file(candidate_content) |
| 444 | stable_record: dict[str, Any] = { |
| 445 | "schema_version": PROMOTION_TRANSACTION_SCHEMA_VERSION, |
| 446 | "week": week, |
| 447 | "run_id": manifest.get("run_id"), |
| 448 | "source_manifest": { |
| 449 | "path": manifest_relative, |
| 450 | "sha256": _sha256_file(manifest_path), |
| 451 | }, |
| 452 | "candidate": { |
| 453 | "summary_path": _relative_to_root(root, candidate_summary), |
| 454 | "summary_sha256": summary_sha, |
| 455 | "content_path": _relative_to_root(root, candidate_content), |
| 456 | "content_sha256": content_sha, |
| 457 | }, |
| 458 | "published_artifacts": [ |
| 459 | { |
| 460 | "role": "analysis_summary", |
| 461 | "path": _relative_to_root(root, canonical_summary), |
| 462 | "source_path": _relative_to_root(root, candidate_summary), |
| 463 | "sha256": summary_sha, |
| 464 | }, |
| 465 | { |
| 466 | "role": "hugo_content", |
| 467 | "path": _relative_to_root(root, canonical_content), |
| 468 | "source_path": _relative_to_root(root, candidate_content), |
| 469 | "sha256": content_sha, |
| 470 | }, |
| 471 | ], |
| 472 | "provenance": { |
| 473 | "source_artifacts": manifest.get("source_artifacts", []), |
| 474 | "analysis": manifest.get("analysis") or manifest.get("ai_provenance"), |
| 475 | "validation": manifest.get("validation") |
| 476 | or {"gate_results": manifest.get("gate_results")}, |
| 477 | "promotion": manifest.get("promotion") |
| 478 | or {"eligible": manifest.get("promotion_eligible")}, |
| 479 | }, |
| 480 | } |
| 481 | transaction_payload = json.dumps(stable_record, sort_keys=True, separators=(",", ":")).encode( |
| 482 | "utf-8" |
| 483 | ) |
| 484 | stable_record["transaction_id"] = hashlib.sha256(transaction_payload).hexdigest() |
| 485 | return stable_record |
| 486 | |
| 487 | |
| 488 | def _write_transactionally(targets: list[tuple[Path, bytes]]) -> None: |
| 489 | originals: list[tuple[Path, bool, bytes | None]] = [] |
| 490 | written: list[Path] = [] |
| 491 | temp_paths: list[Path] = [] |
| 492 | for target, _ in targets: |
| 493 | originals.append( |
| 494 | (target, target.exists(), target.read_bytes() if target.exists() else None) |
| 495 | ) |
| 496 | target.parent.mkdir(parents=True, exist_ok=True) |
| 497 | |
| 498 | try: |
| 499 | for index, (target, payload) in enumerate(targets): |
| 500 | tmp = target.with_name(f".{target.name}.promotion-{os.getpid()}-{index}.tmp") |
| 501 | temp_paths.append(tmp) |
| 502 | tmp.write_bytes(payload) |
| 503 | tmp.replace(target) |
| 504 | written.append(target) |
| 505 | except Exception: |
| 506 | for tmp in temp_paths: |
| 507 | tmp.unlink(missing_ok=True) |
| 508 | for target, existed, payload in reversed(originals): |
| 509 | if existed and payload is not None: |
| 510 | target.parent.mkdir(parents=True, exist_ok=True) |
| 511 | target.write_bytes(payload) |
| 512 | elif target in written or target.exists(): |
| 513 | target.unlink(missing_ok=True) |
| 514 | raise |
| 515 | |
| 516 | |
| 517 | def promote_candidate(manifest_path: Path, *, root: Path | None = None) -> tuple[Path, Path]: |
| 518 | workspace = (root or Path.cwd()).resolve() |
| 519 | resolved_manifest_path = ( |
| 520 | manifest_path if manifest_path.is_absolute() else workspace / manifest_path |
| 521 | ) |
| 522 | manifest: dict[str, Any] | None = None |
| 523 | week = "unknown-week" |
| 524 | |
| 525 | try: |
| 526 | manifest = _load_manifest(resolved_manifest_path) |
| 527 | week, candidate_summary, candidate_content, reasons = _validate_manifest( |
| 528 | manifest, workspace, resolved_manifest_path |
| 529 | ) |
| 530 | if reasons: |
| 531 | raise PromotionBlocked(reasons) |
| 532 | except PromotionBlocked as exc: |
| 533 | _write_diagnostic(workspace, week, manifest, exc.reasons) |
| 534 | raise |
| 535 | |
| 536 | canonical_summary = workspace / "data" / "analyzed" / f"{week}-summary.md" |
| 537 | match = WEEK_PATTERN.fullmatch(week) |
| 538 | if match is None: |
| 539 | raise PromotionBlocked(["week must use YYYY-WNN format."]) |
| 540 | year, week_number = match.group("year"), match.group("week") |
| 541 | canonical_content = workspace / "content" / "weekly" / year / f"W{week_number}.md" |
| 542 | transaction_manifest = workspace / "data" / "published" / week / "promotion-manifest.json" |
| 543 | _write_force_audit(workspace, week, manifest) |
| 544 | transaction_record = _promotion_transaction_record( |
| 545 | root=workspace, |
| 546 | week=week, |
| 547 | manifest_path=resolved_manifest_path, |
| 548 | manifest=manifest, |
| 549 | candidate_summary=candidate_summary, |
| 550 | candidate_content=candidate_content, |
| 551 | canonical_summary=canonical_summary, |
| 552 | canonical_content=canonical_content, |
| 553 | ) |
| 554 | _write_transactionally( |
| 555 | [ |
| 556 | (canonical_summary, candidate_summary.read_bytes()), |
| 557 | (canonical_content, candidate_content.read_bytes()), |
| 558 | ( |
| 559 | transaction_manifest, |
| 560 | (json.dumps(transaction_record, indent=2, sort_keys=True) + "\n").encode("utf-8"), |
| 561 | ), |
| 562 | ] |
| 563 | ) |
| 564 | return canonical_summary, canonical_content |
| 565 | |
| 566 | |
| 567 | def main(argv: list[str] | None = None) -> int: |
| 568 | args = parse_args(argv) |
| 569 | try: |
| 570 | summary_path, content_path = promote_candidate(args.manifest, root=args.root) |
| 571 | except PromotionBlocked as exc: |
| 572 | for reason in exc.reasons: |
| 573 | print(f"::error::{reason}") |
| 574 | return 1 |
| 575 | print(f"Promoted {summary_path} and {content_path}") |
| 576 | return 0 |
| 577 | |
| 578 | |
| 579 | if __name__ == "__main__": |
| 580 | raise SystemExit(main()) |