#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import gzip import hashlib import json import os import shutil import subprocess import sys from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from incremental_sync_state import atomic_write_json, persist_replay_intent DEFAULT_SYNC_CONFIG = "configs/hourly_delta_sync.json" DEFAULT_EXPORT_CONFIG = "configs/daily_full_export.json" DEFAULT_LOAD_PLAN_CONFIG = "configs/tidb_load_business_phase1.json" DEFAULT_MANIFEST_NAME = "manifest.json" DEFAULT_SUMMARY_NAME = "hourly_delta_sync.json" DEFAULT_PROFILE = "full_phase1" MYSQL_CLIENT_CANDIDATES = [ shutil.which("mysql"), "/opt/homebrew/opt/mysql-client@8.4/bin/mysql", ] MYSQL_SSL_MODE_SUPPORT_CACHE: dict[str, bool] = {} def first_existing(items: list[str | None]) -> str | None: for item in items: if item and Path(item).exists(): return item return None MYSQL_CLIENT = first_existing(MYSQL_CLIENT_CANDIDATES) @dataclass class DbConfig: host: str port: int user: str password: str database: str ssl_mode: str def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Export, load, and verify a Salesforce delta window." ) 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( "--export-config", default=str(script_dir / DEFAULT_EXPORT_CONFIG), help="Path to the base export JSON config.", ) parser.add_argument( "--load-plan-config", default=str(script_dir / DEFAULT_LOAD_PLAN_CONFIG), help="Path to the base TiDB load plan JSON config.", ) parser.add_argument("--output-root", required=True, help="Root output directory for dated snapshots.") parser.add_argument("--snapshot-date", help="Snapshot date in YYYY-MM-DD. Defaults to window start UTC date.") parser.add_argument("--run-label", default="hourly-delta", help="Run label under the snapshot date.") parser.add_argument("--since", help="Explicit lower bound watermark in UTC ISO-8601.") parser.add_argument("--until", help="Explicit upper bound watermark in UTC ISO-8601.") parser.add_argument("--skip-export", action="store_true", help="Skip Salesforce export.") parser.add_argument("--skip-load", action="store_true", help="Skip TiDB load.") parser.add_argument( "--dry-run-load", action="store_true", help=( "Run DB-backed identity resolution and write-plan validation without " "calling the loader, replay, source-target checks, or state updates." ), ) parser.add_argument("--skip-check", action="store_true", help="Skip source-vs-target checks.") parser.add_argument( "--export-only", action="store_true", help="Export only. Equivalent to --skip-load --skip-check.", ) parser.add_argument("--s3-bucket", help="Optional S3 bucket for export upload.") parser.add_argument("--s3-prefix", help="Optional S3 prefix for export upload.") parser.add_argument("--skip-s3-upload", action="store_true", help="Skip export upload even when bucket is set.") parser.add_argument("--manifest-name", default=DEFAULT_MANIFEST_NAME, help="Snapshot manifest file name.") parser.add_argument("--summary-name", default=DEFAULT_SUMMARY_NAME, help="Hourly sync summary file name.") parser.add_argument("--mysql-client", help="Path to mysql client.") parser.add_argument("--state-file", help="Override state file path.") parser.add_argument("--skip-state-update", action="store_true", help="Do not persist watermark state after success.") parser.add_argument( "--skip-replay", action="store_true", help="Compute replay candidates but do not execute target runtime replay.", ) parser.add_argument( "--replay-queue-file", help=( "Durable replay queue used to persist deferred Lead assignment intent " "before the first database write." ), ) 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")) parser.add_argument("--target-base-url", default=os.environ.get("NEXUSCRM_TARGET_BASE_URL", "")) parser.add_argument("--target-username", default=os.environ.get("NEXUSCRM_TARGET_USERNAME", "")) parser.add_argument("--target-password", default=os.environ.get("NEXUSCRM_TARGET_PASSWORD", "")) 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: atomic_write_json(path, payload) def utc_now() -> datetime: return datetime.now(timezone.utc) def utc_today() -> str: return utc_now().date().isoformat() 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 format_mysql_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).strftime("%Y-%m-%d %H:%M:%S") def mysql_env(password: str) -> dict[str, str]: env = os.environ.copy() env["MYSQL_PWD"] = password return env def mysql_supports_ssl_mode(mysql_client: str) -> bool: cached = MYSQL_SSL_MODE_SUPPORT_CACHE.get(mysql_client) if cached is not None: return cached result = subprocess.run( [mysql_client, "--help"], capture_output=True, text=True, check=False, env=mysql_env(""), ) combined = f"{result.stdout}\n{result.stderr}" supported = "--ssl-mode" in combined MYSQL_SSL_MODE_SUPPORT_CACHE[mysql_client] = supported return supported def mysql_ssl_args(mysql_client: str, ssl_mode: str) -> list[str]: normalized = str(ssl_mode or "").strip().upper() if not normalized: return [] if mysql_supports_ssl_mode(mysql_client): return ["--ssl-mode", normalized] if normalized == "DISABLED": return [] return ["--ssl"] def run_mysql_query(mysql_client: str, db: DbConfig, sql: str) -> list[str]: cmd = [ mysql_client, "--host", db.host, "--port", str(db.port), "--user", db.user, "--database", db.database, "--batch", "--raw", "--skip-column-names", "-e", sql, ] cmd[10:10] = mysql_ssl_args(mysql_client, db.ssl_mode) result = subprocess.run(cmd, check=False, capture_output=True, text=True, env=mysql_env(db.password)) if result.returncode != 0: stderr = result.stderr.strip() or result.stdout.strip() or f"mysql exit code {result.returncode}" raise RuntimeError(f"mysql query failed: {stderr}; sql={sql}") return [line.rstrip("\n") for line in result.stdout.splitlines()] def sanitize_path_component(value: str) -> str: cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in value.strip()) return cleaned.strip("-") or "snapshot" def ensure_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) def build_export_object_map(export_config: dict[str, Any]) -> dict[str, dict[str, Any]]: result: dict[str, dict[str, Any]] = {} for item in export_config.get("objects", []): if not isinstance(item, dict): continue name = str(item.get("name", "")).strip() if name: result[name] = item return result def intersect_objects(export_profile: list[str], load_objects: list[str]) -> list[str]: load_set = {item.strip() for item in load_objects if str(item).strip()} result: list[str] = [] for name in export_profile: normalized = name.strip() if normalized and normalized in load_set: result.append(normalized) return result def combine_where(existing: str, watermark_field: str, start: str, end: str) -> str: # Salesforce SOQL expects dateTime literals here without surrounding quotes. window_clause = f"({watermark_field} >= {start} AND {watermark_field} < {end})" if not existing.strip(): return window_clause return f"({existing.strip()}) AND {window_clause}" def build_temp_export_config( export_config: dict[str, Any], selected_objects: list[str], window_start: str, window_end: str, watermark_field: str, fallback_watermark_field: str, ) -> dict[str, Any]: export_object_map = build_export_object_map(export_config) temp_objects: list[dict[str, Any]] = [] for name in selected_objects: raw = export_object_map.get(name) if not raw: continue item = dict(raw) chosen_field = str(item.get("incremental_watermark_field") or watermark_field or fallback_watermark_field).strip() item["incremental_watermark_field"] = chosen_field item["where"] = combine_where(str(item.get("where", "")), chosen_field, window_start, window_end) temp_objects.append(item) if not temp_objects: raise ValueError("no selected objects remain after export config filtering") profile_name = "hourly_delta_sync" temp_config: dict[str, Any] = dict(export_config) temp_config["default_profile"] = profile_name temp_config["profiles"] = {profile_name: [item["name"] for item in temp_objects]} temp_config["objects"] = temp_objects return temp_config def build_temp_load_plan(load_plan: dict[str, Any], selected_objects: list[str]) -> dict[str, Any]: temp_plan = dict(load_plan) temp_plan["include_objects"] = selected_objects skip_objects = [str(item).strip() for item in load_plan.get("skip_objects", []) if str(item).strip()] temp_plan["skip_objects"] = skip_objects return temp_plan def run_command(cmd: list[str], *, extra_env: dict[str, str] | None = None) -> None: command_env = os.environ.copy() if extra_env: command_env.update(extra_env) subprocess.run(cmd, check=True, env=command_env) def run_load_with_durable_replay_intent( *, load_cmd: list[str], load_env: dict[str, str], replay_deferred: bool, replay_queue_file: Path | None, window_start: str, window_end: str, description_field: str, replay_candidate_ids: list[str], ) -> dict[str, Any] | None: replay_entry: dict[str, Any] | None = None normalized_candidate_ids = sorted( {str(item).strip() for item in replay_candidate_ids if str(item).strip()} ) if replay_deferred and normalized_candidate_ids: if replay_queue_file is None: raise ValueError( "deferred Lead assignment replay requires --replay-queue-file " "before database loading" ) entry_id = f"{window_start}__{window_end}" replay_entry = persist_replay_intent( replay_queue_file, entry_id=entry_id, window_start=window_start, window_end=window_end, description_field=description_field, lead_ids=normalized_candidate_ids, updated_at=format_sf_timestamp(utc_now()), ) run_command(load_cmd, extra_env=load_env) if replay_entry is None: return None return { "status": str(replay_entry.get("status", "")), "entry_id": str(replay_entry.get("id", "")), "candidate_count": int(replay_entry.get("candidate_count", 0) or 0), "persisted_before_load": True, } def csv_max_timestamp( path: Path, column: str, *, excluded_ids: set[str] | None = None, ) -> datetime | None: if not path.exists(): return None excluded_ids = excluded_ids or set() max_value: datetime | None = None with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) for row in reader: if str(row.get("Id", "")).strip() in excluded_ids: continue raw = str(row.get(column, "")).strip() if not raw: continue parsed = parse_utc_timestamp(raw) if not parsed: continue if max_value is None or parsed > max_value: max_value = parsed return max_value def csv_distinct_key_count(path: Path, key_columns: list[str]) -> int: if not path.exists() or not key_columns: return 0 seen: set[tuple[str, ...]] = set() with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) for row in reader: key = tuple(str(row.get(column, "")).strip() for column in key_columns) if not any(key): continue seen.add(key) return len(seen) def csv_distinct_ids(path: Path) -> list[str]: if not path.exists(): return [] seen: set[str] = set() with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) if "Id" not in (reader.fieldnames or []): return [] for row in reader: row_id = str(row.get("Id", "")).strip() if row_id: seen.add(row_id) return sorted(seen) def csv_rows_by_fields(path: Path, field_names: list[str]) -> list[dict[str, str]]: if not path.exists() or not field_names: return [] rows: list[dict[str, str]] = [] with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) for row in reader: item = {name: str(row.get(name, "")).strip() for name in field_names} if not any(item.values()): continue rows.append(item) return rows def read_gzip_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: if not path.exists(): return [], [] with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) return list(reader.fieldnames or []), [dict(row) for row in reader] def write_gzip_csv(path: Path, fieldnames: list[str], rows: list[dict[str, str]]) -> None: with gzip.open(path, "wt", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in fieldnames}) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def refresh_manifest_sha256_input(manifest: dict[str, Any]) -> str: payload = { "snapshot_date": manifest.get("snapshot_date"), "run_label": manifest.get("run_label"), "objects": manifest.get("objects", []), "config_sha256": manifest.get("config_sha256"), "person_identity_resolution": manifest.get("person_identity_resolution"), } digest = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() manifest["manifest_sha256_input"] = digest return digest SALESFORCE_ID_SUFFIX_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" SALESFORCE_PERSON_ID_PREFIXES = {"Lead": "00Q", "Contact": "003"} DEFAULT_PERSON_IDENTITY_FIELD = "Salesforce_Identity_Key__c" CSV_NULL_SENTINEL = r"\N" def canonical_salesforce_id(value: str, *, expected_prefix: str | None = None) -> str: source_id = str(value or "").strip() if len(source_id) not in {15, 18} or not source_id.isalnum(): raise ValueError(f"invalid Salesforce id: {source_id!r}") if expected_prefix and not source_id.startswith(expected_prefix): raise ValueError( f"Salesforce id {source_id!r} does not have expected prefix {expected_prefix!r}" ) if len(source_id) == 18: return source_id suffix = "" for start in range(0, 15, 5): mask = 0 for offset, char in enumerate(source_id[start : start + 5]): if char.isupper(): mask |= 1 << offset suffix += SALESFORCE_ID_SUFFIX_ALPHABET[mask] return source_id + suffix def salesforce_person_identity_key( source_org_id: str, object_name: str, source_record_id: str, ) -> str: prefix = SALESFORCE_PERSON_ID_PREFIXES.get(object_name) if not prefix: raise ValueError(f"unsupported Salesforce person object: {object_name!r}") canonical_org_id = canonical_salesforce_id(source_org_id, expected_prefix="00D") canonical_record_id = canonical_salesforce_id(source_record_id, expected_prefix=prefix) return f"salesforce:{canonical_org_id}:{object_name}:{canonical_record_id}" def build_person_identity_resolution_plan( *, object_name: str, source_org_id: str, source_rows: list[dict[str, str]], target_rows: list[dict[str, str]], existing_target_ids: set[str], existing_target_identity_by_id: dict[str, str] | None = None, identity_field: str = DEFAULT_PERSON_IDENTITY_FIELD, id_field: str = "Id", ) -> dict[str, Any]: current_identity_by_id = existing_target_identity_by_id or {} identity_by_source_id: dict[str, str] = {} source_ids_by_identity: dict[str, set[str]] = {} conflicts: list[dict[str, Any]] = [] for row in source_rows: source_id = str(row.get(id_field, "")).strip() if not source_id: conflicts.append({"reason": "missing_source_id"}) continue try: identity_key = salesforce_person_identity_key( source_org_id, object_name, source_id, ) except ValueError as exc: conflicts.append( { "source_id": source_id, "reason": "invalid_source_identity", "detail": str(exc), } ) continue identity_by_source_id[source_id] = identity_key source_ids_by_identity.setdefault(identity_key, set()).add(source_id) duplicate_source_ids = { source_id for source_ids in source_ids_by_identity.values() if len(source_ids) > 1 for source_id in source_ids } for identity_key, source_ids in sorted(source_ids_by_identity.items()): if len(source_ids) > 1: conflicts.append( { "identity_key": identity_key, "source_ids": sorted(source_ids), "reason": "duplicate_source_provider_identity", } ) target_ids_by_identity: dict[str, set[str]] = {} for row in target_rows: target_id = str(row.get(id_field, "")).strip() identity_key = str(row.get(identity_field, "")).strip() if target_id and identity_key: target_ids_by_identity.setdefault(identity_key, set()).add(target_id) for identity_key, target_ids in sorted(target_ids_by_identity.items()): if len(target_ids) > 1: conflicts.append( { "identity_key": identity_key, "target_ids": sorted(target_ids), "reason": "duplicate_target_provider_identity", } ) id_remap: dict[str, str] = {} remapped: list[dict[str, Any]] = [] for source_id, identity_key in sorted(identity_by_source_id.items()): if source_id in duplicate_source_ids: continue current_identity = str(current_identity_by_id.get(source_id, "")).strip() if current_identity and current_identity != identity_key: conflicts.append( { "source_id": source_id, "identity_key": identity_key, "current_identity_key": current_identity, "reason": "source_id_owned_by_different_provider_identity", } ) continue target_ids = sorted(target_ids_by_identity.get(identity_key, set())) if len(target_ids) != 1: continue target_id = target_ids[0] if target_id == source_id: continue if source_id in existing_target_ids: conflicts.append( { "source_id": source_id, "identity_key": identity_key, "target_id": target_id, "reason": "source_id_and_provider_identity_owned_by_different_rows", } ) continue id_remap[source_id] = target_id remapped.append( { "source_id": source_id, "target_id": target_id, "identity_key": identity_key, "reason": "provider_identity_match", } ) return { "object": object_name, "identity_field": identity_field, "identity_by_source_id": identity_by_source_id, "id_remap": id_remap, "remapped": remapped, "conflicts": conflicts, } def apply_person_identity_resolution_to_manifest_files( manifest: dict[str, Any], snapshot_dir: Path, *, identities_by_object: dict[str, dict[str, str]], id_remap: dict[str, str], identity_field: str = DEFAULT_PERSON_IDENTITY_FIELD, existing_identity_by_remapped_source_id: dict[str, str] | None = None, ) -> dict[str, Any]: existing_identity_by_remapped_source_id = ( existing_identity_by_remapped_source_id or {} ) if ( not id_remap and not identities_by_object and not existing_identity_by_remapped_source_id ): return { "status": "skipped", "reason": "no_id_resolution_changes", "remapped_reference_count": 0, "prerequisite_parent_id_remap_count": 0, "identity_assignment_count": 0, "existing_identity_preservation_count": 0, } summary: dict[str, Any] = { "status": "updated", "remapped_reference_count": 0, "prerequisite_parent_id_remap_count": 0, "identity_assignment_count": 0, "existing_identity_preservation_count": 0, "objects": [], } for obj in manifest.get("objects", []): if not isinstance(obj, dict): continue object_name = str(obj.get("object", "")).strip() file_name = str(obj.get("file", "")).strip() if not file_name: continue path = snapshot_dir / file_name fieldnames, rows = read_gzip_csv(path) if not fieldnames: continue object_remapped = 0 prerequisite_parent_id_remap_count = 0 object_identity_assignments = 0 object_existing_identity_preservations = 0 object_identities = identities_by_object.get(object_name, {}) reference_fields = { str(field).strip() for field in obj.get("reference_fields", []) if str(field).strip() } rewrite_fields = reference_fields # A cross-provider Lead resolution has no new provider identity to assign, # but it still must rewrite the Lead row's primary Id before child # references are loaded. Provider-identity resolution also needs this # behavior when it remaps an existing target row. if object_identities or id_remap: rewrite_fields = set(reference_fields) rewrite_fields.add("Id") if object_identities and identity_field not in fieldnames: fieldnames.append(identity_field) prerequisite_parent_ids = [ str(source_id).strip() for source_id in obj.get("prerequisite_parent_ids", []) if str(source_id).strip() ] remapped_prerequisite_parent_ids = sorted( {id_remap.get(source_id, source_id) for source_id in prerequisite_parent_ids} ) prerequisite_parent_id_remap_count = sum( 1 for source_id in prerequisite_parent_ids if id_remap.get(source_id, source_id) != source_id ) if remapped_prerequisite_parent_ids != sorted(set(prerequisite_parent_ids)): obj["prerequisite_parent_ids"] = remapped_prerequisite_parent_ids obj["prerequisite_parent_count"] = len(remapped_prerequisite_parent_ids) for row in rows: original_id = str(row.get("Id", "")).strip() identity_key = object_identities.get(original_id) if identity_key: row[identity_field] = identity_key object_identity_assignments += 1 if original_id in existing_identity_by_remapped_source_id: if identity_field not in fieldnames: raise ValueError( f"{object_name} remapped identity preservation requires field " f"{identity_field!r}" ) existing_identity = existing_identity_by_remapped_source_id[ original_id ] if str(row.get(identity_field, "") or "") != existing_identity: row[identity_field] = existing_identity object_existing_identity_preservations += 1 for field in rewrite_fields: current_value = str(row.get(field, "")).strip() if not current_value: continue replacement = id_remap.get(current_value) if replacement: row[field] = replacement object_remapped += 1 if ( object_remapped or prerequisite_parent_id_remap_count or object_identity_assignments or object_existing_identity_preservations ): write_gzip_csv(path, fieldnames, rows) obj["record_count"] = len(rows) obj["sha256"] = sha256_file(path) obj["fields"] = fieldnames obj["field_count"] = len(fieldnames) summary["objects"].append( { "object": object_name, "file": file_name, "remapped_reference_count": object_remapped, "prerequisite_parent_id_remap_count": ( prerequisite_parent_id_remap_count ), "identity_assignment_count": object_identity_assignments, "existing_identity_preservation_count": ( object_existing_identity_preservations ), "record_count": len(rows), } ) summary["remapped_reference_count"] += object_remapped summary["prerequisite_parent_id_remap_count"] += ( prerequisite_parent_id_remap_count ) summary["identity_assignment_count"] += object_identity_assignments summary["existing_identity_preservation_count"] += ( object_existing_identity_preservations ) refresh_manifest_sha256_input(manifest) return summary def override_csv_field_values_by_id( path: Path, *, id_field: str, target_field: str, target_ids: set[str], replacement_value: str, ) -> dict[str, Any]: if not path.exists(): return {"status": "skipped", "reason": "missing_file", "path": str(path)} if not target_ids: return {"status": "skipped", "reason": "no_target_ids", "path": str(path)} temp_path = path.with_suffix(path.suffix + ".tmp") row_count = 0 matched_count = 0 changed_count = 0 with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as src: reader = csv.DictReader(src) fieldnames = list(reader.fieldnames or []) if id_field not in fieldnames or target_field not in fieldnames: return { "status": "skipped", "reason": "missing_required_field", "path": str(path), "id_field": id_field, "target_field": target_field, "fieldnames": fieldnames, } with gzip.open(temp_path, "wt", encoding="utf-8", newline="") as dst: writer = csv.DictWriter(dst, fieldnames=fieldnames) writer.writeheader() for row in reader: row_count += 1 row_id = str(row.get(id_field, "")).strip() if row_id in target_ids: matched_count += 1 current_value = str(row.get(target_field, "") or "") if current_value != replacement_value: row[target_field] = replacement_value changed_count += 1 writer.writerow(row) temp_path.replace(path) return { "status": "updated", "path": str(path), "id_field": id_field, "target_field": target_field, "target_id_count": len(target_ids), "matched_count": matched_count, "changed_count": changed_count, "row_count": row_count, "replacement_value": replacement_value, } def mysql_quote(value: str) -> str: return "'" + str(value).replace("\\", "\\\\").replace("'", "''") + "'" def lead_rows_from_manifest( manifest: dict[str, Any], snapshot_dir: Path, source_object: str, created_at_field: str, ) -> list[dict[str, str]]: manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } lead_obj = manifest_objects.get(source_object) if not lead_obj: return [] lead_file = str(lead_obj.get("file", "")).strip() if not lead_file: return [] return csv_rows_by_fields(snapshot_dir / lead_file, ["Id", created_at_field]) def apply_replay_initial_owner_override( manifest: dict[str, Any], snapshot_dir: Path, source_object: str, lead_ids: list[str], owner_id: str, ) -> dict[str, Any]: manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } lead_obj = manifest_objects.get(source_object) if not lead_obj: return { "status": "skipped", "reason": "source_object_missing_from_manifest", "source_object": source_object, } lead_file = str(lead_obj.get("file", "")).strip() if not lead_file: return { "status": "skipped", "reason": "source_object_missing_file", "source_object": source_object, } return { "source_object": source_object, **override_csv_field_values_by_id( snapshot_dir / lead_file, id_field="Id", target_field="OwnerId", target_ids=set(lead_ids), replacement_value=owner_id, ), } def lead_replay_candidate_ids( lead_rows: list[dict[str, str]], window_start: datetime, window_end: datetime, created_at_field: str, ) -> list[str]: seen: set[str] = set() candidate_ids: list[str] = [] for row in lead_rows: lead_id = str(row.get("Id") or "").strip() created_at = parse_utc_timestamp(str(row.get(created_at_field) or "").strip()) if not lead_id or created_at is None: continue if created_at < window_start or created_at >= window_end: continue if lead_id in seen: continue seen.add(lead_id) candidate_ids.append(lead_id) return candidate_ids def lead_replay_exported_ids(lead_rows: list[dict[str, str]]) -> list[str]: seen: set[str] = set() exported_ids: list[str] = [] for row in lead_rows: lead_id = str(row.get("Id") or "").strip() if not lead_id or lead_id in seen: continue seen.add(lead_id) exported_ids.append(lead_id) return exported_ids def lead_replay_new_target_candidate_ids( exported_ids: list[str], preexisting_target_ids: list[str], ) -> list[str]: preexisting = set(preexisting_target_ids) return sorted({lead_id for lead_id in exported_ids if lead_id not in preexisting}) def run_lead_assignment_replay_from_export( snapshot_dir: Path, args: argparse.Namespace, lead_ids: list[str], description_field: str, replay_key: str, ) -> dict[str, Any]: target_base_url = str(args.target_base_url or "").strip() target_username = str(args.target_username or "").strip() target_password = str(args.target_password or "").strip() if not target_base_url or not target_username or not target_password: raise ValueError("lead assignment replay requires target base URL, username, and password") lead_ids_path = snapshot_dir / "lead_assignment_replay_ids.txt" write_lines(lead_ids_path, lead_ids) summary_path = snapshot_dir / "lead_assignment_replay_summary.json" replay_cmd = [ "python3", str(Path(__file__).resolve().parent / "replay_lead_assignment_from_export.py"), "--lead-ids-file", str(lead_ids_path), "--base-url", target_base_url, "--username", target_username, "--password", target_password, "--description-field", description_field, "--replay-key", replay_key, "--summary-path", str(summary_path), ] run_command(replay_cmd) return load_json(summary_path) def build_staging_sync_report(summary: dict[str, Any]) -> dict[str, Any]: export_summary = summary.get("export") or {} load_summary = summary.get("load") or {} replay_prepare = summary.get("lead_assignment_replay_prepare") or {} replay_summary = summary.get("lead_assignment_replay") or {} lead_org_link_reconcile = summary.get("lead_org_link_reconcile") or {} checks = summary.get("checks") or [] lead_export_count = 0 record_counts = export_summary.get("record_counts") if isinstance(record_counts, dict): lead_export_count = int(record_counts.get("Lead", 0) or 0) lead_load_row_count = 0 loaded_tables: list[dict[str, Any]] = [] for table in load_summary.get("tables", []) or []: if not isinstance(table, dict): continue loaded_tables.append( { "object": str(table.get("object", "")), "target_table": str(table.get("target_table", "")), "status": str(table.get("status", "")), "row_count": int(table.get("row_count", 0) or 0), "chunk_count": int(table.get("chunk_count", 0) or 0), } ) if str(table.get("object", "")) == "Lead": lead_load_row_count = int(table.get("row_count", 0) or 0) lead_target_count = 0 for item in checks: if not isinstance(item, dict): continue if str(item.get("object", "")) == "Lead": lead_target_count = int(item.get("target_count", 0) or 0) break report: dict[str, Any] = { "run_type": "staging_sync_report", "write_mode": str(summary.get("target_write_mode", "")), "window_start": str(summary.get("window_start", "")), "window_end": str(summary.get("window_end", "")), "lead_export_count": lead_export_count, "lead_load_row_count": lead_load_row_count, "lead_target_count": lead_target_count, "loaded_tables": loaded_tables, } if replay_prepare or replay_summary: owner_override = replay_prepare.get("initial_owner_override") or {} report["replay"] = { "candidate_count": int(replay_prepare.get("candidate_count", 0) or 0), "candidate_ids_sample": list(replay_prepare.get("candidate_ids_sample") or []), "initial_owner_id": str(replay_prepare.get("initial_owner_id", "")), "initial_owner_override_changed": int(owner_override.get("changed_count", 0) or 0), "success_count": int(replay_summary.get("success_count", 0) or 0), "success_ids_sample": list(replay_summary.get("success_ids_sample") or []), "missing_target_count": int(replay_summary.get("missing_target_count", 0) or 0), "failure_count": int(replay_summary.get("failure_count", 0) or 0), } if lead_org_link_reconcile: report["lead_org_link_reconcile"] = { "planned_update_count": int(lead_org_link_reconcile.get("planned_update_count", 0) or 0), "updated_org_count": int(lead_org_link_reconcile.get("updated_org_count", 0) or 0), "conflict_count": int(lead_org_link_reconcile.get("conflict_count", 0) or 0), "duplicate_portal_id_count": int(lead_org_link_reconcile.get("duplicate_portal_id_count", 0) or 0), } return report def write_lines(path: Path, values: list[str]) -> None: path.write_text("".join(f"{value}\n" for value in values), encoding="utf-8") def fetch_target_ids_for_window( mysql_client: str, db: DbConfig, target_table: str, watermark_field: str, window_start_sql: str, window_end_sql: str, ) -> list[str]: sql = ( f"SELECT `Id` FROM `{target_table}` " f"WHERE `{watermark_field}` >= '{window_start_sql}' AND `{watermark_field}` < '{window_end_sql}' " "ORDER BY `Id`" ) rows = run_mysql_query(mysql_client, db, sql) return sorted(str(row).strip() for row in rows if str(row).strip()) def fetch_target_existing_ids( mysql_client: str, db: DbConfig, target_table: str, ids: list[str], ) -> list[str]: existing: set[str] = set() chunk_size = 200 for start in range(0, len(ids), chunk_size): chunk = ids[start : start + chunk_size] in_clause = ", ".join(mysql_quote(item) for item in chunk) rows = run_mysql_query( mysql_client, db, f"SELECT `Id` FROM `{target_table}` WHERE `Id` IN ({in_clause})", ) existing.update(str(row).strip() for row in rows if str(row).strip()) return sorted(existing) def fetch_target_identity_by_ids( mysql_client: str, db: DbConfig, *, target_table: str, ids: list[str], identity_field: str, batch_size: int, ) -> dict[str, str]: identities: dict[str, str] = {} normalized_ids = sorted({str(row_id).strip() for row_id in ids if str(row_id).strip()}) for start in range(0, len(normalized_ids), batch_size): chunk = normalized_ids[start : start + batch_size] in_clause = ", ".join(mysql_quote(item) for item in chunk) rows = run_mysql_query( mysql_client, db, f"SELECT `Id`, COALESCE(`{identity_field}`, '') FROM `{target_table}` " f"WHERE `Id` IN ({in_clause})", ) for raw in rows: parts = raw.split("\t") if not parts or not parts[0].strip(): continue identities[parts[0].strip()] = parts[1].strip() if len(parts) > 1 else "" return identities def fetch_target_rows_by_provider_identity( mysql_client: str, db: DbConfig, *, target_table: str, identity_keys: list[str], identity_field: str, batch_size: int, ) -> list[dict[str, str]]: normalized_keys = sorted({str(key).strip() for key in identity_keys if str(key).strip()}) rows: list[dict[str, str]] = [] for start in range(0, len(normalized_keys), batch_size): chunk = normalized_keys[start : start + batch_size] in_clause = ", ".join(mysql_quote(item) for item in chunk) raw_rows = run_mysql_query( mysql_client, db, f"SELECT `Id`, `{identity_field}` FROM `{target_table}` " f"WHERE `{identity_field}` IN ({in_clause})", ) for raw in raw_rows: parts = raw.split("\t") if len(parts) < 2: continue rows.append({"Id": parts[0], identity_field: parts[1]}) return rows def collect_person_reference_source_ids( *, manifest_objects: dict[str, dict[str, Any]], snapshot_dir: Path, reference_sources: Any, ) -> tuple[set[str], list[dict[str, Any]]]: if reference_sources in (None, []): return set(), [] if not isinstance(reference_sources, list): raise ValueError("person identity reference_sources must be an array") source_ids: set[str] = set() sources: list[dict[str, Any]] = [] for item in reference_sources: if not isinstance(item, dict): raise ValueError("person identity reference_sources entries must be objects") object_name = str(item.get("object", "")).strip() field_name = str(item.get("field", "")).strip() if not object_name or not field_name: raise ValueError("person identity reference_sources require object and field") manifest_obj = manifest_objects.get(object_name) if not manifest_obj: continue file_name = str(manifest_obj.get("file", "")).strip() if not file_name: raise ValueError(f"{object_name} manifest entry has no file") fieldnames, rows = read_gzip_csv(snapshot_dir / file_name) if field_name not in fieldnames: raise ValueError( f"{object_name} reference source is missing field {field_name!r}" ) item_ids = { str(row.get(field_name, "")).strip() for row in rows if str(row.get(field_name, "")).strip() } source_ids.update(item_ids) sources.append( { "object": object_name, "field": field_name, "source_id_count": len(item_ids), } ) return source_ids, sources def fetch_salesforce_rows_by_ids( *, readonly_config_path: Path, api_version: str, object_name: str, fields: list[str], source_ids: list[str], batch_size: int = 200, ) -> list[dict[str, str]]: from export_salesforce_snapshot import ( SalesforceClient, load_csv_rows, load_kv_config, run_bulk_query, ) if object_name not in SALESFORCE_PERSON_ID_PREFIXES: raise ValueError(f"unsupported reference hydration object: {object_name!r}") if batch_size <= 0: raise ValueError("reference hydration batch_size must be positive") selected_fields = list(dict.fromkeys(["Id", *fields])) if any(not field.replace("_", "").isalnum() for field in selected_fields): raise ValueError("reference hydration fields must be Salesforce identifiers") canonical_ids = sorted( { canonical_salesforce_id( source_id, expected_prefix=SALESFORCE_PERSON_ID_PREFIXES[object_name], ) for source_id in source_ids } ) if not canonical_ids: return [] client = SalesforceClient( load_kv_config(readonly_config_path), api_version=api_version, ) rows_by_id: dict[str, dict[str, str]] = {} for start in range(0, len(canonical_ids), batch_size): chunk = canonical_ids[start : start + batch_size] id_clause = ", ".join(f"'{source_id}'" for source_id in chunk) soql = ( f"SELECT {', '.join(selected_fields)} FROM {object_name} " f"WHERE Id IN ({id_clause})" ) _job_id, _job_info, content = run_bulk_query( client, soql, poll_seconds=2, timeout_seconds=900, ) header, rows = load_csv_rows(content) missing_fields = sorted(set(selected_fields) - set(header)) if missing_fields: raise ValueError( "reference hydration response is missing required fields; " f"field_count={len(missing_fields)}" ) for row in rows: source_id = str(row.get("Id", "")).strip() if source_id not in chunk: raise ValueError("reference hydration returned an unexpected Salesforce Id") if source_id in rows_by_id: raise ValueError("reference hydration returned a duplicate Salesforce Id") rows_by_id[source_id] = row return [rows_by_id[source_id] for source_id in canonical_ids if source_id in rows_by_id] def hydrate_lead_cross_provider_reference_rows( *, manifest: dict[str, Any], snapshot_dir: Path, readonly_config_path: Path, api_version: str, person_identity_config: dict[str, Any], cross_provider_config: dict[str, Any], ) -> tuple[list[dict[str, str]], dict[str, Any]]: source_object = str( cross_provider_config.get("source_object", "Lead") ).strip() or "Lead" manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } source_manifest = manifest_objects.get(source_object) if not source_manifest: return [], {"status": "skipped", "reason": "source_object_missing"} source_file = str(source_manifest.get("file", "")).strip() if not source_file: return [], {"status": "skipped", "reason": "source_file_missing"} person_objects = person_identity_config.get("objects", {}) person_object_config = ( person_objects.get(source_object, {}) if isinstance(person_objects, dict) else {} ) if not isinstance(person_object_config, dict): person_object_config = {} reference_source_ids, reference_sources = collect_person_reference_source_ids( manifest_objects=manifest_objects, snapshot_dir=snapshot_dir, reference_sources=person_object_config.get("reference_sources", []), ) _fieldnames, source_rows = read_gzip_csv(snapshot_dir / source_file) direct_source_ids = { str(row.get("Id", "")).strip() for row in source_rows if str(row.get("Id", "")).strip() } reference_only_source_ids = sorted(reference_source_ids - direct_source_ids) if not reference_only_source_ids: return [], { "status": "skipped", "reason": "no_reference_only_source_ids", "reference_sources": reference_sources, "reference_only_source_count": 0, } email_field = str(cross_provider_config.get("email_field", "Email")).strip() or "Email" hubspot_id_field = str( cross_provider_config.get("hubspot_id_field", "HubSpot_Id__c") ).strip() or "HubSpot_Id__c" raw_hydration_fields = cross_provider_config.get( "reference_hydration_fields", [email_field] ) if not isinstance(raw_hydration_fields, list): raise ValueError("lead_cross_provider_dedupe.reference_hydration_fields must be an array") hydration_fields = list( dict.fromkeys( str(field).strip() for field in raw_hydration_fields if str(field).strip() ) ) if hubspot_id_field in hydration_fields: raise ValueError( "lead_cross_provider_dedupe.reference_hydration_fields must not include " "the CRM-only HubSpot ID field" ) if email_field not in hydration_fields: raise ValueError( "lead_cross_provider_dedupe.reference_hydration_fields must include the email field" ) batch_size = int(cross_provider_config.get("batch_size", 200) or 200) hydrated_rows = fetch_salesforce_rows_by_ids( readonly_config_path=readonly_config_path, api_version=api_version, object_name=source_object, fields=hydration_fields, source_ids=reference_only_source_ids, batch_size=batch_size, ) hydrated_by_id = { str(row.get("Id", "")).strip(): row for row in hydrated_rows if str(row.get("Id", "")).strip() } unexpected_ids = sorted(set(hydrated_by_id) - set(reference_only_source_ids)) if unexpected_ids: raise ValueError( "reference hydration returned rows outside the requested set; " f"unexpected_count={len(unexpected_ids)}" ) missing_source_ids = sorted(set(reference_only_source_ids) - set(hydrated_by_id)) resolution_rows = [ hydrated_by_id.get(source_id, {"Id": source_id}) for source_id in reference_only_source_ids ] return resolution_rows, { "status": "completed", "source_object": source_object, "reference_sources": reference_sources, "reference_only_source_count": len(reference_only_source_ids), "hydrated_source_count": len(hydrated_by_id), "missing_source_count": len(missing_source_ids), } def fetch_salesforce_rows_by_ids_in_field_chunks( *, readonly_config_path: Path, api_version: str, object_name: str, fields: list[str], source_ids: list[str], row_batch_size: int, field_chunk_size: int, ) -> list[dict[str, str]]: if field_chunk_size <= 0: raise ValueError("reference full-row field_chunk_size must be positive") selected_fields = list( dict.fromkeys( str(field).strip() for field in fields if str(field).strip() and str(field).strip() != "Id" ) ) field_chunks = [ selected_fields[start : start + field_chunk_size] for start in range(0, len(selected_fields), field_chunk_size) ] or [[]] merged_by_id: dict[str, dict[str, str]] = {} ids_present_in_every_chunk: set[str] | None = None for chunk in field_chunks: rows = fetch_salesforce_rows_by_ids( readonly_config_path=readonly_config_path, api_version=api_version, object_name=object_name, fields=chunk, source_ids=source_ids, batch_size=row_batch_size, ) chunk_ids: set[str] = set() for row in rows: source_id = str(row.get("Id", "")).strip() if not source_id: continue chunk_ids.add(source_id) merged_by_id.setdefault(source_id, {"Id": source_id}).update(row) ids_present_in_every_chunk = ( chunk_ids if ids_present_in_every_chunk is None else ids_present_in_every_chunk & chunk_ids ) present_ids = ids_present_in_every_chunk or set() return [ merged_by_id[source_id] for source_id in sorted(present_ids) ] def fetch_salesforce_queryable_field_names( *, readonly_config_path: Path, api_version: str, object_name: str, excluded_fields: set[str], ) -> list[str]: from export_salesforce_snapshot import SalesforceClient, load_kv_config client = SalesforceClient( load_kv_config(readonly_config_path), api_version=api_version, ) describe = client.describe(object_name) fields: list[str] = [] for raw_field in describe.get("fields", []): if not isinstance(raw_field, dict): continue field_name = str(raw_field.get("name", "")).strip() if not field_name or field_name in excluded_fields: continue if raw_field.get("deprecatedAndHidden"): continue if str(raw_field.get("type", "")).strip().lower() in { "address", "location", }: continue fields.append(field_name) fields = list(dict.fromkeys(fields)) if "Id" not in fields: raise ValueError("Salesforce Lead describe is missing queryable Id") return fields def pull_forward_unmatched_lead_reference_rows( *, manifest: dict[str, Any], manifest_path: Path, snapshot_dir: Path, readonly_config_path: Path, api_version: str, cross_provider_config: dict[str, Any], unmatched_reference_ids: list[str], ) -> dict[str, Any]: requested_ids = sorted( { canonical_salesforce_id(source_id, expected_prefix="00Q") for source_id in unmatched_reference_ids if str(source_id).strip() } ) if not requested_ids: return { "status": "skipped", "reason": "no_unmatched_reference_ids", "prerequisite_parent_count": 0, } source_object = str( cross_provider_config.get("source_object", "Lead") ).strip() or "Lead" if source_object != "Lead": raise ValueError("cross-window prerequisite parent loading only supports Lead") manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } source_manifest = manifest_objects.get(source_object) if not source_manifest: raise ValueError("Lead prerequisite parent load requires a Lead manifest entry") source_file = str(source_manifest.get("file", "")).strip() if not source_file: raise ValueError("Lead prerequisite parent load requires a Lead snapshot file") source_path = snapshot_dir / source_file fieldnames, source_rows = read_gzip_csv(source_path) if "Id" not in fieldnames: raise ValueError("Lead prerequisite parent load requires Id in the snapshot") direct_source_ids = { str(row.get("Id", "")).strip() for row in source_rows if str(row.get("Id", "")).strip() } ids_to_fetch = sorted(set(requested_ids) - direct_source_ids) if not ids_to_fetch: return { "status": "skipped", "reason": "parents_already_in_snapshot", "prerequisite_parent_count": 0, } manifest_fields = [ str(field).strip() for field in source_manifest.get("fields", []) if str(field).strip() ] raw_source_config = source_manifest.get("config", {}) source_config = raw_source_config if isinstance(raw_source_config, dict) else {} excluded_fields = { str(field).strip() for field in source_config.get("exclude_fields", []) if str(field).strip() } source_schema_fields = fetch_salesforce_queryable_field_names( readonly_config_path=readonly_config_path, api_version=api_version, object_name=source_object, excluded_fields=excluded_fields, ) full_fields = list( dict.fromkeys([*fieldnames, *manifest_fields, *source_schema_fields]) ) identity_field = str( cross_provider_config.get("identity_field", DEFAULT_PERSON_IDENTITY_FIELD) ).strip() or DEFAULT_PERSON_IDENTITY_FIELD source_query_fields = [ field for field in full_fields if field != identity_field ] row_batch_size = int(cross_provider_config.get("batch_size", 200) or 200) field_chunk_size = int( cross_provider_config.get("reference_full_row_field_chunk_size", 40) or 40 ) fetched_rows = fetch_salesforce_rows_by_ids_in_field_chunks( readonly_config_path=readonly_config_path, api_version=api_version, object_name=source_object, fields=source_query_fields, source_ids=ids_to_fetch, row_batch_size=row_batch_size, field_chunk_size=field_chunk_size, ) fetched_by_id = { str(row.get("Id", "")).strip(): row for row in fetched_rows if str(row.get("Id", "")).strip() } missing_source_ids = sorted(set(ids_to_fetch) - set(fetched_by_id)) deleted_source_ids = sorted( source_id for source_id, row in fetched_by_id.items() if str(row.get("IsDeleted", "")).strip().lower() in {"1", "true"} ) if missing_source_ids or deleted_source_ids: return { "status": "blocked_source_parent_unavailable", "requested_parent_count": len(ids_to_fetch), "missing_source_count": len(missing_source_ids), "deleted_source_count": len(deleted_source_ids), "prerequisite_parent_count": 0, } output_fields = list(full_fields) for row in fetched_rows: for field in row: if field not in output_fields: output_fields.append(field) promoted_rows = [fetched_by_id[source_id] for source_id in ids_to_fetch] write_gzip_csv(source_path, output_fields, [*source_rows, *promoted_rows]) prior_window_record_count = int( source_manifest.get("window_record_count", source_manifest.get("record_count", len(source_rows))) or 0 ) prerequisite_parent_ids = sorted( { str(source_id).strip() for source_id in source_manifest.get("prerequisite_parent_ids", []) if str(source_id).strip() } | set(ids_to_fetch) ) source_manifest["window_record_count"] = prior_window_record_count source_manifest["prerequisite_parent_ids"] = prerequisite_parent_ids source_manifest["prerequisite_parent_count"] = len(prerequisite_parent_ids) source_manifest["record_count"] = len(source_rows) + len(promoted_rows) source_manifest["fields"] = output_fields source_manifest["field_count"] = len(output_fields) source_manifest["sha256"] = sha256_file(source_path) refresh_manifest_sha256_input(manifest) write_json(manifest_path, manifest) return { "status": "completed", "requested_parent_count": len(ids_to_fetch), "missing_source_count": 0, "deleted_source_count": 0, "prerequisite_parent_count": len(promoted_rows), "prerequisite_parent_ids": ids_to_fetch, "window_record_count": prior_window_record_count, "loaded_record_count": source_manifest["record_count"], } def apply_person_identity_resolution_to_snapshot( *, manifest: dict[str, Any], manifest_path: Path, snapshot_dir: Path, mysql_client: str, db: DbConfig, config: dict[str, Any], source_id_scope_by_object: dict[str, set[str]] | None = None, ) -> dict[str, Any]: if not bool(config.get("enabled", False)): return {"enabled": False, "status": "skipped", "reason": "disabled"} source_org_id = str(manifest.get("source_org", "")).strip() source_org_kind = str(manifest.get("source_org_kind", "")).strip() if source_org_kind != "salesforce_organization_id": raise ValueError( "person identity resolution requires source_org_kind=salesforce_organization_id" ) canonical_source_org_id = canonical_salesforce_id(source_org_id, expected_prefix="00D") identity_field = str( config.get("identity_field", DEFAULT_PERSON_IDENTITY_FIELD) ).strip() or DEFAULT_PERSON_IDENTITY_FIELD batch_size = int(config.get("batch_size", 200) or 200) raw_objects = config.get("objects", {}) if not isinstance(raw_objects, dict) or not raw_objects: raise ValueError("person_identity_resolution.objects must be a non-empty object") manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } plans: list[dict[str, Any]] = [] identities_by_object: dict[str, dict[str, str]] = {} global_id_remap: dict[str, str] = {} for object_name, raw_object_config in raw_objects.items(): if object_name not in SALESFORCE_PERSON_ID_PREFIXES: raise ValueError(f"unsupported person identity object: {object_name!r}") if ( source_id_scope_by_object is not None and object_name not in source_id_scope_by_object ): continue object_config = raw_object_config if isinstance(raw_object_config, dict) else {} manifest_obj = manifest_objects.get(object_name) if not manifest_obj: continue file_name = str(manifest_obj.get("file", "")).strip() if not file_name: raise ValueError(f"{object_name} manifest entry has no file") _fieldnames, source_rows = read_gzip_csv(snapshot_dir / file_name) if source_id_scope_by_object is not None: source_id_scope = { str(source_id).strip() for source_id in source_id_scope_by_object[object_name] if str(source_id).strip() } source_rows = [ row for row in source_rows if str(row.get("Id", "")).strip() in source_id_scope ] else: source_id_scope = None direct_source_ids = { str(row.get("Id", "")).strip() for row in source_rows if str(row.get("Id", "")).strip() } reference_source_ids, reference_sources = collect_person_reference_source_ids( manifest_objects=manifest_objects, snapshot_dir=snapshot_dir, reference_sources=object_config.get("reference_sources", []), ) if source_id_scope is not None: reference_source_ids &= source_id_scope source_ids = sorted(direct_source_ids | reference_source_ids) if not source_ids: continue resolution_rows = list(source_rows) resolution_rows.extend( {"Id": source_id} for source_id in sorted(reference_source_ids - direct_source_ids) ) target_table = str( object_config.get("target_table") or manifest_obj.get("target_table") or object_name.lower() ).strip() existing_target_identity_by_id = fetch_target_identity_by_ids( mysql_client, db, target_table=target_table, ids=source_ids, identity_field=identity_field, batch_size=batch_size, ) existing_target_ids = set(existing_target_identity_by_id) identity_keys = [ salesforce_person_identity_key(canonical_source_org_id, object_name, source_id) for source_id in source_ids ] target_rows = fetch_target_rows_by_provider_identity( mysql_client, db, target_table=target_table, identity_keys=identity_keys, identity_field=identity_field, batch_size=batch_size, ) plan = build_person_identity_resolution_plan( object_name=object_name, source_org_id=canonical_source_org_id, source_rows=resolution_rows, target_rows=target_rows, existing_target_ids=existing_target_ids, existing_target_identity_by_id=existing_target_identity_by_id, identity_field=identity_field, ) reference_only_source_ids = reference_source_ids - direct_source_ids unmatched_reference_source_ids = sorted( reference_only_source_ids - existing_target_ids - set(plan["id_remap"]) ) require_existing_target_for_references = object_config.get( "require_existing_target_for_references", False ) if not isinstance(require_existing_target_for_references, bool): raise ValueError( "person identity require_existing_target_for_references must be boolean" ) defer_unmatched_references = object_config.get( "defer_unmatched_references_to_cross_provider", False ) if not isinstance(defer_unmatched_references, bool): raise ValueError( "person identity defer_unmatched_references_to_cross_provider must be boolean" ) if ( require_existing_target_for_references and unmatched_reference_source_ids and not defer_unmatched_references ): plan["conflicts"].append( { "reason": "unmatched_referenced_person_ids", "source_id_count": len(unmatched_reference_source_ids), "source_ids": unmatched_reference_source_ids[:20], } ) plan["target_table"] = target_table plan["reference_sources"] = reference_sources plan["reference_source_id_count"] = len(reference_source_ids) plan["unmatched_reference_source_ids"] = unmatched_reference_source_ids plan["deferred_reference_source_ids"] = ( unmatched_reference_source_ids if defer_unmatched_references else [] ) plans.append(plan) identities_by_object[object_name] = plan["identity_by_source_id"] for source_id, target_id in plan["id_remap"].items(): prior = global_id_remap.get(source_id) if prior and prior != target_id: raise ValueError( f"source id {source_id!r} resolves to multiple target ids: {prior!r}, {target_id!r}" ) global_id_remap[source_id] = target_id conflicts = [ {"object": plan["object"], **conflict} for plan in plans for conflict in plan["conflicts"] ] if conflicts: conflict_reasons = sorted( { str(conflict.get("reason", "unknown")).strip() or "unknown" for conflict in conflicts } ) raise ValueError( "provider-scoped person identity conflicts must be resolved before load; " f"conflict_count={len(conflicts)} " f"conflict_reasons={','.join(conflict_reasons)}" ) rewrite_summary = apply_person_identity_resolution_to_manifest_files( manifest, snapshot_dir, identities_by_object=identities_by_object, id_remap=global_id_remap, identity_field=identity_field, ) resolved_objects = sorted( object_name for object_name in raw_objects if object_name in manifest_objects ) manifest["person_identity_resolution"] = { "status": "completed", "source_org_kind": "salesforce_organization_id", "source_org": canonical_source_org_id, "identity_field": identity_field, "objects": resolved_objects, } refresh_manifest_sha256_input(manifest) write_json(manifest_path, manifest) artifact_dir = snapshot_dir / "_person_identity_resolution" ensure_dir(artifact_dir) artifact = { "enabled": True, "status": "completed", "source_org": canonical_source_org_id, "identity_field": identity_field, "object_count": len(plans), "remapped_count": len(global_id_remap), "deferred_reference_count": sum( len(plan["deferred_reference_source_ids"]) for plan in plans ), "plans": plans, "rewrite": rewrite_summary, } if source_id_scope_by_object is not None: artifact["source_id_scope"] = { object_name: sorted( { str(source_id).strip() for source_id in source_ids if str(source_id).strip() } ) for object_name, source_ids in sorted(source_id_scope_by_object.items()) } artifact_path = artifact_dir / "person_identity_resolution.json" write_json(artifact_path, artifact) summary = { "enabled": True, "status": "completed", "source_org": canonical_source_org_id, "identity_field": identity_field, "artifact_path": str(artifact_path), "object_count": len(plans), "remapped_count": len(global_id_remap), "deferred_reference_count": sum( len(plan["deferred_reference_source_ids"]) for plan in plans ), "remapped_sample": [ {"object": plan["object"], **item} for plan in plans for item in plan["remapped"] ][:20], "rewrite": rewrite_summary, } if source_id_scope_by_object is not None: summary["source_id_scope"] = artifact["source_id_scope"] return summary def normalize_hubspot_id_key(value: Any) -> str: return str(value or "").strip() def normalize_email_key(value: Any) -> str: return str(value or "").strip().lower() def fetch_active_target_leads_by_hubspot_or_email( mysql_client: str, db: DbConfig, *, target_table: str, hubspot_ids: list[str], emails: list[str], hubspot_id_field: str, email_field: str, identity_field: str, is_deleted_field: str, batch_size: int, ) -> list[dict[str, str]]: normalized_hubspot_ids = sorted( {normalize_hubspot_id_key(value) for value in hubspot_ids if normalize_hubspot_id_key(value)} ) normalized_emails = sorted( {normalize_email_key(value) for value in emails if normalize_email_key(value)} ) rows: list[dict[str, str]] = [] query_keys = [("hubspot", value) for value in normalized_hubspot_ids] + [ ("email", value) for value in normalized_emails ] for start in range(0, len(query_keys), batch_size): chunk = query_keys[start : start + batch_size] chunk_hubspot_ids = [value for kind, value in chunk if kind == "hubspot"] chunk_emails = [value for kind, value in chunk if kind == "email"] predicates: list[str] = [] if chunk_hubspot_ids: predicates.append( f"TRIM(`{hubspot_id_field}`) IN ({', '.join(mysql_quote(value) for value in chunk_hubspot_ids)})" ) if chunk_emails: predicates.append( f"LOWER(TRIM(`{email_field}`)) IN ({', '.join(mysql_quote(value) for value in chunk_emails)})" ) if not predicates: continue raw_rows = run_mysql_query( mysql_client, db, " ".join( [ ( f"SELECT `Id`, COALESCE(`{hubspot_id_field}`, ''), " f"COALESCE(`{email_field}`, ''), COALESCE(`{identity_field}`, '')" ), f"FROM `{target_table}`", f"WHERE ({' OR '.join(predicates)})", f"AND (`{is_deleted_field}` IS NULL OR `{is_deleted_field}` = 0)", ] ), ) for raw in raw_rows: parts = raw.split("\t") if len(parts) < 4 or not parts[0].strip(): continue rows.append( { "Id": parts[0].strip(), hubspot_id_field: parts[1], email_field: parts[2], identity_field: parts[3], } ) return rows def build_lead_cross_provider_dedupe_plan( *, source_lead_rows: list[dict[str, str]], target_lead_rows: list[dict[str, str]], existing_target_ids: set[str], id_field: str = "Id", hubspot_id_field: str = "HubSpot_Id__c", email_field: str = "Email", identity_field: str = DEFAULT_PERSON_IDENTITY_FIELD, ) -> dict[str, Any]: source_rows: dict[str, dict[str, str]] = {} source_ids_by_hubspot: dict[str, set[str]] = {} for row in source_lead_rows: source_id = str(row.get(id_field, "")).strip() if not source_id: continue source_rows[source_id] = row hubspot_id = normalize_hubspot_id_key(row.get(hubspot_id_field)) if hubspot_id: source_ids_by_hubspot.setdefault(hubspot_id, set()).add(source_id) target_ids_by_hubspot: dict[str, set[str]] = {} target_ids_by_email: dict[str, set[str]] = {} target_hubspot_id_by_id: dict[str, str] = {} target_identity_by_id: dict[str, str] = {} for row in target_lead_rows: target_id = str(row.get(id_field, "")).strip() if not target_id: continue hubspot_id = normalize_hubspot_id_key(row.get(hubspot_id_field)) target_hubspot_id_by_id[target_id] = hubspot_id target_identity_by_id[target_id] = str(row.get(identity_field, "")).strip() if hubspot_id: target_ids_by_hubspot.setdefault(hubspot_id, set()).add(target_id) email = normalize_email_key(row.get(email_field)) if email: target_ids_by_email.setdefault(email, set()).add(target_id) conflicts_by_source: dict[str, dict[str, Any]] = {} def conflict(source_id: str, reason: str, **details: Any) -> None: conflicts_by_source.setdefault( source_id, {"source_id": source_id, "reason": reason, **details}, ) for hubspot_id, source_ids in source_ids_by_hubspot.items(): if len(source_ids) > 1: for source_id in source_ids: conflict( source_id, "duplicate_source_hubspot_id", source_ids=sorted(source_ids), ) resolved: dict[str, tuple[str, str]] = {} email_fallback_source_ids: dict[str, set[str]] = {} for source_id, row in sorted(source_rows.items()): if source_id in conflicts_by_source: continue hubspot_id = normalize_hubspot_id_key(row.get(hubspot_id_field)) hubspot_targets = sorted(target_ids_by_hubspot.get(hubspot_id, set())) if hubspot_id else [] if len(hubspot_targets) > 1: conflict( source_id, "ambiguous_target_hubspot_id", target_ids=hubspot_targets, ) continue if len(hubspot_targets) == 1: target_id = hubspot_targets[0] if target_id == source_id: continue source_identity = str(row.get(identity_field, "")).strip() target_identity = target_identity_by_id.get(target_id, "") if source_identity and target_identity and source_identity != target_identity: conflict( source_id, "target_salesforce_identity_conflict", target_id=target_id, source_identity=source_identity, target_identity=target_identity, ) continue resolved[source_id] = (target_id, "unique_hubspot_id_match") continue email = normalize_email_key(row.get(email_field)) if email: email_fallback_source_ids.setdefault(email, set()).add(source_id) for email, source_ids in email_fallback_source_ids.items(): if len(source_ids) > 1: for source_id in source_ids: conflict( source_id, "duplicate_source_email_for_fallback", source_ids=sorted(source_ids), ) continue source_id = next(iter(source_ids)) target_ids = sorted(target_ids_by_email.get(email, set())) hubspot_primary_target_ids = [ target_id for target_id in target_ids if target_hubspot_id_by_id.get(target_id, "") ] if source_id in existing_target_ids and not hubspot_primary_target_ids: continue if len(hubspot_primary_target_ids) == 1: target_id = hubspot_primary_target_ids[0] reason = ( "existing_sfdc_duplicate_email_match_to_hubspot_primary" if source_id in existing_target_ids and target_id != source_id else "unique_email_fallback_match" ) elif len(target_ids) > 1: conflict( source_id, "ambiguous_target_email", target_ids=target_ids, ) continue elif len(target_ids) == 1: # Email fallback is only safe for a target that has an immutable # HubSpot identity. A Salesforce-origin target with the same email # is a separate source record, not cross-provider evidence for an # ID remap. continue else: continue if target_id == source_id: continue source_row = source_rows[source_id] source_hubspot_id = normalize_hubspot_id_key(source_row.get(hubspot_id_field)) target_hubspot_id = target_hubspot_id_by_id.get(target_id, "") if source_hubspot_id and target_hubspot_id and source_hubspot_id != target_hubspot_id: conflict( source_id, "hubspot_id_email_target_conflict", target_id=target_id, source_hubspot_id=source_hubspot_id, target_hubspot_id=target_hubspot_id, ) continue source_identity = str(source_row.get(identity_field, "")).strip() target_identity = target_identity_by_id.get(target_id, "") if source_identity and target_identity and source_identity != target_identity: conflict( source_id, "target_salesforce_identity_conflict", target_id=target_id, source_identity=source_identity, target_identity=target_identity, ) continue resolved[source_id] = (target_id, reason) source_ids_by_target: dict[str, set[str]] = {} for source_id, (target_id, _reason) in resolved.items(): source_ids_by_target.setdefault(target_id, set()).add(source_id) for target_id, source_ids in source_ids_by_target.items(): if len(source_ids) > 1: for source_id in source_ids: resolved.pop(source_id, None) conflict( source_id, "multiple_source_rows_resolve_to_same_target", target_id=target_id, source_ids=sorted(source_ids), ) id_remap = {source_id: target_id for source_id, (target_id, _reason) in resolved.items()} unmatched_source_ids = sorted( set(source_rows) - set(resolved) - set(conflicts_by_source) - existing_target_ids ) remapped = [ {"source_id": source_id, "target_id": target_id, "reason": reason} for source_id, (target_id, reason) in sorted(resolved.items()) ] return { "id_remap": id_remap, "remapped": remapped, "conflicts": [conflicts_by_source[source_id] for source_id in sorted(conflicts_by_source)], "unmatched_source_ids": unmatched_source_ids, } def apply_lead_cross_provider_dedupe_to_snapshot( *, manifest: dict[str, Any], manifest_path: Path, snapshot_dir: Path, mysql_client: str, db: DbConfig, config: dict[str, Any], reference_only_source_rows: list[dict[str, str]] | None = None, ) -> dict[str, Any]: if not bool(config.get("enabled", False)): return {"enabled": False, "status": "skipped", "reason": "disabled"} source_object = str(config.get("source_object", "Lead")).strip() or "Lead" target_table = str(config.get("target_table", "lead")).strip() or "lead" hubspot_id_field = str(config.get("hubspot_id_field", "HubSpot_Id__c")).strip() or "HubSpot_Id__c" email_field = str(config.get("email_field", "Email")).strip() or "Email" identity_field = str(config.get("identity_field", DEFAULT_PERSON_IDENTITY_FIELD)).strip() or DEFAULT_PERSON_IDENTITY_FIELD is_deleted_field = str(config.get("is_deleted_field", "IsDeleted")).strip() or "IsDeleted" batch_size = int(config.get("batch_size", 200) or 200) require_existing_target_match = config.get("require_existing_target_match", False) if not isinstance(require_existing_target_match, bool): raise ValueError("lead_cross_provider_dedupe.require_existing_target_match must be boolean") manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } lead_obj = manifest_objects.get(source_object) if not lead_obj: return {"enabled": True, "status": "skipped", "reason": "source_object_missing_from_manifest"} lead_file = str(lead_obj.get("file", "")).strip() if not lead_file: return {"enabled": True, "status": "skipped", "reason": "source_object_missing_file"} _fieldnames, source_lead_rows = read_gzip_csv(snapshot_dir / lead_file) reference_only_source_rows = reference_only_source_rows or [] candidate_rows_by_id = { str(row.get("Id", "")).strip(): row for row in [*reference_only_source_rows, *source_lead_rows] if str(row.get("Id", "")).strip() } candidate_rows = list(candidate_rows_by_id.values()) source_ids = sorted(candidate_rows_by_id) if not source_ids: return {"enabled": True, "status": "skipped", "reason": "no_source_ids"} existing_target_ids = set(fetch_target_existing_ids(mysql_client, db, target_table, source_ids)) target_lead_rows = fetch_active_target_leads_by_hubspot_or_email( mysql_client, db, target_table=target_table, hubspot_ids=[normalize_hubspot_id_key(row.get(hubspot_id_field)) for row in candidate_rows], emails=[normalize_email_key(row.get(email_field)) for row in candidate_rows], hubspot_id_field=hubspot_id_field, email_field=email_field, identity_field=identity_field, is_deleted_field=is_deleted_field, batch_size=batch_size, ) if candidate_rows else [] plan = build_lead_cross_provider_dedupe_plan( source_lead_rows=candidate_rows, target_lead_rows=target_lead_rows, existing_target_ids=existing_target_ids, hubspot_id_field=hubspot_id_field, email_field=email_field, identity_field=identity_field, ) source_lead_ids = { str(row.get("Id", "")).strip() for row in source_lead_rows if str(row.get("Id", "")).strip() } reference_only_ids = { str(row.get("Id", "")).strip() for row in reference_only_source_rows if str(row.get("Id", "")).strip() } - source_lead_ids unmatched_source_ids = set(plan["unmatched_source_ids"]) create_candidate_ids = sorted(unmatched_source_ids & source_lead_ids) unmatched_reference_ids = sorted(unmatched_source_ids & reference_only_ids) target_identity_by_id = { str(row.get("Id", "")).strip(): str( row.get(identity_field, "") or "" ).strip() for row in target_lead_rows if str(row.get("Id", "")).strip() } missing_identity_targets = sorted( set(plan["id_remap"].values()) - set(target_identity_by_id) ) if missing_identity_targets: raise ValueError( "Lead cross-provider remap is missing fetched target identity values; " f"target_count={len(missing_identity_targets)}" ) existing_identity_by_remapped_source_id = { source_id: ( target_identity_by_id[target_id] if target_identity_by_id[target_id] else CSV_NULL_SENTINEL ) for source_id, target_id in plan["id_remap"].items() } rewrite = apply_person_identity_resolution_to_manifest_files( manifest, snapshot_dir, identities_by_object={}, id_remap=plan["id_remap"], identity_field=identity_field, existing_identity_by_remapped_source_id=( existing_identity_by_remapped_source_id ), ) if rewrite.get("status") == "updated": write_json(manifest_path, manifest) artifact_dir = snapshot_dir / "_lead_cross_provider_dedupe" ensure_dir(artifact_dir) artifact = { "enabled": True, "status": ( "conflict" if plan["conflicts"] else "blocked_unmatched_reference" if unmatched_reference_ids else "blocked_unmatched" if require_existing_target_match and create_candidate_ids else "completed" ), "source_object": source_object, "target_table": target_table, "hubspot_id_field": hubspot_id_field, "email_field": email_field, "identity_field": identity_field, "source_row_count": len(source_lead_rows), "reference_only_source_count": len(reference_only_source_rows), "resolution_source_row_count": len(candidate_rows), "existing_target_id_count": len(existing_target_ids), "target_candidate_row_count": len(target_lead_rows), "remapped_count": len(plan["id_remap"]), "conflict_count": len(plan["conflicts"]), "require_existing_target_match": require_existing_target_match, "unmatched_count": len(plan["unmatched_source_ids"]), "create_candidate_count": len(create_candidate_ids), "create_candidate_ids": create_candidate_ids, "unmatched_reference_count": len(unmatched_reference_ids), "unmatched_reference_ids": unmatched_reference_ids, "plan": plan, "rewrite": rewrite, } artifact_path = artifact_dir / "lead_cross_provider_dedupe.json" write_json(artifact_path, artifact) return { "enabled": True, "status": artifact["status"], "artifact_path": str(artifact_path), "remapped_count": len(plan["id_remap"]), "conflict_count": len(plan["conflicts"]), "reference_only_source_count": len(reference_only_source_rows), "require_existing_target_match": require_existing_target_match, "unmatched_count": len(plan["unmatched_source_ids"]), "create_candidate_count": len(create_candidate_ids), "unmatched_reference_count": len(unmatched_reference_ids), "_unmatched_reference_ids": unmatched_reference_ids, "unmatched_source_ids_sample": plan["unmatched_source_ids"][:20], "remapped_sample": plan["remapped"][:20], "conflicts_sample": plan["conflicts"][:20], "rewrite": rewrite, } def validate_lead_cross_provider_dedupe_before_load(summary: dict[str, Any]) -> None: if summary.get("conflict_count", 0): raise ValueError( "Lead cross-provider dedupe conflicts must be resolved before load; " f"conflict_count={int(summary.get('conflict_count', 0) or 0)}" ) if summary.get("unmatched_reference_count", 0): raise ValueError( "Lead child reference has no target Lead and no parent row to create; " f"unmatched_reference_count={int(summary.get('unmatched_reference_count', 0) or 0)}" ) if summary.get("require_existing_target_match", False) and summary.get( "create_candidate_count", summary.get("unmatched_count", 0) ): raise ValueError( "Lead matched-existing writer refuses to insert unmatched Salesforce IDs; " f"unmatched_count={int(summary.get('create_candidate_count', summary.get('unmatched_count', 0)) or 0)}" ) def resolve_lead_identity_and_cross_provider_snapshot( *, manifest: dict[str, Any], manifest_path: Path, snapshot_dir: Path, readonly_config_path: Path, api_version: str, mysql_client: str, db: DbConfig, person_identity_config: dict[str, Any], cross_provider_config: dict[str, Any], reference_only_source_rows: list[dict[str, str]], ) -> dict[str, Any]: result: dict[str, Any] = {} person_identity_enabled = bool(person_identity_config.get("enabled", False)) cross_provider_enabled = bool(cross_provider_config.get("enabled", False)) if person_identity_enabled: result["person_identity_resolution"] = apply_person_identity_resolution_to_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, config=person_identity_config, ) if not cross_provider_enabled: return result dedupe_summary = apply_lead_cross_provider_dedupe_to_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, config=cross_provider_config, reference_only_source_rows=reference_only_source_rows, ) unmatched_reference_ids = [ str(source_id).strip() for source_id in dedupe_summary.get("_unmatched_reference_ids", []) if str(source_id).strip() ] if unmatched_reference_ids: pull_forward = pull_forward_unmatched_lead_reference_rows( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, readonly_config_path=readonly_config_path, api_version=api_version, cross_provider_config=cross_provider_config, unmatched_reference_ids=unmatched_reference_ids, ) result["lead_cross_window_parent_pull_forward"] = pull_forward if ( pull_forward.get("status") == "completed" and int(pull_forward.get("prerequisite_parent_count", 0) or 0) > 0 ): if person_identity_enabled: identity_summary = apply_person_identity_resolution_to_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, config=person_identity_config, source_id_scope_by_object={ "Lead": { str(source_id).strip() for source_id in pull_forward.get( "prerequisite_parent_ids", [] ) if str(source_id).strip() } }, ) identity_summary["rerun_after_prerequisite_parent_pull_forward"] = True result["person_identity_resolution"] = identity_summary dedupe_summary = apply_lead_cross_provider_dedupe_to_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, config=cross_provider_config, reference_only_source_rows=reference_only_source_rows, ) dedupe_summary.pop("_unmatched_reference_ids", None) result["lead_cross_provider_dedupe"] = dedupe_summary validate_lead_cross_provider_dedupe_before_load(dedupe_summary) return result def build_mismatch_diagnostics( snapshot_dir: Path, object_name: str, file_name: str, source_ids: list[str], target_ids: list[str], ) -> dict[str, Any]: diagnostics_root = snapshot_dir / "_check_diffs" ensure_dir(diagnostics_root) object_dir = diagnostics_root / sanitize_path_component(object_name) ensure_dir(object_dir) source_only_ids = sorted(set(source_ids) - set(target_ids)) target_only_ids = sorted(set(target_ids) - set(source_ids)) source_ids_path = object_dir / "source_ids.txt" target_ids_path = object_dir / "target_ids.txt" source_only_path = object_dir / "source_only_ids.txt" target_only_path = object_dir / "target_only_ids.txt" metadata_path = object_dir / "metadata.json" write_lines(source_ids_path, source_ids) write_lines(target_ids_path, target_ids) write_lines(source_only_path, source_only_ids) write_lines(target_only_path, target_only_ids) metadata = { "object": object_name, "source_file": file_name, "source_ids_count": len(source_ids), "target_ids_count": len(target_ids), "source_only_ids_count": len(source_only_ids), "target_only_ids_count": len(target_only_ids), "source_ids_path": source_ids_path.name, "target_ids_path": target_ids_path.name, "source_only_ids_path": source_only_path.name, "target_only_ids_path": target_only_path.name, } write_json(metadata_path, metadata) return { "artifact_dir": str(object_dir), "source_ids_count": len(source_ids), "target_ids_count": len(target_ids), "source_only_ids_count": len(source_only_ids), "target_only_ids_count": len(target_only_ids), "source_ids_sample": source_ids[:20], "target_ids_sample": target_ids[:20], "source_only_ids_sample": source_only_ids[:20], "target_only_ids_sample": target_only_ids[:20], "metadata_path": str(metadata_path), "source_ids_path": str(source_ids_path), "target_ids_path": str(target_ids_path), "source_only_ids_path": str(source_only_path), "target_only_ids_path": str(target_only_path), } def reconcile_lead_owner_shares( mysql_client: str, db: DbConfig, manifest: dict[str, Any], snapshot_dir: Path, ) -> dict[str, Any] | None: manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } lead_obj = manifest_objects.get("Lead") lead_share_obj = manifest_objects.get("LeadShare") if not lead_obj or not lead_share_obj: return None lead_file = str(lead_obj.get("file", "")).strip() lead_share_file = str(lead_share_obj.get("file", "")).strip() if not lead_file or not lead_share_file: return None lead_rows = csv_rows_by_fields(snapshot_dir / lead_file, ["Id", "OwnerId"]) changed_lead_ids = sorted({row["Id"] for row in lead_rows if row.get("Id")}) if not changed_lead_ids: return { "object": "LeadShare", "reason": "no_changed_leads_in_window", "changed_lead_count": 0, "deleted_owner_share_rows": 0, "lead_ids_reconciled": 0, } source_owner_share_ids_by_lead: dict[str, set[str]] = {} for row in csv_rows_by_fields(snapshot_dir / lead_share_file, ["Id", "LeadId", "RowCause"]): lead_id = row.get("LeadId", "") share_id = row.get("Id", "") if not lead_id or not share_id or row.get("RowCause") != "Owner": continue source_owner_share_ids_by_lead.setdefault(lead_id, set()).add(share_id) reconciled_lead_ids = sorted(source_owner_share_ids_by_lead.keys()) if not reconciled_lead_ids: return { "object": "LeadShare", "reason": "no_source_owner_shares_in_window", "changed_lead_count": len(changed_lead_ids), "deleted_owner_share_rows": 0, "lead_ids_reconciled": 0, } delete_ids: list[str] = [] chunk_size = 200 for start in range(0, len(reconciled_lead_ids), chunk_size): chunk = reconciled_lead_ids[start : start + chunk_size] in_clause = ", ".join(mysql_quote(item) for item in chunk) rows = run_mysql_query( mysql_client, db, " ".join( [ "SELECT `Id`, `LeadId`", "FROM `leadshare`", "WHERE (`IsDeleted` = 0 OR `IsDeleted` IS NULL)", "AND `RowCause` = 'Owner'", f"AND `LeadId` IN ({in_clause})", ] ), ) for raw in rows: parts = raw.split("\t") if len(parts) != 2: continue share_id, lead_id = parts source_ids = source_owner_share_ids_by_lead.get(lead_id) if source_ids is None or share_id in source_ids: continue delete_ids.append(share_id) if delete_ids: for start in range(0, len(delete_ids), chunk_size): chunk = delete_ids[start : start + chunk_size] delete_clause = ", ".join(mysql_quote(item) for item in chunk) run_mysql_query( mysql_client, db, f"DELETE FROM `leadshare` WHERE `Id` IN ({delete_clause})", ) return { "object": "LeadShare", "reason": "source_owner_share_reconcile", "changed_lead_count": len(changed_lead_ids), "lead_ids_reconciled": len(reconciled_lead_ids), "deleted_owner_share_rows": len(delete_ids), "sample_deleted_ids": delete_ids[:20], } def fetch_rows_by_ids( mysql_client: str, db: DbConfig, table: str, columns: list[str], ids: list[str], *, id_field: str = "Id", chunk_size: int = 200, ) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] selected_columns = [id_field, *columns] select_clause = ", ".join(f"`{column}`" for column in selected_columns) for start in range(0, len(ids), chunk_size): chunk = ids[start : start + chunk_size] in_clause = ", ".join(mysql_quote(item) for item in chunk) raw_rows = run_mysql_query( mysql_client, db, f"SELECT {select_clause} FROM `{table}` WHERE `{id_field}` IN ({in_clause})", ) for raw in raw_rows: parts = raw.split("\t") item: dict[str, str] = {} for index, column in enumerate(selected_columns): item[column] = parts[index] if index < len(parts) else "" rows.append(item) return rows def build_portal_to_lead_map( lead_rows: list[dict[str, str]], *, lead_id_field: str, portal_id_field: str, ) -> tuple[dict[str, str], list[str]]: portal_to_lead: dict[str, str] = {} duplicate_portal_ids: set[str] = set() for row in lead_rows: lead_id = str(row.get(lead_id_field, "")).strip() portal_id = str(row.get(portal_id_field, "")).strip() if not lead_id or not portal_id: continue existing = portal_to_lead.get(portal_id) if existing is None: portal_to_lead[portal_id] = lead_id continue if existing != lead_id: duplicate_portal_ids.add(portal_id) for portal_id in duplicate_portal_ids: portal_to_lead.pop(portal_id, None) return portal_to_lead, sorted(duplicate_portal_ids) def build_org_lead_update_plan( org_rows: list[dict[str, str]], *, portal_to_lead: dict[str, str], org_lookup_field: str, org_lead_field: str, ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: updates: list[dict[str, str]] = [] conflicts: list[dict[str, str]] = [] for row in org_rows: portal_id = str(row.get(org_lookup_field, "")).strip() current_lead_id = str(row.get(org_lead_field, "")).strip() expected_lead_id = portal_to_lead.get(portal_id, "") if not portal_id or not expected_lead_id: continue if not current_lead_id: updates.append({"portal_id": portal_id, "lead_id": expected_lead_id}) continue if current_lead_id != expected_lead_id: conflicts.append( { "portal_id": portal_id, "current_lead_id": current_lead_id, "expected_lead_id": expected_lead_id, } ) return updates, conflicts def apply_org_lead_updates( mysql_client: str, db: DbConfig, *, org_table: str, org_lookup_field: str, org_lead_field: str, updates: list[dict[str, str]], chunk_size: int, ) -> int: updated_rows = 0 for start in range(0, len(updates), chunk_size): chunk = updates[start : start + chunk_size] when_clauses = [] params: list[str] = [] lookup_values: list[str] = [] for item in chunk: when_clauses.append(f"WHEN {mysql_quote(item['portal_id'])} THEN {mysql_quote(item['lead_id'])}") lookup_values.append(mysql_quote(item["portal_id"])) sql = ( f"UPDATE `{org_table}` " f"SET `{org_lead_field}` = CASE `{org_lookup_field}` {' '.join(when_clauses)} ELSE `{org_lead_field}` END " f"WHERE `{org_lookup_field}` IN ({', '.join(lookup_values)}) " f"AND (`{org_lead_field}` IS NULL OR `{org_lead_field}` = '')" ) raw_result = run_mysql_query(mysql_client, db, sql + "; SELECT ROW_COUNT();") if raw_result: try: updated_rows += int(raw_result[-1].strip()) except ValueError: pass return updated_rows def reconcile_org_lead_linkage( mysql_client: str, db: DbConfig, manifest: dict[str, Any], snapshot_dir: Path, *, source_object: str, lead_table: str, org_table: str, lead_portal_id_field: str, org_lookup_field: str, org_lead_field: str, batch_size: int, ) -> dict[str, Any] | None: manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } lead_obj = manifest_objects.get(source_object) if not lead_obj: return None lead_file = str(lead_obj.get("file", "")).strip() if not lead_file: return None changed_lead_ids = csv_distinct_ids(snapshot_dir / lead_file) if not changed_lead_ids: return { "object": "Org__c", "reason": "no_changed_leads_in_window", "changed_lead_count": 0, "target_leads_with_portal_id": 0, "matched_org_count": 0, "updated_org_count": 0, "conflict_count": 0, "duplicate_portal_id_count": 0, } target_lead_rows = fetch_rows_by_ids( mysql_client, db, lead_table, [lead_portal_id_field], changed_lead_ids, id_field="Id", chunk_size=batch_size, ) portal_to_lead, duplicate_portal_ids = build_portal_to_lead_map( target_lead_rows, lead_id_field="Id", portal_id_field=lead_portal_id_field, ) if not portal_to_lead: return { "object": "Org__c", "reason": "no_target_leads_with_portal_id", "changed_lead_count": len(changed_lead_ids), "target_leads_with_portal_id": 0, "matched_org_count": 0, "updated_org_count": 0, "conflict_count": 0, "duplicate_portal_id_count": len(duplicate_portal_ids), "duplicate_portal_ids_sample": duplicate_portal_ids[:20], } org_rows = fetch_rows_by_ids( mysql_client, db, org_table, [org_lead_field], sorted(portal_to_lead.keys()), id_field=org_lookup_field, chunk_size=batch_size, ) updates, conflicts = build_org_lead_update_plan( org_rows, portal_to_lead=portal_to_lead, org_lookup_field=org_lookup_field, org_lead_field=org_lead_field, ) updated_org_count = apply_org_lead_updates( mysql_client, db, org_table=org_table, org_lookup_field=org_lookup_field, org_lead_field=org_lead_field, updates=updates, chunk_size=batch_size, ) if updates else 0 return { "object": "Org__c", "reason": "lead_to_org_link_materialize", "changed_lead_count": len(changed_lead_ids), "target_leads_with_portal_id": len(portal_to_lead), "matched_org_count": len(org_rows), "planned_update_count": len(updates), "updated_org_count": updated_org_count, "conflict_count": len(conflicts), "duplicate_portal_id_count": len(duplicate_portal_ids), "duplicate_portal_ids_sample": duplicate_portal_ids[:20], "conflicts_sample": conflicts[:20], } def run_source_target_count_check( mysql_client: str, db: DbConfig, manifest: dict[str, Any], snapshot_dir: Path, selected_objects: list[str], window_start: str, window_end: str, watermark_field: str, fallback_watermark_field: str, target_write_mode: str, ) -> list[dict[str, Any]]: manifest_objects = { str(item.get("object", "")).strip(): item for item in manifest.get("objects", []) if isinstance(item, dict) and str(item.get("object", "")).strip() } window_start_sql = format_mysql_timestamp(parse_utc_timestamp(window_start)) window_end_sql = format_mysql_timestamp(parse_utc_timestamp(window_end)) results: list[dict[str, Any]] = [] for name in selected_objects: obj = manifest_objects.get(name) if not obj: continue target_table = str(obj.get("target_table", "")).strip() file_name = str(obj.get("file", "")).strip() if not target_table: continue chosen_field = str(obj.get("incremental_watermark_field") or watermark_field or fallback_watermark_field).strip() sql = ( f"SELECT COUNT(*) FROM `{target_table}` " f"WHERE `{chosen_field}` >= '{window_start_sql}' AND `{chosen_field}` < '{window_end_sql}'" ) rows = run_mysql_query(mysql_client, db, sql) target_count = int(rows[0]) if rows and rows[0].strip() else 0 source_count_mode = "raw_rows" source_key_fields: list[str] = [] source_count = int(obj.get("record_count", 0)) prerequisite_parent_ids = { str(source_id).strip() for source_id in obj.get("prerequisite_parent_ids", []) if str(source_id).strip() } fields = [str(item).strip() for item in obj.get("fields", []) if str(item).strip()] if prerequisite_parent_ids: source_key_fields = ["Id"] source_count = int(obj.get("window_record_count", 0) or 0) source_count_mode = "window_rows_excluding_prerequisite_parents" elif target_write_mode == "replace" and "Id" in fields and file_name: source_key_fields = ["Id"] source_count = csv_distinct_key_count(snapshot_dir / file_name, source_key_fields) source_count_mode = "distinct_keys" status = "ok" if source_count == target_count else "mismatch" result = { "object": name, "target_table": target_table, "watermark_field": chosen_field, "source_count": source_count, "source_count_mode": source_count_mode, "source_key_fields": source_key_fields, "target_count": target_count, "status": status, } if status != "ok": if source_key_fields == ["Id"] and file_name: source_ids = sorted( set(csv_distinct_ids(snapshot_dir / file_name)) - prerequisite_parent_ids ) target_ids = fetch_target_ids_for_window( mysql_client=mysql_client, db=db, target_table=target_table, watermark_field=chosen_field, window_start_sql=window_start_sql, window_end_sql=window_end_sql, ) result["mismatch_diagnostics"] = build_mismatch_diagnostics( snapshot_dir=snapshot_dir, object_name=name, file_name=file_name, source_ids=source_ids, target_ids=target_ids, ) print( f"[warn] count mismatch for {name}: source={source_count} " f"target={target_count} window={window_start}..{window_end}", file=sys.stderr, flush=True, ) results.append(result) return results def require_db_config(args: argparse.Namespace) -> DbConfig: db = DbConfig( host=args.db_host, port=args.db_port, user=args.db_user, password=args.db_password, database=args.db_database, ssl_mode=args.ssl_mode, ) if not all([db.host, db.user, db.database]): raise ValueError("db host/user/database are required") return db def main() -> int: args = parse_args() if args.export_only and args.dry_run_load: raise ValueError("--export-only and --dry-run-load are mutually exclusive") if args.export_only: args.skip_load = True args.skip_check = True if args.dry_run_load: args.skip_check = True args.skip_replay = True args.skip_state_update = True sync_config = load_json(Path(args.sync_config).expanduser().resolve()) export_config_path = Path(args.export_config).expanduser().resolve() load_plan_path = Path(args.load_plan_config).expanduser().resolve() readonly_config_path = Path(args.config).expanduser().resolve() export_config = load_json(export_config_path) load_plan = load_json(load_plan_path) person_identity_resolution_config = sync_config.get("person_identity_resolution", {}) if not isinstance(person_identity_resolution_config, dict): person_identity_resolution_config = {} person_identity_resolution_enabled = bool( person_identity_resolution_config.get("enabled", False) ) lead_cross_provider_dedupe_config = sync_config.get("lead_cross_provider_dedupe", {}) if not isinstance(lead_cross_provider_dedupe_config, dict): lead_cross_provider_dedupe_config = {} lead_cross_provider_dedupe_enabled = bool( lead_cross_provider_dedupe_config.get("enabled", False) ) source_profile = str(sync_config.get("source_profile", DEFAULT_PROFILE)).strip() or DEFAULT_PROFILE default_watermark_field = str(sync_config.get("default_watermark_field", "SystemModstamp")).strip() or "SystemModstamp" fallback_watermark_field = str(sync_config.get("fallback_watermark_field", "LastModifiedDate")).strip() or "LastModifiedDate" bootstrap_lookback_hours = int(sync_config.get("bootstrap_lookback_hours", 48)) steady_state_lookback_minutes = int(sync_config.get("steady_state_lookback_minutes", 15)) require_state_for_implicit_window = bool(sync_config.get("require_state_for_implicit_window", False)) max_implicit_window_hours = sync_config.get("max_implicit_window_hours") state_file = Path(args.state_file).expanduser() if args.state_file else Path( str(sync_config.get("state_file", "~/.cache/nexuscrm/salesforce/hourly_delta_state.json")) ).expanduser() target_write_mode = str(sync_config.get("target_write_mode", "replace")).strip() or "replace" lead_assignment_replay_config = sync_config.get("lead_assignment_replay", {}) replay_enabled = bool(lead_assignment_replay_config.get("enabled", False)) replay_source_object = str(lead_assignment_replay_config.get("source_object", "Lead")).strip() or "Lead" replay_new_lead_created_at_field = str(lead_assignment_replay_config.get("new_lead_created_at_field", "CreatedDate")).strip() or "CreatedDate" replay_target_description_field = str(lead_assignment_replay_config.get("target_description_field", "Description")).strip() or "Description" replay_initial_owner_id = str(lead_assignment_replay_config.get("initial_owner_id", "")).strip() replay_target_table = str(lead_assignment_replay_config.get("target_table", "lead")).strip() or "lead" replay_override_initial_owner = bool(lead_assignment_replay_config.get("override_initial_owner_before_load", False)) lead_org_link_reconcile_config = sync_config.get("lead_org_link_reconcile", {}) lead_org_link_reconcile_enabled = bool(lead_org_link_reconcile_config.get("enabled", False)) lead_org_link_source_object = str(lead_org_link_reconcile_config.get("source_object", "Lead")).strip() or "Lead" lead_org_link_lead_table = str(lead_org_link_reconcile_config.get("lead_target_table", "lead")).strip() or "lead" lead_org_link_org_table = str(lead_org_link_reconcile_config.get("org_target_table", "org__c")).strip() or "org__c" lead_org_link_lead_portal_id_field = str(lead_org_link_reconcile_config.get("lead_portal_id_field", "Portal_ID__c")).strip() or "Portal_ID__c" lead_org_link_org_lookup_field = str(lead_org_link_reconcile_config.get("org_lookup_field", "ID__c")).strip() or "ID__c" lead_org_link_org_lead_field = str(lead_org_link_reconcile_config.get("org_lead_field", "Lead__c")).strip() or "Lead__c" lead_org_link_batch_size = int(lead_org_link_reconcile_config.get("batch_size", 200) or 200) summary_name = args.summary_name export_profiles = export_config.get("profiles", {}) if source_profile not in export_profiles or not isinstance(export_profiles[source_profile], list): raise ValueError(f"source profile {source_profile!r} not found in export config") load_objects = [str(item).strip() for item in load_plan.get("include_objects", []) if str(item).strip()] selected_objects = intersect_objects([str(item) for item in export_profiles[source_profile]], load_objects) if not selected_objects: raise ValueError("no shared objects between source profile and load plan") now = utc_now().replace(microsecond=0) if args.since: window_start = parse_utc_timestamp(args.since) else: state = load_json(state_file) if state_file.exists() else {} last_watermark = parse_utc_timestamp(str(state.get("last_successful_watermark", "")).strip()) if last_watermark: window_start = last_watermark - timedelta(minutes=steady_state_lookback_minutes) elif require_state_for_implicit_window: raise ValueError( f"missing required state watermark for implicit window: state_file={state_file}" ) else: window_start = now - timedelta(hours=bootstrap_lookback_hours) if window_start is None: raise ValueError("unable to determine delta window start") window_end = parse_utc_timestamp(args.until) if args.until else now if window_end is None: raise ValueError("unable to determine delta window end") if window_end <= window_start: raise ValueError(f"window end must be after window start: start={window_start} end={window_end}") if not args.since and max_implicit_window_hours is not None: max_window = timedelta(hours=float(max_implicit_window_hours)) actual_window = window_end - window_start if actual_window > max_window: raise ValueError( f"implicit window too large: actual={actual_window} max={max_window} " f"state_file={state_file} start={window_start.isoformat()} end={window_end.isoformat()}" ) snapshot_date = args.snapshot_date or window_start.date().isoformat() run_label = sanitize_path_component(args.run_label) output_root = Path(args.output_root).expanduser().resolve() snapshot_dir = output_root / snapshot_date / run_label ensure_dir(snapshot_dir) window_start_str = format_sf_timestamp(window_start) window_end_str = format_sf_timestamp(window_end) temp_export_config = build_temp_export_config( export_config=export_config, selected_objects=selected_objects, window_start=window_start_str, window_end=window_end_str, watermark_field=default_watermark_field, fallback_watermark_field=fallback_watermark_field, ) temp_load_plan = build_temp_load_plan(load_plan, selected_objects) temp_export_config_path = snapshot_dir / "hourly_delta_export_config.json" temp_load_plan_path = snapshot_dir / "hourly_delta_load_plan.json" write_json(temp_export_config_path, temp_export_config) write_json(temp_load_plan_path, temp_load_plan) mysql_client = args.mysql_client or MYSQL_CLIENT db: DbConfig | None = None if not args.skip_load or not args.skip_check: if not mysql_client: raise RuntimeError("mysql client not found") db = require_db_config(args) summary: dict[str, Any] = { "manifest_version": 1, "run_type": "hourly_delta_sync", "snapshot_dir": str(snapshot_dir), "source_profile": source_profile, "selected_objects": selected_objects, "window_start": window_start_str, "window_end": window_end_str, "watermark_field": default_watermark_field, "fallback_watermark_field": fallback_watermark_field, "bootstrap_lookback_hours": bootstrap_lookback_hours, "steady_state_lookback_minutes": steady_state_lookback_minutes, "state_file": str(state_file), "export_config": str(temp_export_config_path), "load_plan_config": str(temp_load_plan_path), "target_write_mode": target_write_mode, "dry_run_load": args.dry_run_load, "person_identity_resolution": None, "lead_cross_provider_reference_hydration": None, "lead_cross_provider_dedupe": None, "export": None, "load": None, "checks": [], } mismatch_diagnostics_root = snapshot_dir / "_check_diffs" if not args.skip_export: export_cmd = [ "python3", str(Path(__file__).resolve().parent / "export_salesforce_snapshot.py"), "--config", str(readonly_config_path), "--export-config", str(temp_export_config_path), "--output-root", str(output_root), "--profile", "hourly_delta_sync", "--snapshot-date", snapshot_date, "--run-label", run_label, ] if args.s3_bucket: export_cmd.extend(["--s3-bucket", args.s3_bucket]) if args.s3_prefix: export_cmd.extend(["--s3-prefix", args.s3_prefix]) if args.skip_s3_upload: export_cmd.append("--skip-s3-upload") run_command(export_cmd) manifest_path = snapshot_dir / args.manifest_name manifest = load_json(manifest_path) reference_only_source_rows: list[dict[str, str]] = [] if ( person_identity_resolution_enabled and lead_cross_provider_dedupe_enabled and (not args.skip_load or args.dry_run_load) and db is not None and mysql_client ): ( reference_only_source_rows, summary["lead_cross_provider_reference_hydration"], ) = hydrate_lead_cross_provider_reference_rows( manifest=manifest, snapshot_dir=snapshot_dir, readonly_config_path=readonly_config_path, api_version=str(export_config.get("api_version") or "v63.0"), person_identity_config=person_identity_resolution_config, cross_provider_config=lead_cross_provider_dedupe_config, ) if ( (person_identity_resolution_enabled or lead_cross_provider_dedupe_enabled) and (not args.skip_load or args.dry_run_load) and db is not None and mysql_client ): summary.update(resolve_lead_identity_and_cross_provider_snapshot( manifest=manifest, manifest_path=manifest_path, snapshot_dir=snapshot_dir, readonly_config_path=readonly_config_path, api_version=str(export_config.get("api_version") or "v63.0"), mysql_client=mysql_client, db=db, person_identity_config=person_identity_resolution_config, cross_provider_config=lead_cross_provider_dedupe_config, reference_only_source_rows=reference_only_source_rows, )) lead_rows = lead_rows_from_manifest( manifest, snapshot_dir, replay_source_object, replay_new_lead_created_at_field, ) replay_source_candidate_ids = ( lead_replay_candidate_ids( lead_rows, window_start, window_end, replay_new_lead_created_at_field, ) if replay_enabled else [] ) replay_exported_ids = lead_replay_exported_ids(lead_rows) if replay_enabled else [] replay_preexisting_target_ids: list[str] = [] if replay_enabled and db is not None and mysql_client and replay_exported_ids: replay_preexisting_target_ids = fetch_target_existing_ids( mysql_client=mysql_client, db=db, target_table=replay_target_table, ids=replay_exported_ids, ) if replay_enabled and db is not None and mysql_client: replay_candidate_ids = lead_replay_new_target_candidate_ids( replay_exported_ids, replay_preexisting_target_ids, ) replay_candidate_selection = "exported_ids_missing_from_target_before_load" else: replay_candidate_ids = sorted(set(replay_source_candidate_ids)) if replay_enabled else [] replay_candidate_selection = "created_date_window_fallback" replay_initial_owner_override_summary: dict[str, Any] | None = None if replay_enabled and replay_override_initial_owner and replay_initial_owner_id and replay_candidate_ids: replay_initial_owner_override_summary = apply_replay_initial_owner_override( manifest=manifest, snapshot_dir=snapshot_dir, source_object=replay_source_object, lead_ids=replay_candidate_ids, owner_id=replay_initial_owner_id, ) summary["export"] = { "manifest": str(manifest_path), "object_count": manifest.get("object_count"), "record_counts": { str(item.get("object", "")): item.get("record_count") for item in manifest.get("objects", []) if isinstance(item, dict) }, "window_record_counts": { str(item.get("object", "")): item.get( "window_record_count", item.get("record_count") ) for item in manifest.get("objects", []) if isinstance(item, dict) }, "prerequisite_parent_counts": { str(item.get("object", "")): item.get("prerequisite_parent_count", 0) for item in manifest.get("objects", []) if isinstance(item, dict) and int(item.get("prerequisite_parent_count", 0) or 0) > 0 }, } if not args.skip_load and not args.dry_run_load and db is not None: load_cmd = [ "python3", str(Path(__file__).resolve().parent / "load_salesforce_snapshot_to_tidb.py"), "--snapshot-dir", str(snapshot_dir), "--manifest-name", args.manifest_name, "--summary-name", summary_name, "--export-config", str(temp_export_config_path), "--load-plan-config", str(temp_load_plan_path), "--write-mode", target_write_mode, "--db-host", db.host, "--db-port", str(db.port), "--db-user", db.user, "--db-database", db.database, "--ssl-mode", db.ssl_mode, ] replay_queue_file = ( Path(args.replay_queue_file).expanduser().resolve() if str(args.replay_queue_file or "").strip() else None ) replay_intent = run_load_with_durable_replay_intent( load_cmd=load_cmd, load_env={"TIDB_PASSWORD": db.password}, replay_deferred=bool(args.skip_replay and replay_enabled), replay_queue_file=replay_queue_file, window_start=window_start_str, window_end=window_end_str, description_field=replay_target_description_field, replay_candidate_ids=replay_candidate_ids, ) if replay_intent is not None: summary["lead_assignment_replay_intent"] = replay_intent lead_share_reconcile = reconcile_lead_owner_shares( mysql_client=mysql_client, db=db, manifest=manifest, snapshot_dir=snapshot_dir, ) if lead_share_reconcile is not None: summary["leadshare_owner_reconcile"] = lead_share_reconcile if lead_org_link_reconcile_enabled: lead_org_link_reconcile = reconcile_org_lead_linkage( mysql_client=mysql_client, db=db, manifest=manifest, snapshot_dir=snapshot_dir, source_object=lead_org_link_source_object, lead_table=lead_org_link_lead_table, org_table=lead_org_link_org_table, lead_portal_id_field=lead_org_link_lead_portal_id_field, org_lookup_field=lead_org_link_org_lookup_field, org_lead_field=lead_org_link_org_lead_field, batch_size=lead_org_link_batch_size, ) if lead_org_link_reconcile is not None: summary["lead_org_link_reconcile"] = lead_org_link_reconcile load_summary_path = snapshot_dir / summary_name if load_summary_path.exists(): summary["load"] = load_json(load_summary_path) if replay_enabled: replay_candidate_ids_path = snapshot_dir / "lead_assignment_replay_ids.txt" write_lines(replay_candidate_ids_path, replay_candidate_ids) summary["lead_assignment_replay_prepare"] = { "enabled": True, "source_object": replay_source_object, "target_table": replay_target_table, "new_lead_created_at_field": replay_new_lead_created_at_field, "target_description_field": replay_target_description_field, "initial_owner_id": replay_initial_owner_id, "override_initial_owner_before_load": replay_override_initial_owner, "candidate_selection": replay_candidate_selection, "candidate_ids_path": str(replay_candidate_ids_path), "exported_lead_count": len(replay_exported_ids), "exported_lead_ids_sample": replay_exported_ids[:20], "source_candidate_count": len(replay_source_candidate_ids), "source_candidate_ids_sample": replay_source_candidate_ids[:20], "preexisting_lead_count": len(replay_preexisting_target_ids), "preexisting_target_count": len(replay_preexisting_target_ids), "preexisting_target_ids_sample": replay_preexisting_target_ids[:20], "new_lead_count": len(replay_candidate_ids), "candidate_count": len(replay_candidate_ids), "candidate_ids_sample": replay_candidate_ids[:20], "candidate_ids_path": str(replay_candidate_ids_path), } if replay_initial_owner_override_summary is not None: summary["lead_assignment_replay_prepare"]["initial_owner_override"] = replay_initial_owner_override_summary if args.skip_replay: summary["lead_assignment_replay"] = { "run_type": "lead_assignment_replay_from_export", "status": "deferred", "reason": "skip_replay_requested", "source_object": replay_source_object, "new_lead_created_at_field": replay_new_lead_created_at_field, "description_field": replay_target_description_field, "candidate_count": len(replay_candidate_ids), "candidate_ids_path": str(replay_candidate_ids_path), "success_count": 0, "missing_target_count": 0, "failure_count": 0, } elif replay_candidate_ids: summary["lead_assignment_replay"] = run_lead_assignment_replay_from_export( snapshot_dir=snapshot_dir, args=args, lead_ids=replay_candidate_ids, description_field=replay_target_description_field, replay_key=f"{window_start_str}__{window_end_str}", ) else: summary["lead_assignment_replay"] = { "run_type": "lead_assignment_replay_from_export", "status": "skipped", "reason": "no_replay_candidates_in_window", "source_object": replay_source_object, "new_lead_created_at_field": replay_new_lead_created_at_field, "description_field": replay_target_description_field, "success_count": 0, "missing_target_count": 0, "failure_count": 0, } if not args.skip_check and db is not None and mysql_client: checks = run_source_target_count_check( mysql_client=mysql_client, db=db, manifest=manifest, snapshot_dir=snapshot_dir, selected_objects=selected_objects, window_start=window_start_str, window_end=window_end_str, watermark_field=default_watermark_field, fallback_watermark_field=fallback_watermark_field, target_write_mode=target_write_mode, ) summary["checks"] = checks if mismatch_diagnostics_root.exists(): summary["mismatch_diagnostics_root"] = str(mismatch_diagnostics_root) summary["staging_sync_report"] = build_staging_sync_report(summary) overall_max_watermark: datetime | None = None for obj in manifest.get("objects", []): if not isinstance(obj, dict): continue file_name = str(obj.get("file", "")).strip() if not file_name: continue watermark_column = str(obj.get("incremental_watermark_field") or default_watermark_field).strip() prerequisite_parent_ids = { str(source_id).strip() for source_id in obj.get("prerequisite_parent_ids", []) if str(source_id).strip() } file_max = csv_max_timestamp( snapshot_dir / file_name, watermark_column, excluded_ids=prerequisite_parent_ids, ) if file_max and (overall_max_watermark is None or file_max > overall_max_watermark): overall_max_watermark = file_max if not args.skip_state_update: state_payload = { "last_successful_watermark": format_sf_timestamp(overall_max_watermark or window_end), "last_window_start": window_start_str, "last_window_end": window_end_str, "last_snapshot_dir": str(snapshot_dir), "last_manifest": str(manifest_path), "selected_objects": selected_objects, "updated_at": format_sf_timestamp(utc_now()), } ensure_dir(state_file.parent) write_json(state_file, state_payload) summary_path = snapshot_dir / summary_name write_json(summary_path, summary) print(f"[ok] summary_manifest={summary_path}") if not args.skip_state_update: print(f"[ok] state_file={state_file}") return 0 if __name__ == "__main__": raise SystemExit(main())