#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os from pathlib import Path from typing import Any import run_hourly_delta_sync as hourly_delta_sync from load_salesforce_snapshot_to_tidb import load_person_identity_authority_policy DEFAULT_MANIFEST_NAME = "manifest.json" DEFAULT_BATCH_SIZE = 200 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 parse_objects(values: list[str] | None) -> set[str]: result: set[str] = set() for raw in values or []: result.update(item.strip() for item in str(raw).split(",") if item.strip()) return result def resolve_person_identity_snapshot( *, snapshot_dir: Path, mysql_client: str, db: hourly_delta_sync.DbConfig, manifest_name: str = DEFAULT_MANIFEST_NAME, objects: set[str] | None = None, expected_identity_field: str | None = None, require_requested_objects: bool = False, batch_size: int = DEFAULT_BATCH_SIZE, policy_path: Path | None = None, ) -> dict[str, Any]: manifest_path = snapshot_dir / manifest_name manifest = load_json(manifest_path) policy = load_person_identity_authority_policy(policy_path) identity_field = str(policy["identity_field"]).strip() if expected_identity_field and expected_identity_field != identity_field: raise ValueError( "person identity requirement identity_field does not match the canonical policy" ) policy_objects = { str(item).strip() for item in policy.get("objects", []) if str(item).strip() } requested_objects = set(objects or policy_objects) if require_requested_objects: unsupported = sorted(requested_objects - policy_objects) if unsupported: raise ValueError( "requested object(s) are not covered by the person identity policy: " + ", ".join(unsupported) ) manifest_object_map = { str(item.get("object") or "").strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object") or "").strip() } selected_objects = sorted( policy_objects & requested_objects & set(manifest_object_map) ) if require_requested_objects: missing = sorted((requested_objects & policy_objects) - set(manifest_object_map)) if missing: raise ValueError( "requested person object(s) are missing from the snapshot manifest: " + ", ".join(missing) ) if not selected_objects: return { "enabled": True, "status": "skipped", "reason": "no_person_objects_selected", "objects": [], } object_config: dict[str, dict[str, str]] = {} for object_name in selected_objects: target_table = str( manifest_object_map[object_name].get("target_table") or "" ).strip() if not target_table: raise ValueError( f"{object_name} manifest entry requires target_table before identity resolution" ) object_config[object_name] = {"target_table": target_table} result = hourly_delta_sync.apply_person_identity_resolution_to_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, config={ "enabled": True, "identity_field": identity_field, "batch_size": max(1, int(batch_size)), "objects": object_config, }, ) completed_manifest = load_json(manifest_path) marker = completed_manifest.get("person_identity_resolution") if not isinstance(marker, dict) or marker.get("status") != "completed": raise ValueError("person identity resolution did not write a completed marker") if str(marker.get("identity_field") or "").strip() != identity_field: raise ValueError("person identity resolution marker field mismatch") if marker.get("source_org_kind") != "salesforce_organization_id": raise ValueError( "person identity resolution marker must use the Salesforce Organization.Id" ) marker_objects = { str(item).strip() for item in marker.get("objects", []) if str(item).strip() } if not set(selected_objects).issubset(marker_objects): raise ValueError("person identity resolution marker is missing selected objects") return result def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Resolve provider-scoped Salesforce person identities in a snapshot " "before schema mutation or load." ) ) parser.add_argument("--snapshot-dir", required=True) parser.add_argument("--manifest-name", default=DEFAULT_MANIFEST_NAME) parser.add_argument( "--objects", action="append", default=[], help="Optional object allowlist; repeat or pass comma-separated values.", ) parser.add_argument("--person-identity-policy") parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE) parser.add_argument("--mysql-client", default=hourly_delta_sync.MYSQL_CLIENT) parser.add_argument( "--db-host", default=os.environ.get("TIDB_PROD_HOST", os.environ.get("TIDB_HOST", "")), ) parser.add_argument( "--db-port", type=int, default=int(os.environ.get("TIDB_PROD_PORT", os.environ.get("TIDB_PORT", "4000"))), ) parser.add_argument( "--db-user", default=os.environ.get("TIDB_PROD_USER", os.environ.get("TIDB_USER", "")), ) parser.add_argument( "--db-password", default=os.environ.get("TIDB_PROD_PASSWORD", os.environ.get("TIDB_PASSWORD", "")), ) parser.add_argument( "--db-database", default=os.environ.get("TIDB_PROD_DATABASE", os.environ.get("TIDB_DATABASE", "")), ) parser.add_argument("--ssl-mode", default=os.environ.get("TIDB_SSL_MODE", "REQUIRED")) return parser.parse_args() def main() -> int: args = parse_args() if not args.mysql_client: raise RuntimeError("mysql client is required for person identity resolution") db = hourly_delta_sync.DbConfig( host=str(args.db_host).strip(), port=int(args.db_port), user=str(args.db_user).strip(), password=str(args.db_password), database=str(args.db_database).strip(), ssl_mode=str(args.ssl_mode).strip(), ) if not all([db.host, db.user, db.database]): raise ValueError("db host/user/database are required for person identity resolution") result = resolve_person_identity_snapshot( snapshot_dir=Path(args.snapshot_dir).expanduser().resolve(), manifest_name=args.manifest_name, objects=parse_objects(args.objects) or None, mysql_client=args.mysql_client, db=db, batch_size=args.batch_size, policy_path=( Path(args.person_identity_policy).expanduser().resolve() if args.person_identity_policy else None ), ) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())