#!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any DEFAULT_SYNC_CONFIG = "configs/hourly_delta_sync.json" DEFAULT_SUMMARY_MANIFEST = "manifest_phase1_source_snapshot.json" DEFAULT_BASE_MANIFEST = "manifest.json" DEFAULT_WINDOW_HOURS = 1 DEFAULT_REWIND_HOURS = 2 DEFAULT_RUN_LABEL_PREFIX = "hourly-delta" DEFAULT_S3_PREFIX = "salesforce/hourly" def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Backfill Salesforce hourly delta exports to local disk and S3 from an existing baseline snapshot." ) parser.add_argument("--config", required=True, help="Path to readonly Salesforce config file.") parser.add_argument( "--sync-config", default=str(script_dir / DEFAULT_SYNC_CONFIG), help="Path to the hourly delta sync config.", ) parser.add_argument( "--baseline-dir", required=True, help="Existing baseline snapshot directory, for example /tmp/sf-export/2026-05-17/full-phase1-source-access.", ) parser.add_argument( "--baseline-summary-manifest", default=DEFAULT_SUMMARY_MANIFEST, help="Summary manifest file name inside baseline-dir.", ) parser.add_argument( "--baseline-base-manifest", default=DEFAULT_BASE_MANIFEST, help="Fallback base manifest file name inside baseline-dir.", ) parser.add_argument("--output-root", required=True, help="Root output directory for hourly snapshots.") parser.add_argument("--start", help="Explicit UTC ISO-8601 start. Overrides baseline rewind logic.") parser.add_argument("--until", help="Explicit UTC ISO-8601 end. Defaults to current UTC hour.") parser.add_argument("--rewind-hours", type=int, default=DEFAULT_REWIND_HOURS, help="Safety rewind applied before the baseline end.") parser.add_argument("--window-hours", type=int, default=DEFAULT_WINDOW_HOURS, help="Window size in hours.") parser.add_argument("--run-label-prefix", default=DEFAULT_RUN_LABEL_PREFIX, help="Prefix for each hourly run label.") parser.add_argument("--s3-bucket", help="Optional S3 bucket for export upload.") parser.add_argument("--s3-prefix", default=DEFAULT_S3_PREFIX, help="S3 prefix root for hourly export upload.") parser.add_argument("--skip-s3-upload", action="store_true", help="Skip S3 upload even when bucket is set.") parser.add_argument("--max-windows", type=int, help="Stop after N hourly windows.") parser.add_argument("--dry-run", action="store_true", help="Print the planned windows without executing exports.") return parser.parse_args() def load_json(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must be a JSON object") return payload def write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2), encoding="utf-8") def parse_utc_timestamp(value: str | None) -> datetime | None: if not value: return None raw = value.strip() if not raw: return None normalized = raw.replace("Z", "+00:00") dt = datetime.fromisoformat(normalized) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def format_sf_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def floor_to_hour(value: datetime) -> datetime: return value.astimezone(timezone.utc).replace(minute=0, second=0, microsecond=0) def infer_baseline_end(baseline_dir: Path, summary_manifest_name: str, base_manifest_name: str) -> datetime: summary_path = baseline_dir / summary_manifest_name if summary_path.exists(): summary = load_json(summary_path) summary_ts = parse_utc_timestamp(str(summary.get("snapshot_time", "")).strip()) if summary_ts: return summary_ts base_path = baseline_dir / base_manifest_name if not base_path.exists(): raise FileNotFoundError(f"baseline manifest not found: {summary_path} or {base_path}") base_manifest = load_json(base_path) base_ts = parse_utc_timestamp(str(base_manifest.get("snapshot_time", "")).strip()) if not base_ts: raise ValueError(f"baseline manifest missing snapshot_time: {base_path}") return base_ts def build_windows(start: datetime, end: datetime, window_hours: int, max_windows: int | None) -> list[tuple[datetime, datetime]]: if end <= start: raise ValueError(f"end must be after start: start={start} end={end}") if window_hours <= 0: raise ValueError("window_hours must be positive") windows: list[tuple[datetime, datetime]] = [] cursor = start step = timedelta(hours=window_hours) while cursor < end: if max_windows is not None and len(windows) >= max_windows: break next_cursor = min(cursor + step, end) windows.append((cursor, next_cursor)) cursor = next_cursor return windows def main() -> int: args = parse_args() script_dir = Path(__file__).resolve().parent baseline_dir = Path(args.baseline_dir).expanduser().resolve() if not baseline_dir.exists(): raise FileNotFoundError(f"baseline dir does not exist: {baseline_dir}") baseline_end = infer_baseline_end( baseline_dir=baseline_dir, summary_manifest_name=args.baseline_summary_manifest, base_manifest_name=args.baseline_base_manifest, ) computed_start = floor_to_hour(baseline_end - timedelta(hours=args.rewind_hours)) start = parse_utc_timestamp(args.start) or computed_start until = parse_utc_timestamp(args.until) or floor_to_hour(datetime.now(timezone.utc)) start = floor_to_hour(start) until = floor_to_hour(until) windows = build_windows( start=start, end=until, window_hours=args.window_hours, max_windows=args.max_windows, ) plan = { "manifest_version": 1, "run_type": "hourly_delta_backfill", "baseline_dir": str(baseline_dir), "baseline_end": format_sf_timestamp(baseline_end), "rewind_hours": args.rewind_hours, "computed_start": format_sf_timestamp(computed_start), "effective_start": format_sf_timestamp(start), "effective_until": format_sf_timestamp(until), "window_hours": args.window_hours, "window_count": len(windows), "s3_bucket": args.s3_bucket, "s3_prefix": args.s3_prefix, "dry_run": args.dry_run, "windows": [ { "since": format_sf_timestamp(window_start), "until": format_sf_timestamp(window_end), "snapshot_date": window_start.date().isoformat(), "run_label": f"{args.run_label_prefix}-{window_start.strftime('%Y%m%dT%H00Z')}", } for window_start, window_end in windows ], } plan_path = Path(args.output_root).expanduser().resolve() / "_plans" / ( f"hourly_delta_backfill_{start.strftime('%Y%m%dT%H%M%SZ')}_{until.strftime('%Y%m%dT%H%M%SZ')}.json" ) write_json(plan_path, plan) print(f"[ok] plan={plan_path}") if args.dry_run: print(json.dumps(plan, indent=2)) return 0 sync_script = script_dir / "run_hourly_delta_sync.py" for window_start, window_end in windows: run_label = f"{args.run_label_prefix}-{window_start.strftime('%Y%m%dT%H00Z')}" cmd = [ "python3", str(sync_script), "--config", args.config, "--sync-config", args.sync_config, "--output-root", args.output_root, "--snapshot-date", window_start.date().isoformat(), "--run-label", run_label, "--since", format_sf_timestamp(window_start), "--until", format_sf_timestamp(window_end), "--export-only", "--skip-state-update", ] if args.s3_bucket: cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: cmd.append("--skip-s3-upload") subprocess.run(cmd, check=True) return 0 if __name__ == "__main__": raise SystemExit(main())