#!/usr/bin/env python3 """Apply the shared deploy baseline through its existing bulk/REST engine.""" from __future__ import annotations import argparse import json import os from pathlib import Path import subprocess import sys import time from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPT_DIR.parent RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime" if str(RUNTIME_DIR) not in sys.path: sys.path.insert(0, str(RUNTIME_DIR)) import deploy_offline_env # noqa: E402 SUMMARY_FILENAME = "summary.json" class DeployBaselineAdapterError(RuntimeError): """Raised when the reviewed deploy baseline adapter contract is violated.""" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Apply the shared deploy baseline Seed module.") parser.add_argument("--source-root", required=True) parser.add_argument("--seed-file", required=True) parser.add_argument("--output-dir", required=True) parser.add_argument("--target", required=True, choices=("local", "test", "exp", "staging")) parser.add_argument("--dry-run", action="store_true") return parser.parse_args(argv) def _read_json(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise DeployBaselineAdapterError(f"invalid bulk baseline summary: {path}") from exc if not isinstance(payload, dict): raise DeployBaselineAdapterError("bulk baseline summary must be an object") return payload def _write_json(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _baseline_objects(bulk_summary: dict[str, Any]) -> list[str]: objects: list[str] = [] for value in bulk_summary.get("bulk_objects") or []: if isinstance(value, str) and value and value not in objects: objects.append(value) for bucket in ("fallback_objects", "unsupported"): for value in bulk_summary.get(bucket) or []: object_api_name = value.get("object") if isinstance(value, dict) else None if isinstance(object_api_name, str) and object_api_name and object_api_name not in objects: objects.append(object_api_name) return objects def normalize_summary( bulk_summary: dict[str, Any], *, dry_run: bool, payload: dict[str, object] | None, bulk_record_count: int, duration_ms: int, ) -> dict[str, Any]: total_alias_count = int(bulk_summary.get("total_alias_count", 0)) collapsed_duplicate_records = int(bulk_summary.get("collapsed_duplicate_records", 0)) baseline_record_count = total_alias_count - collapsed_duplicate_records fallback_record_count = baseline_record_count - bulk_record_count bulk_coverage_complete = bulk_summary.get("coverage_complete") is True if baseline_record_count < 0 or fallback_record_count != 0: raise DeployBaselineAdapterError("bulk baseline counts are inconsistent") if not bulk_coverage_complete: raise DeployBaselineAdapterError( "bulk baseline coverage is incomplete; REST fallback is retired" ) materialized_record_count = 0 failed_record_count = 0 if payload is not None: materialized_record_count = len(payload.get("applied_records", []) or []) + len( payload.get("hydrated_skipped_records", []) or [] ) failed_record_count = len(payload.get("failed_records", []) or []) coverage_complete = ( bulk_coverage_complete if dry_run else failed_record_count == 0 and materialized_record_count == baseline_record_count ) return { "apply_engine": ( "bulk-plan" if dry_run else "bulk" ), "baseline_alias_count": total_alias_count, "baseline_objects": _baseline_objects(bulk_summary), "baseline_record_count": baseline_record_count, "bulk_record_count": bulk_record_count, "collapsed_duplicate_records": collapsed_duplicate_records, "coverage_complete": coverage_complete, "dry_run": dry_run, "duration_ms": duration_ms, "fallback_record_count": fallback_record_count, "fallback_used": False, "materialized_record_count": materialized_record_count, } def _dry_run_bulk(source_root: Path, seed_file: Path, output_dir: Path) -> dict[str, Any]: bulk_output_dir = output_dir / "bulk" state_file = bulk_output_dir / "runtime-state.json" summary_file = bulk_output_dir / "summary.json" completed = subprocess.run( [ "bash", str(source_root / "scripts" / "apply_exp_bulk_baseline.sh"), "--seed-file", str(seed_file), "--output-dir", str(bulk_output_dir), "--state-file", str(state_file), "--summary-file", str(summary_file), "--dry-run", ], cwd=source_root, env={**os.environ, "EXP_BULK_HASH_USER_PASSWORDS": "0"}, check=False, ) if completed.returncode != 0: raise DeployBaselineAdapterError( f"bulk baseline dry-run failed: exit={completed.returncode}" ) return _read_json(summary_file) def run_adapter(args: argparse.Namespace) -> tuple[dict[str, Any], Path]: started = time.perf_counter() source_root = Path(args.source_root).expanduser().resolve() seed_file = Path(args.seed_file).expanduser().resolve() expected_seed_file = ( source_root / "deployments" / "offline" / "runtime" / "seeds" / "deploy-baseline.seed.yaml" ) if seed_file != expected_seed_file or not seed_file.is_file(): raise DeployBaselineAdapterError("seed file is not the reviewed deploy baseline source") output_dir = Path(args.output_dir).expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=False) environment = dict(os.environ) payload: dict[str, object] | None = None if args.dry_run: bulk_summary = _dry_run_bulk(source_root, seed_file, output_dir) applied_bulk_record_count = int(bulk_summary.get("bulk_record_count", 0)) else: nexuscrm_worktree_raw = environment.get("NEXUSCRM_WORKTREE", "").strip() if not nexuscrm_worktree_raw: raise DeployBaselineAdapterError("NEXUSCRM_WORKTREE is required in apply mode") nexuscrm_worktree = Path(nexuscrm_worktree_raw).expanduser().resolve() if not nexuscrm_worktree.is_dir(): raise DeployBaselineAdapterError( f"NEXUSCRM_WORKTREE is not a directory: {nexuscrm_worktree}" ) payload, bulk_result = deploy_offline_env.apply_shared_deploy_baseline_seed( environment, source_root, nexuscrm_worktree, ) bulk_summary = bulk_result.summary applied_bulk_record_count = bulk_result.records if not bulk_summary.get("total_alias_count"): bulk_summary = _dry_run_bulk(source_root, seed_file, output_dir) summary = normalize_summary( bulk_summary, dry_run=args.dry_run, payload=payload, bulk_record_count=applied_bulk_record_count, duration_ms=round((time.perf_counter() - started) * 1000), ) summary_path = output_dir / SUMMARY_FILENAME _write_json(summary_path, summary) print(json.dumps(summary, indent=2, sort_keys=True)) return summary, summary_path def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: run_adapter(args) except DeployBaselineAdapterError as exc: print(f"deploy baseline adapter failed: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())