#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import re import subprocess import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any import resolve_person_identity_snapshot as person_identity_resolver DEFAULT_EXPORT_CONFIG = "configs/daily_full_export.json" DEFAULT_MONITOR_CONFIG = "configs/prod_reconcile_monitor.json" DEFAULT_LOAD_PLAN_CONFIGS = ( "configs/tidb_load_business_phase1.json", "configs/tidb_load_business_supplemental_batch2.json", "configs/tidb_load_emailmessage_batch1.json", "configs/tidb_load_identity_reference_batch2.json", ) SUPPORTED_LOAD_MODES = {"chunked_load_data", "chunked_insert"} AUTO_INSERT_OBJECTS = { "CampaignMember", "ContentDocumentLink", "EmailMessage", "LeadHistory", "LeadShare", "Opportunity_Product__c", "Task", "TaskRelation", } NO_TRUNCATE_OBJECTS = { "Profile", "PermissionSet", } SOURCE_UNSUPPORTED_OBJECTS = { "ContractShare", } DEFAULT_TARGET_CLIENT_ID = "nexuscrm-ui" DEFAULT_TARGET_CLIENT_SECRET = "nexuscrm-ui-internal" LOAD_RETRYABLE_PATTERNS = ( "TLS/SSL error", "unexpected eof while reading", "Incorrect datetime value", ) def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Run serial production repair: live reconcile, targeted export, ensure table, load, and post-reconcile." ) parser.add_argument("--readonly-config", required=True, help="Path to Salesforce readonly config.") parser.add_argument("--snapshot-root", required=True, help="Snapshot root for reconcile discovery.") parser.add_argument("--output-root", help="Export output root. Defaults to snapshot root.") parser.add_argument("--monitor-config", default=str(script_dir / DEFAULT_MONITOR_CONFIG)) parser.add_argument("--export-config", default=str(script_dir / DEFAULT_EXPORT_CONFIG)) parser.add_argument( "--load-plan-config", action="append", default=[], help="Base load-plan config. Repeatable. Defaults to the built-in prod set.", ) parser.add_argument("--object", action="append", help="Optional object allowlist. Repeatable.") parser.add_argument("--skip-object", action="append", help="Optional object denylist. Repeatable.") parser.add_argument("--run-label", help="Optional cycle label.") parser.add_argument("--summary-json", help="Optional path for the final cycle summary.") parser.add_argument("--state-dir", help="Persistent state directory for resumable per-object repair.") parser.add_argument("--retry-count", type=int, default=2, help="Max load attempts per object.") parser.add_argument("--sleep-seconds-on-empty", type=int, default=0, help="Optional sleep after an all-ok cycle.") 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("--source-admin-config", default=os.environ.get("SF_ADMIN_CONFIG_PATH", "")) parser.add_argument("--target-base-url", default=os.environ.get("NEXUSCRM_TARGET_BASE_URL", "")) parser.add_argument("--target-client-id", default=os.environ.get("NEXUSCRM_TARGET_CLIENT_ID", DEFAULT_TARGET_CLIENT_ID)) parser.add_argument( "--target-client-secret", default=os.environ.get("NEXUSCRM_TARGET_CLIENT_SECRET", DEFAULT_TARGET_CLIENT_SECRET), ) parser.add_argument("--target-username", default=os.environ.get("NEXUSCRM_TARGET_USERNAME", "")) parser.add_argument("--target-password", default=os.environ.get("NEXUSCRM_TARGET_PASSWORD", "")) parser.add_argument("--skip-export", action="store_true") parser.add_argument("--skip-load", action="store_true") parser.add_argument("--skip-post-reconcile", action="store_true") parser.add_argument("--truncate-first", action="store_true", default=True) return parser.parse_args() def utc_now() -> datetime: return datetime.now(timezone.utc) def now_label() -> str: return utc_now().strftime("prod-repair-%Y%m%dT%H%M%SZ") def today_utc() -> str: return utc_now().date().isoformat() def sanitize_component(value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()) return cleaned.strip("-") or "item" def load_json(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must be a JSON object") return payload def write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") def nonempty(value: str) -> str: return value.strip() def parse_items(values: list[str] | None) -> set[str]: result: set[str] = set() for raw in values or []: for item in str(raw).split(","): stripped = item.strip() if stripped: result.add(stripped) return result def build_env(args: argparse.Namespace) -> dict[str, str]: env = os.environ.copy() env["TIDB_PROD_HOST"] = args.db_host env["TIDB_PROD_PORT"] = str(args.db_port) env["TIDB_PROD_USER"] = args.db_user env["TIDB_PROD_PASSWORD"] = args.db_password env["TIDB_PROD_DATABASE"] = args.db_database env["TIDB_SSL_MODE"] = args.ssl_mode env["TIDB_HOST"] = args.db_host env["TIDB_PORT"] = str(args.db_port) env["TIDB_USER"] = args.db_user env["TIDB_PASSWORD"] = args.db_password env["TIDB_DATABASE"] = args.db_database return env def run_cmd( cmd: list[str], env: dict[str, str], *, check: bool = True, capture_output: bool = False, ) -> subprocess.CompletedProcess[str]: return subprocess.run( cmd, env=env, text=True, check=check, capture_output=capture_output, ) def load_plan_payload(path: Path) -> dict[str, Any]: payload = load_json(path) raw_map = payload.get("object_load_strategies", {}) if raw_map is not None and not isinstance(raw_map, dict): raise ValueError(f"{path} object_load_strategies must be an object") for object_name, strategy in (raw_map or {}).items(): if not isinstance(strategy, dict): raise ValueError(f"{path} strategy for {object_name} must be an object") mode = str(strategy.get("mode") or "").strip() if mode and mode not in SUPPORTED_LOAD_MODES: raise ValueError(f"{path} has unsupported mode {mode!r} for object {object_name!r}") return payload def merge_strategy_maps(paths: list[Path]) -> dict[str, dict[str, Any]]: merged: dict[str, dict[str, Any]] = {} for path in paths: payload = load_plan_payload(path) for object_name, strategy in payload.get("object_load_strategies", {}).items(): merged[str(object_name).strip()] = dict(strategy) return merged def merge_person_identity_resolution_requirements( paths: list[Path], ) -> dict[str, dict[str, Any]]: requirements_by_object: dict[str, dict[str, Any]] = {} for path in paths: payload = load_plan_payload(path) raw_requirement = payload.get("person_identity_resolution_requirement") if raw_requirement is None: continue if not isinstance(raw_requirement, dict): raise ValueError( f"{path} person_identity_resolution_requirement must be an object" ) identity_field = str(raw_requirement.get("identity_field") or "").strip() raw_objects = raw_requirement.get("objects") if not identity_field: raise ValueError( f"{path} person_identity_resolution_requirement.identity_field is required" ) if not isinstance(raw_objects, list) or not raw_objects: raise ValueError( f"{path} person_identity_resolution_requirement.objects must be a non-empty array" ) for raw_object_name in raw_objects: object_name = str(raw_object_name).strip() if not object_name: continue scoped_requirement = { **raw_requirement, "identity_field": identity_field, "objects": [object_name], } existing = requirements_by_object.get(object_name) if existing is not None and existing != scoped_requirement: raise ValueError( f"conflicting person identity resolution requirements for {object_name}" ) requirements_by_object[object_name] = scoped_requirement return requirements_by_object def run_reconcile( script_dir: Path, env: dict[str, str], snapshot_root: Path, monitor_config: Path, export_config: Path, objects: list[str] | None, output_json: Path, ) -> dict[str, Any]: cmd = [ "python3", str(script_dir / "verify_latest_snapshot_vs_tidb.py"), "--snapshot-root", str(snapshot_root), "--monitor-config", str(monitor_config), "--export-config", str(export_config), "--output-json", str(output_json), ] for object_name in objects or []: cmd.extend(["--object", object_name]) run_cmd(cmd, env=env, check=False) return load_json(output_json) def select_mismatches(payload: dict[str, Any], allow: set[str], deny: set[str]) -> list[str]: selected: list[str] = [] for item in payload.get("objects", []): object_name = str(item.get("object") or "").strip() status = str(item.get("status") or "").strip() if not object_name or status == "ok": continue if allow and object_name not in allow: continue if object_name in deny: continue selected.append(object_name) return selected def extract_snapshot_dir(manifest_path: Path) -> Path: return manifest_path.parent def load_manifest(snapshot_dir: Path) -> dict[str, Any]: return load_json(snapshot_dir / "manifest.json") def snapshot_file_for_object(snapshot_dir: Path, object_name: str) -> Path: manifest = load_manifest(snapshot_dir) for item in manifest.get("objects", []): current = str(item.get("object") or "").strip() if current != object_name: continue file_name = str(item.get("file") or "").strip() if not file_name: break path = snapshot_dir / file_name if path.exists(): return path break raise FileNotFoundError(f"snapshot file not found for {object_name} under {snapshot_dir}") def build_object_load_plan( strategy_map: dict[str, dict[str, Any]], object_name: str, output_path: Path, *, mode_override: str | None = None, person_identity_resolution_requirement: dict[str, Any] | None = None, ) -> Path: strategy = dict( strategy_map.get( object_name, {"mode": "chunked_load_data", "chunk_size": 500, "min_fallback_rows": 25}, ) ) if object_name in AUTO_INSERT_OBJECTS and mode_override is None: strategy["mode"] = "chunked_insert" if mode_override is not None: strategy["mode"] = mode_override payload = { "config_version": 1, "name": f"prod_repair_{sanitize_component(object_name)}", "include_objects": [object_name], "object_load_strategies": { object_name: strategy, }, } if person_identity_resolution_requirement is not None: payload["person_identity_resolution_requirement"] = dict( person_identity_resolution_requirement ) write_json(output_path, payload) return output_path def truncate_first_for_object( *, requested: bool, object_name: str, person_identity_requirements: dict[str, dict[str, Any]], ) -> bool: return ( requested and object_name not in NO_TRUNCATE_OBJECTS and object_name not in person_identity_requirements ) def should_retry_with_insert(error_text: str) -> bool: return any(pattern in error_text for pattern in LOAD_RETRYABLE_PATTERNS) def object_state_path(state_dir: Path, object_name: str) -> Path: return state_dir / f"{sanitize_component(object_name)}.json" def load_object_state(state_dir: Path, object_name: str) -> dict[str, Any]: path = object_state_path(state_dir, object_name) if not path.exists(): return {"object": object_name} return load_json(path) def save_object_state(state_dir: Path, object_name: str, payload: dict[str, Any]) -> None: path = object_state_path(state_dir, object_name) write_json(path, payload) def run_export_for_object( script_dir: Path, env: dict[str, str], readonly_config: Path, export_config: Path, output_root: Path, object_name: str, cycle_label: str, snapshot_date: str, ) -> Path: if object_name in SOURCE_UNSUPPORTED_OBJECTS: raise RuntimeError(f"source unsupported for {object_name}") object_label = sanitize_component(object_name) run_label = f"{cycle_label}-{object_label}" manifest_path = output_root / snapshot_date / run_label / "manifest.json" if object_name == "TaskWhoRelation": snapshot_dir = output_root / snapshot_date / run_label snapshot_dir.mkdir(parents=True, exist_ok=True) base_manifest_path = snapshot_dir / "manifest.json" if not base_manifest_path.exists(): write_json( base_manifest_path, { "manifest_version": 1, "snapshot_date": snapshot_date, "run_label": run_label, "local_output_dir": str(snapshot_dir), "objects": [], "object_count": 0, }, ) cmd = [ "python3", str(script_dir / "export_task_who_relations.py"), "--config", str(readonly_config), "--snapshot-dir", str(snapshot_dir), "--base-manifest-name", "manifest.json", ] run_cmd(cmd, env=env) task_manifest_path = snapshot_dir / "manifest_task_who_relation.json" if not task_manifest_path.exists(): raise FileNotFoundError(f"expected TaskWhoRelation manifest not found: {task_manifest_path}") dedicated_manifest = load_json(task_manifest_path) objects = [ { "object": "TaskWhoRelation", "target_table": "taskwhorelation", "file": str(dedicated_manifest.get("file") or "TaskWhoRelation.csv.gz"), "record_count": int(dedicated_manifest.get("record_count") or 0), "fields": list(dedicated_manifest.get("fields") or []), } ] base_manifest = load_json(base_manifest_path) base_manifest["objects"] = objects base_manifest["object_count"] = len(objects) base_manifest["snapshot_time"] = dedicated_manifest.get("snapshot_time") base_manifest["source_org"] = dedicated_manifest.get("source_org") base_manifest["api_version"] = dedicated_manifest.get("api_version") write_json(base_manifest_path, base_manifest) return snapshot_dir cmd = [ "python3", str(script_dir / "export_salesforce_snapshot.py"), "--config", str(readonly_config), "--export-config", str(export_config), "--output-root", str(output_root), "--snapshot-date", snapshot_date, "--run-label", run_label, "--objects", object_name, ] run_cmd(cmd, env=env) if not manifest_path.exists(): raise FileNotFoundError(f"expected export manifest not found: {manifest_path}") return extract_snapshot_dir(manifest_path) def ensure_table_for_object( script_dir: Path, env: dict[str, str], snapshot_dir: Path, export_config: Path, object_name: str, ) -> None: cmd = [ "python3", str(script_dir / "ensure_tidb_tables_from_snapshot.py"), "--snapshot-dir", str(snapshot_dir), "--export-config", str(export_config), "--objects", object_name, "--apply", ] run_cmd(cmd, env=env) def resolve_person_identity_before_schema_mutation( *, snapshot_dir: Path, object_name: str, requirement: dict[str, Any] | None, env: dict[str, str], ) -> dict[str, Any] | None: if requirement is None: return None mysql_client = person_identity_resolver.hourly_delta_sync.MYSQL_CLIENT if not mysql_client: raise RuntimeError( "mysql client is required to resolve person identity before repair" ) identity_field = str(requirement.get("identity_field") or "").strip() required_objects = { str(item).strip() for item in requirement.get("objects", []) if str(item).strip() } if not identity_field or required_objects != {object_name}: raise ValueError( f"repair person identity requirement must be scoped exactly to {object_name}" ) db = person_identity_resolver.hourly_delta_sync.DbConfig( host=str(env.get("TIDB_HOST") or "").strip(), port=int(env.get("TIDB_PORT") or "4000"), user=str(env.get("TIDB_USER") or "").strip(), password=str(env.get("TIDB_PASSWORD") or ""), database=str(env.get("TIDB_DATABASE") or "").strip(), ssl_mode=str(env.get("TIDB_SSL_MODE") or "REQUIRED").strip(), ) if not all([db.host, db.user, db.database]): raise ValueError( "db host/user/database are required for repair person identity resolution" ) result = person_identity_resolver.resolve_person_identity_snapshot( snapshot_dir=snapshot_dir, mysql_client=mysql_client, db=db, objects={object_name}, expected_identity_field=identity_field, require_requested_objects=True, ) return result def load_object_once( script_dir: Path, env: dict[str, str], snapshot_dir: Path, export_config: Path, object_name: str, plan_path: Path, truncate_first: bool, ) -> subprocess.CompletedProcess[str]: cmd = [ "python3", str(script_dir / "load_salesforce_snapshot_to_tidb.py"), "--snapshot-dir", str(snapshot_dir), "--export-config", str(export_config), "--load-plan-config", str(plan_path), ] if truncate_first: cmd.append("--truncate-first") return run_cmd(cmd, env=env, check=False, capture_output=True) def maybe_repair_oli_fields( *, script_dir: Path, env: dict[str, str], snapshot_dir: Path, object_name: str, source_admin_config: Path | None, target_base_url: str, target_client_id: str, target_client_secret: str, target_username: str, target_password: str, ) -> dict[str, Any] | None: if object_name != "OpportunityLineItem": return None if not source_admin_config: return {"status": "skipped", "reason": "missing_source_admin_config"} if not all([target_base_url, target_username, target_password]): return {"status": "skipped", "reason": "missing_target_credentials"} snapshot_file = snapshot_file_for_object(snapshot_dir, object_name) cmd = [ "python3", str(script_dir / "repair_oli_fields_from_source.py"), "--source-config", str(source_admin_config), "--target-base-url", target_base_url, "--target-client-id", target_client_id, "--target-client-secret", target_client_secret, "--target-username", target_username, "--target-password", target_password, "--snapshot-file", str(snapshot_file), "--apply", ] result = run_cmd(cmd, env=env, capture_output=True) return { "status": "applied", "snapshot_file": str(snapshot_file), "summary": json.loads(result.stdout), } def repair_one_object( *, script_dir: Path, env: dict[str, str], readonly_config: Path, export_config: Path, output_root: Path, strategy_map: dict[str, dict[str, Any]], person_identity_requirements: dict[str, dict[str, Any]], object_name: str, cycle_label: str, snapshot_date: str, retry_count: int, skip_export: bool, skip_load: bool, truncate_first: bool, state_dir: Path, source_admin_config: Path | None, target_base_url: str, target_client_id: str, target_client_secret: str, target_username: str, target_password: str, ) -> dict[str, Any]: state = load_object_state(state_dir, object_name) state.update( { "object": object_name, "cycle_label": cycle_label, "updated_at": utc_now().isoformat(), "status": "running", } ) save_object_state(state_dir, object_name, state) if object_name in SOURCE_UNSUPPORTED_OBJECTS: state["status"] = "source_unsupported" state["updated_at"] = utc_now().isoformat() save_object_state(state_dir, object_name, state) return state if skip_export: existing_snapshot_dir = str(state.get("snapshot_dir") or "").strip() if not existing_snapshot_dir: raise ValueError(f"skip-export requires existing snapshot_dir state for {object_name}") snapshot_dir = Path(existing_snapshot_dir) else: snapshot_dir = run_export_for_object( script_dir=script_dir, env=env, readonly_config=readonly_config, export_config=export_config, output_root=output_root, object_name=object_name, cycle_label=cycle_label, snapshot_date=snapshot_date, ) state["snapshot_dir"] = str(snapshot_dir) save_object_state(state_dir, object_name, state) person_identity_resolution = resolve_person_identity_before_schema_mutation( snapshot_dir=snapshot_dir, object_name=object_name, requirement=person_identity_requirements.get(object_name), env=env, ) if person_identity_resolution is not None: state["person_identity_resolution"] = person_identity_resolution save_object_state(state_dir, object_name, state) ensure_table_for_object( script_dir=script_dir, env=env, snapshot_dir=snapshot_dir, export_config=export_config, object_name=object_name, ) if not skip_load: attempt_summaries: list[dict[str, Any]] = [] force_insert_mode = object_name in AUTO_INSERT_OBJECTS object_truncate_first = truncate_first_for_object( requested=truncate_first, object_name=object_name, person_identity_requirements=person_identity_requirements, ) for attempt in range(1, retry_count + 1): selected_mode = "chunked_insert" if force_insert_mode else None plan_path = build_object_load_plan( strategy_map, object_name, state_dir / f"{sanitize_component(object_name)}.load-plan.json", mode_override=selected_mode, person_identity_resolution_requirement=person_identity_requirements.get( object_name ), ) result = load_object_once( script_dir=script_dir, env=env, snapshot_dir=snapshot_dir, export_config=export_config, object_name=object_name, plan_path=plan_path, truncate_first=object_truncate_first, ) attempt_payload = { "attempt": attempt, "mode": selected_mode or str( strategy_map.get(object_name, {}).get("mode") or "chunked_load_data" ), "truncate_first": object_truncate_first, "exit_code": result.returncode, "stdout": result.stdout[-4000:], "stderr": result.stderr[-4000:], } attempt_summaries.append(attempt_payload) state["load_attempts"] = attempt_summaries save_object_state(state_dir, object_name, state) if result.returncode == 0: break combined = f"{result.stdout}\n{result.stderr}" if not force_insert_mode and should_retry_with_insert(combined): force_insert_mode = True continue raise RuntimeError( f"load failed for {object_name} on attempt {attempt}: {combined[-2000:]}" ) if attempt_summaries and attempt_summaries[-1]["exit_code"] != 0: raise RuntimeError(f"load failed for {object_name} after {retry_count} attempts") post_load_hook = maybe_repair_oli_fields( script_dir=script_dir, env=env, snapshot_dir=snapshot_dir, object_name=object_name, source_admin_config=source_admin_config, target_base_url=target_base_url, target_client_id=target_client_id, target_client_secret=target_client_secret, target_username=target_username, target_password=target_password, ) if post_load_hook is not None: state["post_load_hook"] = post_load_hook save_object_state(state_dir, object_name, state) state["status"] = "loaded" if skip_load else "post_reconcile_pending" save_object_state(state_dir, object_name, state) return state def post_reconcile_one_object( script_dir: Path, env: dict[str, str], snapshot_root: Path, monitor_config: Path, export_config: Path, object_name: str, output_json: Path, ) -> dict[str, Any]: payload = run_reconcile( script_dir=script_dir, env=env, snapshot_root=snapshot_root, monitor_config=monitor_config, export_config=export_config, objects=[object_name], output_json=output_json, ) objects = payload.get("objects", []) if not objects: raise RuntimeError(f"missing reconcile result for {object_name}") return dict(objects[0]) def main() -> int: args = parse_args() script_dir = Path(__file__).resolve().parent snapshot_root = Path(args.snapshot_root).expanduser().resolve() output_root = Path(args.output_root or args.snapshot_root).expanduser().resolve() monitor_config = Path(args.monitor_config).expanduser().resolve() export_config = Path(args.export_config).expanduser().resolve() readonly_config = Path(args.readonly_config).expanduser().resolve() source_admin_config = Path(args.source_admin_config).expanduser().resolve() if nonempty(args.source_admin_config) else None allow = parse_items(args.object) deny = parse_items(args.skip_object) load_plan_paths = [ Path(path).expanduser().resolve() for path in (args.load_plan_config or [str(script_dir / item) for item in DEFAULT_LOAD_PLAN_CONFIGS]) ] strategy_map = merge_strategy_maps(load_plan_paths) person_identity_requirements = merge_person_identity_resolution_requirements( load_plan_paths ) env = build_env(args) cycle_label = args.run_label or now_label() snapshot_date = today_utc() state_dir = Path(args.state_dir or (output_root / "_repair_state")).expanduser().resolve() state_dir.mkdir(parents=True, exist_ok=True) if not all([args.db_host, args.db_user, args.db_database, args.db_password]): raise ValueError("db host/user/password/database are required") summary: dict[str, Any] = { "cycle_label": cycle_label, "snapshot_root": str(snapshot_root), "output_root": str(output_root), "state_dir": str(state_dir), "started_at": utc_now().isoformat(), "objects": [], "status": "running", } with tempfile.TemporaryDirectory(prefix="prod-repair-cycle-") as temp_dir: temp_root = Path(temp_dir) pre_reconcile_json = temp_root / "pre_reconcile.json" pre_payload = run_reconcile( script_dir=script_dir, env=env, snapshot_root=snapshot_root, monitor_config=monitor_config, export_config=export_config, objects=sorted(allow) if allow else None, output_json=pre_reconcile_json, ) mismatches = select_mismatches(pre_payload, allow, deny) summary["pre_reconcile_json"] = str(pre_reconcile_json) summary["mismatch_objects"] = mismatches if not mismatches: summary["status"] = "ok" summary["finished_at"] = utc_now().isoformat() if args.summary_json: write_json(Path(args.summary_json).expanduser().resolve(), summary) print(json.dumps(summary, indent=2)) if args.sleep_seconds_on_empty > 0: import time time.sleep(args.sleep_seconds_on_empty) return 0 for object_name in mismatches: object_result: dict[str, Any] = {"object": object_name, "status": "pending"} try: state = repair_one_object( script_dir=script_dir, env=env, readonly_config=readonly_config, export_config=export_config, output_root=output_root, strategy_map=strategy_map, person_identity_requirements=person_identity_requirements, object_name=object_name, cycle_label=cycle_label, snapshot_date=snapshot_date, retry_count=args.retry_count, skip_export=args.skip_export, skip_load=args.skip_load, truncate_first=args.truncate_first, state_dir=state_dir, source_admin_config=source_admin_config, target_base_url=nonempty(args.target_base_url), target_client_id=nonempty(args.target_client_id), target_client_secret=nonempty(args.target_client_secret), target_username=nonempty(args.target_username), target_password=nonempty(args.target_password), ) object_result["snapshot_dir"] = state.get("snapshot_dir") object_result["load_attempts"] = state.get("load_attempts", []) if "person_identity_resolution" in state: object_result["person_identity_resolution"] = state[ "person_identity_resolution" ] if "post_load_hook" in state: object_result["post_load_hook"] = state["post_load_hook"] if state.get("status") == "source_unsupported": object_result["status"] = "source_unsupported" summary["objects"].append(object_result) if args.summary_json: write_json(Path(args.summary_json).expanduser().resolve(), summary) continue if args.skip_post_reconcile: object_result["status"] = "loaded" else: post_json = temp_root / f"post_{sanitize_component(object_name)}.json" post_item = post_reconcile_one_object( script_dir=script_dir, env=env, snapshot_root=snapshot_root, monitor_config=monitor_config, export_config=export_config, object_name=object_name, output_json=post_json, ) object_result["post_reconcile_json"] = str(post_json) object_result["post_reconcile"] = post_item object_result["status"] = str(post_item.get("status") or "unknown") state["post_reconcile"] = post_item state["status"] = object_result["status"] state["updated_at"] = utc_now().isoformat() save_object_state(state_dir, object_name, state) if object_result["status"] == "ok": object_result["finished_at"] = utc_now().isoformat() except Exception as exc: object_result["status"] = "failed" object_result["error"] = str(exc) failed_state = load_object_state(state_dir, object_name) failed_state["status"] = "failed" failed_state["error"] = str(exc) failed_state["updated_at"] = utc_now().isoformat() save_object_state(state_dir, object_name, failed_state) summary["objects"].append(object_result) if args.summary_json: write_json(Path(args.summary_json).expanduser().resolve(), summary) remaining = [ item["object"] for item in summary["objects"] if item.get("status") not in {"ok", "loaded"} ] summary["remaining_mismatches"] = remaining summary["finished_at"] = utc_now().isoformat() summary["status"] = "ok" if not remaining else "mismatch" if args.summary_json: write_json(Path(args.summary_json).expanduser().resolve(), summary) print(json.dumps(summary, indent=2)) return 0 if summary["status"] == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())