581 lines
26 KiB
Python
581 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
cleanup_cross_branch_duplicates.py
|
|
|
|
Limpieza generalizada de contactos en Marca que existen fisicamente en mas de
|
|
una sucursal (duplicados cruzados). Es la version generica del script puntual
|
|
cleanup_puebla_qro_duplicates.py.
|
|
|
|
Reglas de decision para elegir el lado a CONSERVAR:
|
|
|
|
1. VALOR monetario: el lado con mayor suma de monetary_value en sus opps gana.
|
|
- Si solo un lado tiene valor > 0, ese se conserva.
|
|
2. TIENDA en Marca: si hay empate de valor (incluido el caso "ambos en $0"),
|
|
se conserva el lado cuya sucursal coincide con el campo TIENDA del
|
|
contacto Marca (resuelto via el verificador de sucursales).
|
|
3. AMBIGUO: si ninguna regla resuelve, el caso se marca needs_review y NO
|
|
se procesa automaticamente. Hay que correrlo con --include-ambiguous y
|
|
explicitar --keeper-loc o --only-contact para forzar la decision manual.
|
|
|
|
Para cada par decidido:
|
|
- En la(s) sucursal(es) PERDEDORA(S): elimina la opp residual y luego el
|
|
contacto residual (orden importante para que el audit log mantenga el
|
|
contact_id de la opp aunque GHL cascadee).
|
|
- En MARCA: si la opp Marca tiene monetary_value o status distintos al
|
|
lado ganador, hace PUT para sincronizar (la opp Marca pudo haberse
|
|
quedado pegada a los datos del lado perdedor en la ultima sync).
|
|
|
|
Reglas de seguridad:
|
|
- Dry-run por default. --apply para escribir.
|
|
- Recalcula el plan desde DB en cada corrida (nunca consume JSON externo).
|
|
- audit con run_id; los DELETE registran snapshot completo en old_value_json.
|
|
- --exclude-contact ID para excluir casos especificos (p.ej. miguel angel).
|
|
- --only-contact ID para procesar un solo caso.
|
|
- --skip-pair "A,B" para excluir todos los casos cuyo par sea exactamente
|
|
esas dos sucursales (case-insensitive, por nombre corto).
|
|
|
|
Uso:
|
|
python scripts/cleanup_cross_branch_duplicates.py # dry-run de todo
|
|
python scripts/cleanup_cross_branch_duplicates.py --apply --yes # apply real
|
|
python scripts/cleanup_cross_branch_duplicates.py --exclude-contact <id> # excluir uno
|
|
python scripts/cleanup_cross_branch_duplicates.py --only-contact <id> # uno solo
|
|
python scripts/cleanup_cross_branch_duplicates.py --skip-pair "Eugenia,Temixco"
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import uuid
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if ROOT_DIR not in sys.path:
|
|
sys.path.insert(0, ROOT_DIR)
|
|
|
|
from scripts.audit_brand_vs_branches_totals import ( # noqa: E402
|
|
load_accounts_filtered, load_contacts, load_opps,
|
|
build_contact_index, find_match,
|
|
extract_tienda_from_custom_fields, resolve_tienda_field_id,
|
|
load_verifier, normalize_tienda,
|
|
BRAND_LOCATION_ID, DB_PATH,
|
|
)
|
|
import sync_engine # noqa: E402
|
|
import script_audit # noqa: E402
|
|
|
|
SCRIPT_NAME = "cleanup_cross_branch_duplicates.py"
|
|
|
|
|
|
def safe_print(*args, **kwargs):
|
|
sep = kwargs.get("sep", " ")
|
|
end = kwargs.get("end", "\n")
|
|
text = sep.join(str(a) for a in args)
|
|
encoding = sys.stdout.encoding or "utf-8"
|
|
try:
|
|
sys.stdout.write(text + end)
|
|
sys.stdout.flush()
|
|
except UnicodeEncodeError:
|
|
sys.stdout.write(text.encode(encoding, errors="replace").decode(encoding) + end)
|
|
sys.stdout.flush()
|
|
|
|
|
|
def _short_branch_name(full):
|
|
"""Devuelve la parte despues del ultimo '-' del nombre de la sucursal."""
|
|
return (full or "").split("-")[-1].strip()
|
|
|
|
|
|
def _norm_branch(name):
|
|
"""Normalizacion para comparar pares ignorando acentos/case."""
|
|
s = (name or "").lower()
|
|
return (s.replace("é", "e").replace("á", "a").replace("í", "i")
|
|
.replace("ó", "o").replace("ú", "u").strip())
|
|
|
|
|
|
def build_plan(log=safe_print, skip_pairs=None, exclude_contacts=None, only_contacts=None,
|
|
include_ambiguous=False):
|
|
"""Recalcula el plan completo. Devuelve dict:
|
|
{
|
|
decisions: [...], # cada item con keeper + losers + criterio
|
|
ambiguous: [...], # casos sin regla clara (excluidos del apply)
|
|
excluded_by_filter: [...], # casos saltados por skip_pairs/exclude_contacts/only_contacts
|
|
}
|
|
"""
|
|
if not os.path.exists(DB_PATH):
|
|
raise FileNotFoundError(f"No existe {DB_PATH}. Sincroniza primero.")
|
|
|
|
skip_pairs_norm = set()
|
|
for sp in skip_pairs or []:
|
|
parts = [_norm_branch(p) for p in sp.split(",")]
|
|
if len(parts) == 2:
|
|
skip_pairs_norm.add(tuple(sorted(parts)))
|
|
exclude_set = set(exclude_contacts or [])
|
|
only_set = set(only_contacts or [])
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
_brand, branches, _ = load_accounts_filtered(conn)
|
|
brand_contacts = load_contacts(conn, BRAND_LOCATION_ID)
|
|
brand_opps = load_opps(conn, BRAND_LOCATION_ID)
|
|
brand_tienda_field_id = resolve_tienda_field_id(conn, BRAND_LOCATION_ID)
|
|
verifier_by_loc, verifier_by_tienda = load_verifier()
|
|
|
|
branch_idx = {}
|
|
branch_name_by_loc = {}
|
|
branch_opps_by_cid = {}
|
|
branch_contacts_by_loc = {}
|
|
for b in branches:
|
|
loc = b["location_id"]
|
|
branch_name_by_loc[loc] = b["nombre"]
|
|
bc = load_contacts(conn, loc)
|
|
bo = load_opps(conn, loc)
|
|
branch_idx[loc] = build_contact_index(bc)
|
|
branch_contacts_by_loc[loc] = {c["id"]: c for c in bc}
|
|
grp = defaultdict(list)
|
|
for o in bo:
|
|
grp[o["contact_id"]].append(o)
|
|
branch_opps_by_cid[loc] = grp
|
|
|
|
brand_by_id = {c["id"]: c for c in brand_contacts}
|
|
brand_opps_by_cid_marca = defaultdict(list)
|
|
for o in brand_opps:
|
|
brand_opps_by_cid_marca[o["contact_id"]].append(o)
|
|
|
|
contact_branches = defaultdict(list)
|
|
phone_collisions_per_brand = defaultdict(list)
|
|
for c in brand_contacts:
|
|
for loc in branch_idx:
|
|
idxp, idxe, idxn = branch_idx[loc]
|
|
matches, collisions = find_match(
|
|
c, idxp, idxe, idxn, return_collisions=True
|
|
)
|
|
for m in matches:
|
|
contact_branches[c["id"]].append((loc, m["id"]))
|
|
for m in collisions:
|
|
phone_collisions_per_brand[c["id"]].append((loc, m["id"]))
|
|
|
|
decisions = []
|
|
ambiguous = []
|
|
excluded_by_filter = []
|
|
# Colisiones de teléfono: contactos Marca con phone idéntico al de
|
|
# algún contacto en sucursal pero con nombre divergente. No los
|
|
# tratamos como duplicados (probablemente son personas distintas
|
|
# compartiendo número, ej. una pareja). Se reportan aparte.
|
|
phone_collisions = []
|
|
for bid, hits in phone_collisions_per_brand.items():
|
|
if not hits:
|
|
continue
|
|
c = brand_by_id.get(bid)
|
|
if not c:
|
|
continue
|
|
phone_collisions.append({
|
|
"brand_contact_id": bid,
|
|
"name": (f"{c.get('first_name') or ''} {c.get('last_name') or ''}".strip() or "(sin nombre)"),
|
|
"phone": c.get("phone") or "",
|
|
"colliding_branches": [
|
|
{
|
|
"location_id": loc,
|
|
"location_name": branch_name_by_loc.get(loc, loc),
|
|
"branch_contact_id": bcid,
|
|
}
|
|
for loc, bcid in hits
|
|
],
|
|
})
|
|
|
|
for bid, locs in contact_branches.items():
|
|
distinct_locs = {l[0] for l in locs}
|
|
if len(distinct_locs) < 2:
|
|
continue # intra-sucursal no aplica aqui (otro script)
|
|
|
|
c = brand_by_id[bid]
|
|
name = (f"{c.get('first_name') or ''} {c.get('last_name') or ''}".strip() or "(sin nombre)")
|
|
tienda_raw = extract_tienda_from_custom_fields(c.get("custom_fields_json"), brand_tienda_field_id) or ""
|
|
tienda_norm = normalize_tienda(tienda_raw) if tienda_raw else ""
|
|
tienda_loc = verifier_by_tienda.get(tienda_norm)
|
|
|
|
# Filtros
|
|
if only_set and bid not in only_set:
|
|
continue
|
|
if bid in exclude_set:
|
|
excluded_by_filter.append({"brand_contact_id": bid, "name": name, "reason": "exclude_contact"})
|
|
continue
|
|
|
|
# Construir candidatos por sucursal
|
|
per = []
|
|
for loc, bcid in locs:
|
|
opps = branch_opps_by_cid[loc].get(bcid, [])
|
|
per.append({
|
|
"loc": loc,
|
|
"bcid": bcid,
|
|
"short_name": _short_branch_name(branch_name_by_loc[loc]),
|
|
"full_name": branch_name_by_loc[loc],
|
|
"n_opps": len(opps),
|
|
"total_val": sum(float(o.get("monetary_value") or 0) for o in opps),
|
|
"opps": opps,
|
|
})
|
|
|
|
# Skip-pair filter (por nombre normalizado de las sucursales del par)
|
|
pair_norm = tuple(sorted({_norm_branch(p["short_name"]) for p in per}))
|
|
if len(pair_norm) == 2 and pair_norm in skip_pairs_norm:
|
|
excluded_by_filter.append({
|
|
"brand_contact_id": bid, "name": name,
|
|
"reason": f"skip_pair {pair_norm[0]}<->{pair_norm[1]}",
|
|
})
|
|
continue
|
|
|
|
# Regla 1: VALOR
|
|
by_val = sorted(per, key=lambda x: -x["total_val"])
|
|
max_val = by_val[0]["total_val"]
|
|
candidates_max = [p for p in per if p["total_val"] == max_val]
|
|
keeper = None
|
|
criterio = None
|
|
if max_val > 0 and len(candidates_max) == 1:
|
|
keeper = candidates_max[0]
|
|
criterio = "VALOR"
|
|
elif max_val > 0 and len(candidates_max) > 1:
|
|
# Empate de valor -> TIENDA decide
|
|
match = [p for p in candidates_max if p["loc"] == tienda_loc]
|
|
if len(match) == 1:
|
|
keeper = match[0]
|
|
criterio = "VAL+TIENDA"
|
|
else:
|
|
# Todos en $0 -> TIENDA decide
|
|
match = [p for p in per if p["loc"] == tienda_loc]
|
|
if len(match) == 1:
|
|
keeper = match[0]
|
|
criterio = "TIENDA"
|
|
|
|
if not keeper:
|
|
amb = {
|
|
"brand_contact_id": bid, "name": name, "tienda": tienda_raw,
|
|
"candidates": [{"loc": p["loc"], "name": p["short_name"],
|
|
"n_opps": p["n_opps"], "total_val": p["total_val"]} for p in per],
|
|
}
|
|
ambiguous.append(amb)
|
|
continue
|
|
|
|
losers = [p for p in per if p["loc"] != keeper["loc"]]
|
|
|
|
# Plan de delete en sucursales perdedoras
|
|
loser_deletions = []
|
|
for l in losers:
|
|
loser_deletions.append({
|
|
"location_id": l["loc"],
|
|
"location_name": l["full_name"],
|
|
"contact_id": l["bcid"],
|
|
"contact_snapshot": {
|
|
"id": l["bcid"],
|
|
"first_name": branch_contacts_by_loc[l["loc"]][l["bcid"]].get("first_name"),
|
|
"last_name": branch_contacts_by_loc[l["loc"]][l["bcid"]].get("last_name"),
|
|
"phone": branch_contacts_by_loc[l["loc"]][l["bcid"]].get("phone"),
|
|
"email": branch_contacts_by_loc[l["loc"]][l["bcid"]].get("email"),
|
|
},
|
|
"opps_snapshots": [
|
|
{"id": o["id"], "name": o.get("name"), "status": o.get("status"),
|
|
"monetary_value": o.get("monetary_value"),
|
|
"pipeline_id": o.get("pipeline_id"),
|
|
"pipeline_stage_id": o.get("pipeline_stage_id")}
|
|
for o in l["opps"]
|
|
],
|
|
})
|
|
|
|
# Plan de update en Marca: si la opp Marca difiere del keeper en valor/status
|
|
brand_opp_updates = []
|
|
m_opps = brand_opps_by_cid_marca.get(bid, [])
|
|
k_opps = keeper["opps"]
|
|
if m_opps and k_opps:
|
|
m_o = m_opps[0]
|
|
k_o = k_opps[0]
|
|
m_val = float(m_o.get("monetary_value") or 0)
|
|
k_val = float(k_o.get("monetary_value") or 0)
|
|
m_st = (m_o.get("status") or "").lower()
|
|
k_st = (k_o.get("status") or "").lower()
|
|
if m_val != k_val or m_st != k_st:
|
|
brand_opp_updates.append({
|
|
"brand_opp_id": m_o["id"],
|
|
"old": {"monetary_value": m_val, "status": m_st},
|
|
"new": {"monetary_value": k_val, "status": k_st},
|
|
})
|
|
|
|
decisions.append({
|
|
"brand_contact_id": bid,
|
|
"name": name,
|
|
"tienda": tienda_raw,
|
|
"criterio": criterio,
|
|
"keeper": {
|
|
"location_id": keeper["loc"],
|
|
"location_name": keeper["full_name"],
|
|
"short_name": keeper["short_name"],
|
|
"total_val": keeper["total_val"],
|
|
},
|
|
"losers": loser_deletions,
|
|
"brand_opp_updates": brand_opp_updates,
|
|
})
|
|
|
|
return {
|
|
"decisions": decisions,
|
|
"ambiguous": ambiguous,
|
|
"excluded_by_filter": excluded_by_filter,
|
|
"phone_collisions": phone_collisions,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _empty_summary():
|
|
return {
|
|
"decisions": 0,
|
|
"ambiguous_skipped": 0,
|
|
"filter_excluded": 0,
|
|
"phone_collisions_unresolved": 0,
|
|
"loser_contacts_to_delete": 0,
|
|
"loser_opps_to_delete": 0,
|
|
"brand_opps_to_update": 0,
|
|
"loser_contacts_deleted": 0,
|
|
"loser_opps_deleted": 0,
|
|
"brand_opps_updated": 0,
|
|
"errors": 0,
|
|
}
|
|
|
|
|
|
def run_cleanup(dry_run=True, log=None, run_id=None,
|
|
skip_pairs=None, exclude_contacts=None, only_contacts=None,
|
|
include_ambiguous=False):
|
|
if log is None:
|
|
log = safe_print
|
|
|
|
plan = build_plan(
|
|
log=log, skip_pairs=skip_pairs,
|
|
exclude_contacts=exclude_contacts, only_contacts=only_contacts,
|
|
include_ambiguous=include_ambiguous,
|
|
)
|
|
|
|
summary = _empty_summary()
|
|
summary["decisions"] = len(plan["decisions"])
|
|
summary["ambiguous_skipped"] = len(plan["ambiguous"])
|
|
summary["filter_excluded"] = len(plan["excluded_by_filter"])
|
|
summary["phone_collisions_unresolved"] = len(plan.get("phone_collisions") or [])
|
|
summary["loser_contacts_to_delete"] = sum(len(d["losers"]) for d in plan["decisions"])
|
|
summary["loser_opps_to_delete"] = sum(len(l["opps_snapshots"]) for d in plan["decisions"] for l in d["losers"])
|
|
summary["brand_opps_to_update"] = sum(len(d["brand_opp_updates"]) for d in plan["decisions"])
|
|
|
|
log(f"[{datetime.now().strftime('%H:%M:%S')}] === cleanup_cross_branch_duplicates ===")
|
|
log(f"Modo: {'DRY-RUN (no escribe)' if dry_run else 'APPLY (escribe en GHL)'}")
|
|
log(f"Decisiones automaticas: {summary['decisions']} | ambiguos saltados: {summary['ambiguous_skipped']} | filtrados: {summary['filter_excluded']}")
|
|
log(f"A eliminar en sucursales perdedoras: {summary['loser_contacts_to_delete']} contactos, {summary['loser_opps_to_delete']} opps")
|
|
log(f"A actualizar en Marca: {summary['brand_opps_to_update']} opps")
|
|
if summary["phone_collisions_unresolved"]:
|
|
log(f"Colisiones telefono sin match por nombre (NO se procesan): {summary['phone_collisions_unresolved']}")
|
|
|
|
if plan.get("phone_collisions"):
|
|
log("\nColisiones de telefono (mismo numero, distinto nombre - revision manual):")
|
|
for col in plan["phone_collisions"]:
|
|
branches_str = ", ".join(
|
|
f"{cb['location_name']}:{cb['branch_contact_id']}" for cb in col["colliding_branches"]
|
|
)
|
|
log(f" - Marca {col['name']:<30} tel={col['phone']:<14} contra: {branches_str}")
|
|
|
|
if plan["ambiguous"]:
|
|
log("\nAmbiguos (NO se procesan):")
|
|
for a in plan["ambiguous"]:
|
|
cands_str = " vs ".join(f"{c['name']}({c['n_opps']}/${c['total_val']:,.0f})" for c in a["candidates"])
|
|
log(f" - {a['name']:<30} TIENDA={a['tienda']} candidatos: {cands_str}")
|
|
|
|
if plan["excluded_by_filter"]:
|
|
log("\nExcluidos por filtro:")
|
|
for e in plan["excluded_by_filter"]:
|
|
log(f" - {e['name']} ({e['reason']})")
|
|
|
|
if not dry_run:
|
|
tokens_map = sync_engine.get_tokens_map()
|
|
brand_token = tokens_map.get(BRAND_LOCATION_ID)
|
|
if not brand_token:
|
|
raise RuntimeError(f"No hay token para Marca {BRAND_LOCATION_ID}")
|
|
client = sync_engine.ghl_client
|
|
|
|
items = []
|
|
for d in plan["decisions"]:
|
|
item = {
|
|
"brand_contact_id": d["brand_contact_id"],
|
|
"name": d["name"],
|
|
"criterio": d["criterio"],
|
|
"keeper": d["keeper"]["short_name"],
|
|
"deletions": [],
|
|
"brand_updates": [],
|
|
"status": "pending",
|
|
"error": None,
|
|
}
|
|
try:
|
|
# ---- Borrar en sucursales perdedoras ----
|
|
for loser in d["losers"]:
|
|
loc = loser["location_id"]
|
|
loc_name = loser["location_name"]
|
|
short = _short_branch_name(loc_name)
|
|
if not dry_run:
|
|
loser_token = tokens_map.get(loc)
|
|
if not loser_token:
|
|
raise RuntimeError(f"No hay token para {loc_name} ({loc})")
|
|
|
|
# Opps de la perdedora
|
|
for opp_snap in loser["opps_snapshots"]:
|
|
opp_id = opp_snap["id"]
|
|
if dry_run:
|
|
item["deletions"].append({"loc": short, "opp_id": opp_id, "status": "would_delete"})
|
|
continue
|
|
cid = script_audit.record_change(
|
|
run_id, loc, "opportunity", opp_id,
|
|
"", "deleted_residual_dup", opp_snap, None,
|
|
) if run_id else None
|
|
try:
|
|
client.delete_opportunity(loser_token, opp_id, loc)
|
|
summary["loser_opps_deleted"] += 1
|
|
item["deletions"].append({"loc": short, "opp_id": opp_id, "status": "deleted"})
|
|
if cid: script_audit.mark_change(cid, "applied")
|
|
except Exception as e:
|
|
summary["errors"] += 1
|
|
item["deletions"].append({"loc": short, "opp_id": opp_id, "status": "error", "error": str(e)})
|
|
if cid: script_audit.mark_change(cid, "failed", error_message=str(e))
|
|
raise
|
|
|
|
# Contacto perdedor
|
|
if dry_run:
|
|
item["deletions"].append({"loc": short, "contact_id": loser["contact_id"], "status": "would_delete"})
|
|
else:
|
|
cid = script_audit.record_change(
|
|
run_id, loc, "contact", loser["contact_id"],
|
|
"", "deleted_residual_dup", loser["contact_snapshot"], None,
|
|
) if run_id else None
|
|
try:
|
|
client.delete_contact(loser_token, loser["contact_id"], loc)
|
|
summary["loser_contacts_deleted"] += 1
|
|
item["deletions"].append({"loc": short, "contact_id": loser["contact_id"], "status": "deleted"})
|
|
if cid: script_audit.mark_change(cid, "applied")
|
|
except Exception as e:
|
|
summary["errors"] += 1
|
|
item["deletions"].append({"loc": short, "contact_id": loser["contact_id"], "status": "error", "error": str(e)})
|
|
if cid: script_audit.mark_change(cid, "failed", error_message=str(e))
|
|
raise
|
|
|
|
# ---- Actualizar opps Marca con datos del keeper ----
|
|
for u in d["brand_opp_updates"]:
|
|
if dry_run:
|
|
item["brand_updates"].append({"opp_id": u["brand_opp_id"], "status": "would_update", **u})
|
|
else:
|
|
cid = script_audit.record_change(
|
|
run_id, BRAND_LOCATION_ID, "opportunity", u["brand_opp_id"],
|
|
"monetary_value+status", "updated_from_keeper",
|
|
u["old"], u["new"],
|
|
) if run_id else None
|
|
try:
|
|
payload = {"monetaryValue": u["new"]["monetary_value"]}
|
|
client.update_opportunity(brand_token, u["brand_opp_id"], payload)
|
|
if u["old"]["status"] != u["new"]["status"]:
|
|
client.update_opportunity_status(brand_token, u["brand_opp_id"], u["new"]["status"])
|
|
summary["brand_opps_updated"] += 1
|
|
item["brand_updates"].append({"opp_id": u["brand_opp_id"], "status": "updated", **u})
|
|
if cid: script_audit.mark_change(cid, "applied")
|
|
except Exception as e:
|
|
summary["errors"] += 1
|
|
item["brand_updates"].append({"opp_id": u["brand_opp_id"], "status": "error", "error": str(e), **u})
|
|
if cid: script_audit.mark_change(cid, "failed", error_message=str(e))
|
|
raise
|
|
|
|
item["status"] = "ok"
|
|
log(f" [{'DRY' if dry_run else 'OK'}] {d['name']:<30} keeper={d['keeper']['short_name']:<12} "
|
|
f"criterio={d['criterio']:<11} losers={','.join(_short_branch_name(l['location_name']) for l in d['losers'])}")
|
|
except Exception as e:
|
|
item["status"] = "error"
|
|
item["error"] = str(e)
|
|
log(f" [ERROR] {d['name']}: {e}")
|
|
|
|
items.append(item)
|
|
|
|
log("\n=== RESUMEN ===")
|
|
for k, v in summary.items():
|
|
log(f" {k:<32}: {v}")
|
|
|
|
return {
|
|
"dry_run": dry_run,
|
|
"summary": summary,
|
|
"items": items,
|
|
"ambiguous": plan["ambiguous"],
|
|
"excluded_by_filter": plan["excluded_by_filter"],
|
|
"phone_collisions": plan.get("phone_collisions") or [],
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("--apply", action="store_true", help="Ejecuta escrituras en GHL. Default dry-run.")
|
|
parser.add_argument("--yes", action="store_true", help="Skip confirmacion interactiva.")
|
|
parser.add_argument("--only-contact", action="append", default=[], help="Procesa solo el brand_contact_id dado.")
|
|
parser.add_argument("--exclude-contact", action="append", default=[], help="Excluye brand_contact_id (repetible).")
|
|
parser.add_argument("--skip-pair", action="append", default=[], help='Excluye casos del par "SucA,SucB" (nombre corto).')
|
|
parser.add_argument("--include-ambiguous", action="store_true", help="Procesa tambien los ambiguos (NO recomendado sin --only-contact).")
|
|
parser.add_argument("--json", action="store_true", help="Imprime resultado como JSON.")
|
|
parser.add_argument("--run-id", type=str, default=None, help="Id de script_audit existente.")
|
|
args = parser.parse_args()
|
|
|
|
dry_run = not args.apply
|
|
if not dry_run and not args.yes:
|
|
safe_print("\nEsto eliminara contactos+opps en sucursales perdedoras y actualizara opps en Marca.")
|
|
safe_print("Los DELETE son destructivos (audit log guarda snapshots para reconstruccion manual).")
|
|
confirm = input("Continuar? (y/N): ").strip().lower()
|
|
if confirm not in ("y", "yes", "s", "si", "sí"):
|
|
safe_print("Cancelado.")
|
|
return 1
|
|
|
|
run_id = args.run_id
|
|
if not dry_run:
|
|
if not run_id:
|
|
run_id = f"ccbd-{uuid.uuid4().hex[:12]}"
|
|
try:
|
|
script_audit.init_audit_db()
|
|
script_audit.create_run(
|
|
run_id, SCRIPT_NAME,
|
|
arguments=f"--apply filters: only={args.only_contact or 'all'} exclude={args.exclude_contact or 'none'} skip_pairs={args.skip_pair or 'none'}",
|
|
locations=[BRAND_LOCATION_ID],
|
|
execution_mode="sequential",
|
|
)
|
|
except Exception as e:
|
|
safe_print(f"[warn] no se pudo iniciar audit run: {e}")
|
|
run_id = None
|
|
|
|
try:
|
|
result = run_cleanup(
|
|
dry_run=dry_run, log=safe_print, run_id=run_id,
|
|
skip_pairs=args.skip_pair or None,
|
|
exclude_contacts=args.exclude_contact or None,
|
|
only_contacts=args.only_contact or None,
|
|
include_ambiguous=args.include_ambiguous,
|
|
)
|
|
except Exception as e:
|
|
if run_id:
|
|
try: script_audit.update_run_status(run_id, "failed", str(e))
|
|
except Exception: pass
|
|
safe_print(f"[FATAL] {e}")
|
|
return 2
|
|
|
|
if run_id:
|
|
try:
|
|
errors = result["summary"]["errors"]
|
|
success_any = (result["summary"]["loser_contacts_deleted"]
|
|
+ result["summary"]["loser_opps_deleted"]
|
|
+ result["summary"]["brand_opps_updated"]) > 0
|
|
status = "failed" if errors and not success_any else "success"
|
|
script_audit.update_run_status(run_id, status)
|
|
except Exception:
|
|
pass
|
|
result["run_id"] = run_id
|
|
|
|
if args.json:
|
|
safe_print(json.dumps(result, default=str, ensure_ascii=False, indent=2))
|
|
return 0 if result["summary"]["errors"] == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main() or 0)
|