#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import shutil import subprocess from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any 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) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Repair Lead assignment replay for recent target Leads still owned by the initial owner." ) parser.add_argument("--output-root", required=True, help="Output directory for repair replay artifacts.") 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", required=True, help="Target NexusCRM password.") parser.add_argument("--initial-owner-id", required=True, help="OwnerId that should be considered pending replay repair.") parser.add_argument("--target-table", default="lead", help="Target Lead table name.") parser.add_argument("--description-field", default="Description", help="Target Lead field to patch.") parser.add_argument("--lookback-hours", type=int, default=48, help="CreatedDate lookback window for repair candidates.") parser.add_argument("--mysql-client", default=MYSQL_CLIENT or "", help="Path to mysql client.") parser.add_argument("--db-host", default=os.environ.get("TIDB_PROD_HOST", os.environ.get("TIDB_HOST", ""))) parser.add_argument("--db-port", type=int, default=int(os.environ.get("TIDB_PROD_PORT", os.environ.get("TIDB_PORT", "4000")))) parser.add_argument("--db-user", default=os.environ.get("TIDB_PROD_USER", os.environ.get("TIDB_USER", ""))) parser.add_argument("--db-password", default=os.environ.get("TIDB_PROD_PASSWORD", os.environ.get("TIDB_PASSWORD", ""))) parser.add_argument("--db-database", default=os.environ.get("TIDB_PROD_DATABASE", os.environ.get("TIDB_DATABASE", ""))) parser.add_argument("--ssl-mode", default=os.environ.get("TIDB_SSL_MODE", "REQUIRED")) parser.add_argument("--summary-path", help="Optional repair replay summary path.") parser.add_argument("--dry-run", action="store_true", help="Write candidate ids without PATCHing.") return parser.parse_args() def utc_now() -> datetime: return datetime.now(timezone.utc) def format_mysql_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).strftime("%Y-%m-%d %H:%M:%S") def format_sf_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") 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 mysql_quote(value: str) -> str: return "'" + str(value).replace("\\", "\\\\").replace("'", "''") + "'" def run_mysql_query(args: argparse.Namespace, sql: str) -> list[str]: cmd = [ args.mysql_client, "--host", args.db_host, "--port", str(args.db_port), "--user", args.db_user, "--database", args.db_database, "--batch", "--raw", "--skip-column-names", "-e", sql, ] cmd[10:10] = mysql_ssl_args(args.mysql_client, args.ssl_mode) result = subprocess.run(cmd, check=False, capture_output=True, text=True, env=mysql_env(args.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 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) + "\n", encoding="utf-8") 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 fetch_repair_candidate_ids(args: argparse.Namespace, since: datetime) -> list[str]: sql = ( f"SELECT `Id` FROM `{args.target_table}` " f"WHERE `OwnerId` = {mysql_quote(args.initial_owner_id)} " f"AND `CreatedDate` >= {mysql_quote(format_mysql_timestamp(since))} " "AND (`IsDeleted` = 0 OR `IsDeleted` IS NULL) " "ORDER BY `CreatedDate`, `Id`" ) return [line.strip() for line in run_mysql_query(args, sql) if line.strip()] def main() -> int: args = parse_args() if args.lookback_hours <= 0: raise ValueError("lookback-hours must be positive") if not args.mysql_client: raise RuntimeError("mysql client not found") if not all([args.db_host, args.db_user, args.db_password, args.db_database]): raise ValueError("db host/user/password/database are required") output_root = Path(args.output_root).expanduser().resolve() output_root.mkdir(parents=True, exist_ok=True) summary_path = Path(args.summary_path).expanduser().resolve() if args.summary_path else output_root / "lead_assignment_repair_replay_summary.json" lead_ids_path = output_root / "lead_assignment_repair_replay_ids.txt" since = utc_now() - timedelta(hours=args.lookback_hours) lead_ids = fetch_repair_candidate_ids(args, since) lead_ids_path.write_text("".join(f"{lead_id}\n" for lead_id in lead_ids), encoding="utf-8") summary: dict[str, Any] = { "run_type": "lead_assignment_repair_replay", "target_table": args.target_table, "initial_owner_id": args.initial_owner_id, "lookback_hours": args.lookback_hours, "since": format_sf_timestamp(since), "candidate_count": len(lead_ids), "candidate_ids_path": str(lead_ids_path), "candidate_ids_sample": lead_ids[:20], "dry_run": args.dry_run, } if not lead_ids or args.dry_run: summary["replay"] = { "status": "skipped", "reason": "dry_run" if args.dry_run else "no_repair_candidates", } write_json(summary_path, summary) print(json.dumps(summary, ensure_ascii=True)) return 0 replay_summary_path = output_root / "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", args.base_url, "--username", args.username, "--password", args.password, "--description-field", args.description_field, "--summary-path", str(replay_summary_path), ] subprocess.run(replay_cmd, check=True) summary["replay"] = load_json(replay_summary_path) summary["completed_at"] = format_sf_timestamp(utc_now()) write_json(summary_path, summary) print(json.dumps(summary, ensure_ascii=True)) return 0 if __name__ == "__main__": raise SystemExit(main())