#!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any DEFAULT_BASE_MANIFEST = "manifest_phase1_source_snapshot.json" def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Run the phase-1 Salesforce source snapshot pipeline: base export + Document + AcceptedEventRelation + TaskWhoRelation + ContentDocumentLink + ContentVersion." ) parser.add_argument("--config", required=True, help="Path to readonly Salesforce config file.") parser.add_argument( "--export-config", default=str(script_dir / "configs" / "daily_full_export.json"), help="Path to export JSON config.", ) parser.add_argument("--output-root", required=True, help="Root output directory for dated snapshots.") parser.add_argument("--profile", default="full_phase1_source_snapshot", help="Base export profile.") parser.add_argument("--snapshot-date", required=True, help="Snapshot date in YYYY-MM-DD.") parser.add_argument("--run-label", required=True, help="Run label under snapshot date.") parser.add_argument("--s3-bucket", help="Optional S3 bucket.") parser.add_argument("--s3-prefix", help="Optional S3 prefix.") parser.add_argument("--chunk-size-cdl", type=int, default=200, help="ContentDocumentLink chunk size.") parser.add_argument("--chunk-size-cv", type=int, default=100, help="ContentVersion chunk size.") parser.add_argument( "--content-document-link-linked-entity-file", action="append", default=["Order.csv.gz", "Contract.csv.gz"], help=( "CSV.gz file inside the snapshot whose Id values should seed " "ContentDocumentLink.LinkedEntityId export queries. May be repeated." ), ) parser.add_argument("--skip-s3-upload", action="store_true", help="Skip all S3 uploads.") parser.add_argument("--skip-document", action="store_true", help="Skip Document export.") parser.add_argument("--skip-document-blob", action="store_true", help="Skip Document body upload.") parser.add_argument("--skip-accepted-event-relation", action="store_true", help="Skip AcceptedEventRelation export.") parser.add_argument("--skip-task-who-relation", action="store_true", help="Skip TaskWhoRelation export.") parser.add_argument("--skip-content-document-link", action="store_true", help="Skip ContentDocumentLink export.") parser.add_argument("--skip-content-version", action="store_true", help="Skip ContentVersion export.") parser.add_argument( "--skip-content-version-blob", action="store_true", help="Deprecated fail-closed flag; use --skip-content-version instead.", ) parser.add_argument("--skip-sharing-rules", action="store_true", help="Skip SharingRules metadata export.") parser.add_argument( "--summary-manifest-name", default=DEFAULT_BASE_MANIFEST, help="Summary manifest written into the snapshot directory.", ) return parser.parse_args() def run_command(cmd: list[str]) -> None: subprocess.run(cmd, check=True) 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 main() -> int: args = parse_args() if args.skip_content_version_blob and not args.skip_content_version: raise RuntimeError( "ContentVersion metadata-only manifests are forbidden; use " "--skip-content-version to omit the whole layer" ) script_dir = Path(__file__).resolve().parent snapshot_dir = Path(args.output_root).expanduser().resolve() / args.snapshot_date / args.run_label snapshot_dir.mkdir(parents=True, exist_ok=True) base_cmd = [ "python3", str(script_dir / "export_salesforce_snapshot.py"), "--config", str(Path(args.config).expanduser().resolve()), "--export-config", str(Path(args.export_config).expanduser().resolve()), "--output-root", str(Path(args.output_root).expanduser().resolve()), "--profile", args.profile, "--snapshot-date", args.snapshot_date, "--run-label", args.run_label, ] if args.s3_bucket: base_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: base_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: base_cmd.append("--skip-s3-upload") run_command(base_cmd) document_manifest = None if not args.skip_document: document_cmd = [ "python3", str(script_dir / "export_documents.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), ] if args.s3_bucket: document_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: document_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: document_cmd.append("--skip-s3-upload") if args.skip_document_blob: document_cmd.append("--skip-blob-upload") run_command(document_cmd) document_manifest = load_json(snapshot_dir / "manifest_document.json") accepted_event_manifest = None if not args.skip_accepted_event_relation: accepted_event_cmd = [ "python3", str(script_dir / "export_accepted_event_relations.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), ] if args.s3_bucket: accepted_event_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: accepted_event_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: accepted_event_cmd.append("--skip-s3-upload") run_command(accepted_event_cmd) accepted_event_manifest = load_json(snapshot_dir / "manifest_accepted_event_relation.json") task_who_manifest = None if not args.skip_task_who_relation: task_who_cmd = [ "python3", str(script_dir / "export_task_who_relations.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), ] if args.s3_bucket: task_who_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: task_who_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: task_who_cmd.append("--skip-s3-upload") run_command(task_who_cmd) task_who_manifest = load_json(snapshot_dir / "manifest_task_who_relation.json") cdl_manifest = None if not args.skip_content_document_link: cdl_cmd = [ "python3", str(script_dir / "export_content_document_links.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), "--chunk-size", str(args.chunk_size_cdl), ] for file_name in args.content_document_link_linked_entity_file: cdl_cmd.extend(["--linked-entity-file", file_name]) if args.s3_bucket: cdl_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: cdl_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: cdl_cmd.append("--skip-s3-upload") run_command(cdl_cmd) cdl_manifest = load_json(snapshot_dir / "manifest_content_document_link.json") cv_manifest = None if not args.skip_content_version: cv_cmd = [ "python3", str(script_dir / "export_content_versions.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), "--chunk-size", str(args.chunk_size_cv), ] if args.s3_bucket: cv_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: cv_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: cv_cmd.append("--skip-s3-upload") run_command(cv_cmd) cv_manifest = load_json(snapshot_dir / "manifest_content_version.json") sharing_rules_manifest = None if not args.skip_sharing_rules: sharing_rules_cmd = [ "python3", str(script_dir / "export_sharing_rules_metadata.py"), "--config", str(Path(args.config).expanduser().resolve()), "--snapshot-dir", str(snapshot_dir), ] if args.s3_bucket: sharing_rules_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: sharing_rules_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: sharing_rules_cmd.append("--skip-s3-upload") run_command(sharing_rules_cmd) sharing_rules_manifest = load_json(snapshot_dir / "manifest_sharing_rules.json") base_manifest = load_json(snapshot_dir / "manifest.json") summary = { "manifest_version": 1, "run_type": "phase1_source_snapshot", "snapshot_time": utc_now(), "snapshot_date": args.snapshot_date, "run_label": args.run_label, "snapshot_dir": str(snapshot_dir), "profile": args.profile, "base_manifest": "manifest.json", "document_manifest": "manifest_document.json" if document_manifest else None, "accepted_event_relation_manifest": "manifest_accepted_event_relation.json" if accepted_event_manifest else None, "task_who_relation_manifest": "manifest_task_who_relation.json" if task_who_manifest else None, "content_document_link_manifest": "manifest_content_document_link.json" if cdl_manifest else None, "content_version_manifest": "manifest_content_version.json" if cv_manifest else None, "sharing_rules_manifest": "manifest_sharing_rules.json" if sharing_rules_manifest else None, "objects": { "base_object_count": base_manifest.get("object_count"), "document_record_count": None if not document_manifest else document_manifest.get("record_count"), "document_blob_uploaded": None if not document_manifest else document_manifest.get("blob_uploaded"), "accepted_event_relation_record_count": None if not accepted_event_manifest else accepted_event_manifest.get("record_count"), "task_who_relation_record_count": None if not task_who_manifest else task_who_manifest.get("record_count"), "content_document_link_record_count": None if not cdl_manifest else cdl_manifest.get("record_count"), "content_version_record_count": None if not cv_manifest else cv_manifest.get("record_count"), "content_version_blob_uploaded": None if not cv_manifest else cv_manifest.get("blob_uploaded"), "content_version_blob_skipped": None if not cv_manifest else cv_manifest.get("blob_skipped"), "content_version_blob_failed": None if not cv_manifest else cv_manifest.get("blob_failed"), "sharing_rules_object_count": None if not sharing_rules_manifest else len(sharing_rules_manifest.get("objects_extracted") or []), }, } summary_path = snapshot_dir / args.summary_manifest_name summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8") print(f"[ok] summary_manifest={summary_path}") return 0 if __name__ == "__main__": raise SystemExit(main())