#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any from incremental_sync_state import atomic_write_json, sanitize_error_category_counts DEFAULT_SUMMARY_NAME = "hourly_delta_replay_queue.json" DEFAULT_MAX_ENTRIES = 4 DEFAULT_MAX_ATTEMPTS = 5 def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Consume the durable hourly replay queue for Lead assignment replay." ) parser.add_argument("--replay-queue-file", required=True, help="Durable replay queue file.") parser.add_argument("--base-url", required=True, help="Target NexusCRM base URL.") parser.add_argument("--username", required=True, help="Target NexusCRM username.") parser.add_argument("--password", help="Target NexusCRM password.") parser.add_argument( "--password-env", help="Environment variable containing the target NexusCRM password.", ) parser.add_argument("--output-root", required=True, help="Root output directory for replay summaries.") parser.add_argument("--summary-path", help="Optional batch replay summary path.") parser.add_argument( "--entry-id", help="Optional exact queue entry ID to consume; used for cursor-coupled replay.", ) parser.add_argument("--max-entries", type=int, default=DEFAULT_MAX_ENTRIES, help="Maximum pending replay entries to process.") parser.add_argument( "--max-attempts", type=int, default=DEFAULT_MAX_ATTEMPTS, help="Consecutive no-progress attempts before a replay entry becomes dead_letter.", ) parser.add_argument("--api-version", default="v63.0", help="REST API version.") parser.add_argument("--query-timeout-seconds", type=int, default=60, help="Timeout for replay target read/query requests.") parser.add_argument("--patch-timeout-seconds", type=int, default=20, help="Timeout for each replay PATCH request.") parser.add_argument("--patch-retries", type=int, default=2, help="Retry count for each replay PATCH request.") parser.add_argument("--max-workers", type=int, default=8, help="Concurrent worker count for replay PATCH requests.") parser.add_argument("--progress-every", type=int, default=25, help="Emit replay progress every N completed PATCH attempts.") parser.add_argument("--dry-run", action="store_true", help="Print planned queue entries without consuming them.") 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 format_sf_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def record_replay_result( entry: dict[str, Any], *, attempted_lead_ids: list[str], success_ids: set[str], failure_count: int, missing_target_count: int, max_consecutive_no_progress: int, error_category_counts: dict[str, Any] | None = None, ) -> None: if max_consecutive_no_progress <= 0: raise ValueError("max_consecutive_no_progress must be positive") attempted_ids = [ str(item).strip() for item in attempted_lead_ids if str(item).strip() ] attempted_id_set = set(attempted_ids) accepted_success_ids = { str(item).strip() for item in success_ids if str(item).strip() in attempted_id_set } remaining_lead_ids = [ lead_id for lead_id in attempted_ids if lead_id not in accepted_success_ids ] completed_lead_ids = { str(item).strip() for item in entry.get("completed_lead_ids", []) if str(item).strip() } | accepted_success_ids all_candidate_ids = { str(item).strip() for item in entry.get("all_candidate_ids", []) if str(item).strip() } | attempted_id_set | completed_lead_ids entry["all_candidate_ids"] = sorted(all_candidate_ids) entry["completed_lead_ids"] = sorted(completed_lead_ids) entry["original_candidate_count"] = len(all_candidate_ids) entry["lead_ids"] = remaining_lead_ids entry["candidate_count"] = len(remaining_lead_ids) entry["completed_candidate_count"] = len(completed_lead_ids) entry["last_error_category_counts"] = sanitize_error_category_counts( error_category_counts ) if failure_count == 0 and missing_target_count == 0 and not remaining_lead_ids: entry["status"] = "success" entry["consecutive_no_progress_count"] = 0 entry["last_error"] = "" return if len(remaining_lead_ids) < len(attempted_ids): consecutive_no_progress_count = 0 else: consecutive_no_progress_count = ( int(entry.get("consecutive_no_progress_count", 0) or 0) + 1 ) entry["consecutive_no_progress_count"] = consecutive_no_progress_count entry["status"] = ( "dead_letter" if consecutive_no_progress_count >= max_consecutive_no_progress else "failed" ) entry["last_error"] = ( f"replay_failures={failure_count} missing_targets={missing_target_count} " f"remaining_candidates={len(remaining_lead_ids)}" ) def main() -> int: args = parse_args() password_env = str(args.password_env or "").strip() password = str(args.password or "") if password_env: password = os.environ.get(password_env, "") if not password: raise ValueError("target NexusCRM password is required") if args.max_attempts <= 0: raise ValueError("max_attempts must be positive") queue_file = Path(args.replay_queue_file).expanduser().resolve() if not queue_file.exists(): raise FileNotFoundError(f"replay queue file does not exist: {queue_file}") payload = load_json(queue_file) entries = [entry for entry in payload.get("entries", []) if isinstance(entry, dict)] pending_entries = [ entry for entry in entries if str(entry.get("status", "")) in {"pending", "failed"} ] if args.entry_id: pending_entries = [ entry for entry in pending_entries if str(entry.get("id", "")) == args.entry_id ] pending_entries.sort(key=lambda entry: str(entry.get("window_start", ""))) selected_entries = pending_entries[: args.max_entries] output_root = Path(args.output_root).expanduser().resolve() summary_path = Path(args.summary_path).expanduser().resolve() if args.summary_path else output_root / "_replay_manifests" / DEFAULT_SUMMARY_NAME summary: dict[str, Any] = { "manifest_version": 1, "run_type": "hourly_delta_replay_queue", "replay_queue_file": str(queue_file), "selected_entry_count": len(selected_entries), "dry_run": args.dry_run, "entries": [], } if args.dry_run: for entry in selected_entries: summary["entries"].append( { "id": str(entry.get("id", "")), "window_start": str(entry.get("window_start", "")), "window_end": str(entry.get("window_end", "")), "candidate_count": int(entry.get("candidate_count", 0) or 0), "status": str(entry.get("status", "")), } ) write_json(summary_path, summary) print(f"[ok] replay_summary={summary_path}") return 0 replay_script = Path(__file__).resolve().parent / "replay_lead_assignment_from_export.py" for entry in selected_entries: entry_id = str(entry.get("id", "")).strip() lead_ids = [str(item).strip() for item in entry.get("lead_ids", []) if str(item).strip()] entry["attempt_count"] = int(entry.get("attempt_count", 0) or 0) + 1 entry["updated_at"] = format_sf_timestamp(utc_now()) if not lead_ids: entry["status"] = "dead_letter" entry["last_error"] = "empty_lead_ids" summary["entries"].append( { "id": entry_id, "status": "dead_letter", "reason": "empty_lead_ids", } ) continue temp_dir = output_root / "_replay_queue_inputs" / entry_id.replace(":", "_") temp_dir.mkdir(parents=True, exist_ok=True) lead_ids_file = temp_dir / "lead_ids.txt" lead_ids_file.write_text("".join(f"{lead_id}\n" for lead_id in lead_ids), encoding="utf-8") entry_summary_path = temp_dir / "lead_assignment_replay_summary.json" cmd = [ "python3", str(replay_script), "--lead-ids-file", str(lead_ids_file), "--base-url", args.base_url, "--username", args.username, "--api-version", args.api_version, "--description-field", str(entry.get("description_field", "Description")), "--replay-key", entry_id, "--summary-path", str(entry_summary_path), "--query-timeout-seconds", str(args.query_timeout_seconds), "--patch-timeout-seconds", str(args.patch_timeout_seconds), "--patch-retries", str(args.patch_retries), "--max-workers", str(args.max_workers), "--progress-every", str(args.progress_every), ] if password_env: cmd.extend(["--password-env", password_env]) else: cmd.extend(["--password", password]) try: subprocess.run(cmd, check=True) entry_summary = load_json(entry_summary_path) failure_count = int(entry_summary.get("failure_count", 0) or 0) missing_target_count = int(entry_summary.get("missing_target_count", 0) or 0) success_count = int(entry_summary.get("success_count", 0) or 0) raw_success_ids = entry_summary.get("success_ids") success_ids = { str(item).strip() for item in raw_success_ids if str(item).strip() } if isinstance(raw_success_ids, list) else set() if failure_count == 0 and missing_target_count == 0 and success_count == len(lead_ids): success_ids = set(lead_ids) success_ids &= set(lead_ids) record_replay_result( entry, attempted_lead_ids=lead_ids, success_ids=success_ids, failure_count=failure_count, missing_target_count=missing_target_count, max_consecutive_no_progress=args.max_attempts, error_category_counts=entry_summary.get("error_category_counts"), ) remaining_lead_ids = list(entry["lead_ids"]) entry["last_summary_path"] = str(entry_summary_path) summary["entries"].append( { "id": entry_id, "window_start": str(entry.get("window_start", "")), "window_end": str(entry.get("window_end", "")), "candidate_count": len(lead_ids), "success_count": success_count, "remaining_candidate_count": len(remaining_lead_ids), "attempt_count": int(entry.get("attempt_count", 0) or 0), "consecutive_no_progress_count": int( entry.get("consecutive_no_progress_count", 0) or 0 ), "status": str(entry.get("status", "")), "summary_path": str(entry_summary_path), "failure_count": failure_count, "missing_target_count": missing_target_count, "error_category_counts": dict( entry.get("last_error_category_counts", {}) ), } ) except subprocess.CalledProcessError as exc: record_replay_result( entry, attempted_lead_ids=lead_ids, success_ids=set(), failure_count=-1, missing_target_count=-1, max_consecutive_no_progress=args.max_attempts, error_category_counts={"command_failed": 1}, ) entry["last_error"] = f"replay_command_failed exit_code={exc.returncode}" summary["entries"].append( { "id": entry_id, "window_start": str(entry.get("window_start", "")), "window_end": str(entry.get("window_end", "")), "candidate_count": len(lead_ids), "attempt_count": int(entry.get("attempt_count", 0) or 0), "consecutive_no_progress_count": int( entry.get("consecutive_no_progress_count", 0) or 0 ), "status": str(entry.get("status", "")), "summary_path": str(entry_summary_path), "failure_count": -1, "missing_target_count": -1, "error_category_counts": dict( entry.get("last_error_category_counts", {}) ), } ) entry["updated_at"] = format_sf_timestamp(utc_now()) write_json(queue_file, payload) write_json(summary_path, summary) print(f"[ok] replay_summary={summary_path}") print(f"[ok] replay_queue_file={queue_file}") return 0 if __name__ == "__main__": raise SystemExit(main())