373 lines
15 KiB
Python
373 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Crea en cada sucursal los custom fields que existen en Marca pero le faltan.
|
|
|
|
Contraparte de la categoría CREATE_IN_BRANCH del audit (audit_custom_fields_*).
|
|
Toma Marca como referencia y, para cada cuenta destino, detecta los custom fields
|
|
de Marca ausentes (match por fieldKey, fallback nombre normalizado) y los crea con
|
|
la MISMA definición:
|
|
- name, dataType, placeholder
|
|
- options (si es picklist)
|
|
- folder homólogo (parentId resuelto por NOMBRE: los ids difieren por cuenta)
|
|
- position de Marca
|
|
|
|
Caso de uso inmediato (mayo 2026): el campo opportunity "ID Oportunidad Sucursal"
|
|
(TEXT, folder "Opportunity Details") se creó en Marca y debe existir en TODAS las
|
|
cuentas. Es el único MISSING_IN_BRANCH que reporta la auditoría.
|
|
|
|
Verificado contra la API V2 de GHL:
|
|
- `POST /locations/{loc}/customFields` con `{name, dataType, model, options?,
|
|
placeholder?, position?, parentId?}` → crea el field. El fieldKey lo deriva GHL
|
|
del nombre, así que si el nombre es idéntico el fieldKey coincide entre cuentas.
|
|
|
|
Idempotente: si el campo ya existe en la cuenta (por fieldKey o nombre), se omite.
|
|
Sin `--apply` corre en DRY-RUN.
|
|
|
|
Uso:
|
|
python scripts/create_missing_brand_fields.py --location <id> # dry-run
|
|
python scripts/create_missing_brand_fields.py --all # dry-run global
|
|
python scripts/create_missing_brand_fields.py --demos --apply --run-id <uuid>
|
|
python scripts/create_missing_brand_fields.py --all --apply --run-id <uuid>
|
|
# Acotar a un campo concreto (por nombre normalizado):
|
|
python scripts/create_missing_brand_fields.py --all --only "ID Oportunidad Sucursal"
|
|
"""
|
|
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
import warnings
|
|
|
|
warnings.filterwarnings("ignore", message=r"urllib3 .* doesn't match a supported version!")
|
|
import requests
|
|
|
|
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)
|
|
|
|
import sync_engine # noqa: E402
|
|
import script_audit # noqa: E402
|
|
from paths import MIGRATIONS_DIR # noqa: E402
|
|
|
|
BASE_URL = "https://services.leadconnectorhq.com"
|
|
API_VERSION = "2021-07-28"
|
|
BRAND_LOCATION_ID = "GbKkBpCmKu2QmloKFHy3"
|
|
DEMO_LOCATION_IDS = ["Vf7qQl3L9vakJ8hDtQ8e", "Z64WQKORPVwXb5mn68Ef"]
|
|
PICKLIST_TYPES = ("SINGLE_OPTIONS", "MULTIPLE_OPTIONS", "RADIO", "CHECKBOX")
|
|
|
|
_last_request_by_token: dict = {}
|
|
|
|
|
|
def normalize_name(value):
|
|
s = str(value or "").strip().lower()
|
|
s = unicodedata.normalize("NFKD", s)
|
|
s = "".join(c for c in s if not unicodedata.combining(c))
|
|
return " ".join(s.split())
|
|
|
|
|
|
def wait_for_rate_limit(token):
|
|
elapsed = time.time() - _last_request_by_token.get(token, 0)
|
|
if elapsed < 0.110:
|
|
time.sleep(0.110 - elapsed)
|
|
_last_request_by_token[token] = time.time()
|
|
|
|
|
|
def ghl_request(method, endpoint, token, *, params=None, json_body=None,
|
|
version=API_VERSION, retries=3):
|
|
wait_for_rate_limit(token)
|
|
url = endpoint if endpoint.startswith("http") else f"{BASE_URL}{endpoint}"
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {token}",
|
|
"Version": version,
|
|
"Content-Type": "application/json",
|
|
}
|
|
last_exc = None
|
|
for attempt in range(retries):
|
|
try:
|
|
resp = requests.request(method, url, headers=headers, params=params,
|
|
json=json_body, timeout=30)
|
|
except requests.RequestException as exc:
|
|
last_exc = exc
|
|
time.sleep(0.5 * (attempt + 1))
|
|
continue
|
|
if resp.status_code == 429:
|
|
time.sleep(1.0 * (attempt + 1))
|
|
continue
|
|
if resp.status_code >= 500:
|
|
time.sleep(1.0 * (attempt + 1))
|
|
continue
|
|
resp.raise_for_status()
|
|
if resp.status_code == 204:
|
|
return {}
|
|
return resp.json()
|
|
if last_exc:
|
|
raise last_exc
|
|
|
|
|
|
def extract_option_labels(field):
|
|
raw = field.get("options")
|
|
if isinstance(raw, list) and raw:
|
|
labels = []
|
|
for opt in raw:
|
|
if isinstance(opt, dict):
|
|
lbl = opt.get("label")
|
|
if lbl is None:
|
|
lbl = opt.get("key")
|
|
if lbl is not None:
|
|
labels.append(lbl)
|
|
elif opt is not None:
|
|
labels.append(str(opt))
|
|
return labels
|
|
legacy = field.get("picklistOptions")
|
|
if isinstance(legacy, list):
|
|
return [str(o) for o in legacy if o is not None]
|
|
return []
|
|
|
|
|
|
def list_fields_and_folders(account, model):
|
|
"""Devuelve (fields, folders_by_id) del endpoint legacy."""
|
|
data = ghl_request("GET", f"/locations/{account['location_id']}/customFields",
|
|
account["token"], params={"model": model})
|
|
items = data.get("customFields") or []
|
|
fields = [f for f in items if f.get("documentType") != "folder"]
|
|
parent_ids = {f.get("parentId") for f in fields if f.get("parentId")}
|
|
folders_by_id = {}
|
|
for pid in parent_ids:
|
|
try:
|
|
d = ghl_request("GET", f"/locations/{account['location_id']}/customFields/{pid}",
|
|
account["token"])
|
|
cf = d.get("customField") or d
|
|
if cf.get("documentType") == "folder":
|
|
folders_by_id[pid] = cf
|
|
except Exception as exc:
|
|
print(f" ⚠ No pude leer folder {pid}: {exc}")
|
|
return fields, folders_by_id
|
|
|
|
|
|
def build_brand_reference(brand, model, only_norm=None):
|
|
"""Lista de defs de Marca a clonar. Cada def: {name, fieldKey, dataType,
|
|
options, placeholder, folder_name, position}."""
|
|
fields, folders_by_id = list_fields_and_folders(brand, model)
|
|
out = []
|
|
for f in fields:
|
|
norm = normalize_name(f.get("name"))
|
|
if only_norm and norm != only_norm:
|
|
continue
|
|
folder = folders_by_id.get(f.get("parentId")) or {}
|
|
out.append({
|
|
"name": f.get("name"),
|
|
"fieldKey": f.get("fieldKey"),
|
|
"norm": norm,
|
|
"dataType": f.get("dataType"),
|
|
"options": extract_option_labels(f),
|
|
"placeholder": f.get("placeholder") or "",
|
|
"folder_name": folder.get("name"),
|
|
"folder_norm": normalize_name(folder.get("name")),
|
|
"position": f.get("position"),
|
|
"model": model,
|
|
})
|
|
return out
|
|
|
|
|
|
def plan_account_model(account, model, brand_defs):
|
|
"""Acciones para crear en `account` los campos de Marca ausentes."""
|
|
actions = []
|
|
fields, folders_by_id = list_fields_and_folders(account, model)
|
|
have_fk = {f.get("fieldKey") for f in fields if f.get("fieldKey")}
|
|
have_norm = {normalize_name(f.get("name")) for f in fields}
|
|
folder_id_by_norm = {normalize_name(fo.get("name")): pid
|
|
for pid, fo in folders_by_id.items()}
|
|
|
|
for d in brand_defs:
|
|
if d["fieldKey"] in have_fk or d["norm"] in have_norm:
|
|
actions.append({"status": "already_exists", "object": model, "name": d["name"]})
|
|
continue
|
|
target_pid = folder_id_by_norm.get(d["folder_norm"]) if d["folder_norm"] else None
|
|
actions.append({
|
|
"status": "to_create", "object": model, "name": d["name"],
|
|
"dataType": d["dataType"], "options": d["options"],
|
|
"placeholder": d["placeholder"],
|
|
"folder_name": d["folder_name"], "parentId": target_pid,
|
|
"position": d["position"],
|
|
"folder_resolved": target_pid is not None or not d["folder_norm"],
|
|
"def": d,
|
|
})
|
|
return actions
|
|
|
|
|
|
def apply_actions(account, actions, *, dry_run, run_id):
|
|
stats = {"created": 0, "skipped": 0, "errors": []}
|
|
snap = {
|
|
"account": account["nombre"], "location_id": account["location_id"],
|
|
"timestamp_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"dry_run": dry_run, "actions": actions,
|
|
}
|
|
os.makedirs(MIGRATIONS_DIR, exist_ok=True)
|
|
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
safe = "".join(c if c.isalnum() else "_" for c in account["nombre"])[:40]
|
|
snap_path = os.path.join(MIGRATIONS_DIR, f"create_missing_{safe}_{ts}.json")
|
|
with open(snap_path, "w", encoding="utf-8") as fh:
|
|
json.dump(snap, fh, ensure_ascii=False, indent=2, default=str)
|
|
|
|
for a in actions:
|
|
if a["status"] != "to_create":
|
|
stats["skipped"] += 1
|
|
continue
|
|
if dry_run:
|
|
continue
|
|
body = {"name": a["name"], "dataType": a["dataType"], "model": a["object"]}
|
|
if a["dataType"] in PICKLIST_TYPES and a["options"]:
|
|
body["options"] = a["options"]
|
|
if a["placeholder"]:
|
|
body["placeholder"] = a["placeholder"]
|
|
if a["position"] is not None:
|
|
body["position"] = a["position"]
|
|
if a["parentId"]:
|
|
body["parentId"] = a["parentId"]
|
|
|
|
change_id = None
|
|
if run_id:
|
|
change_id = script_audit.record_change(
|
|
run_id, account["location_id"], "__schema__",
|
|
"NEW", "NEW", a["name"], None, body,
|
|
)
|
|
try:
|
|
resp = ghl_request("POST",
|
|
f"/locations/{account['location_id']}/customFields",
|
|
account["token"], json_body=body)
|
|
new_field = resp.get("customField") or resp
|
|
if change_id:
|
|
# Guardar el id del campo recién creado para permitir rollback (DELETE).
|
|
script_audit.mark_change(change_id, "applied")
|
|
a["created_id"] = new_field.get("id")
|
|
stats["created"] += 1
|
|
except Exception as exc:
|
|
if change_id:
|
|
script_audit.mark_change(change_id, "failed", str(exc))
|
|
stats["errors"].append({"name": a["name"], "error": str(exc)})
|
|
|
|
# Reescribir snapshot con ids creados
|
|
if not dry_run:
|
|
with open(snap_path, "w", encoding="utf-8") as fh:
|
|
json.dump(snap, fh, ensure_ascii=False, indent=2, default=str)
|
|
return stats, snap_path
|
|
|
|
|
|
def print_plan(account, actions):
|
|
to_create = [a for a in actions if a["status"] == "to_create"]
|
|
exists = sum(1 for a in actions if a["status"] == "already_exists")
|
|
print(f"\n[Plan] {account['nombre']} ({account['location_id']})")
|
|
print(f" ya existen={exists} a crear={len(to_create)}")
|
|
for a in to_create:
|
|
folder_note = a["folder_name"] if a["folder_resolved"] else f"{a['folder_name']} (folder NO existe → se crea sin folder)"
|
|
opts = f" options={a['options']}" if a["options"] else ""
|
|
print(f" ▸ CREAR {a['object']}.{a['name']!r} [{a['dataType']}] folder={folder_note!r} pos={a['position']}{opts}")
|
|
|
|
|
|
def select_targets(args, accounts):
|
|
if args.location:
|
|
m = [a for a in accounts if a["location_id"] == args.location]
|
|
if not m:
|
|
raise SystemExit(f"Location {args.location} no encontrada.")
|
|
return m
|
|
if args.demos:
|
|
return [a for a in accounts if a["location_id"] in DEMO_LOCATION_IDS]
|
|
if args.all_productive:
|
|
return [a for a in accounts
|
|
if a["location_id"] not in DEMO_LOCATION_IDS
|
|
and a["location_id"] != BRAND_LOCATION_ID]
|
|
if args.all:
|
|
return [a for a in accounts if a["location_id"] != BRAND_LOCATION_ID]
|
|
raise SystemExit("Especifica --location <id>, --demos, --all-productive o --all.")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Crea en cada sucursal los custom fields de Marca que le faltan.")
|
|
parser.add_argument("--location", help="Procesar solo esta cuenta.")
|
|
parser.add_argument("--demos", action="store_true", help="Procesar solo las 2 cuentas DEMO.")
|
|
parser.add_argument("--all-productive", dest="all_productive", action="store_true",
|
|
help="Todas las sucursales productivas (excluye Marca y DEMOs).")
|
|
parser.add_argument("--all", action="store_true", help="Todas las sucursales (excluye Marca).")
|
|
parser.add_argument("--object", choices=["contact", "opportunity", "both"], default="both",
|
|
help="Objeto a procesar. Default both.")
|
|
parser.add_argument("--only", help="Acotar a un solo campo por nombre (normalizado).")
|
|
parser.add_argument("--apply", action="store_true",
|
|
help="Ejecuta las creaciones. Sin este flag corre en DRY-RUN.")
|
|
parser.add_argument("--run-id", help="ID para registrar en script_audit (rollback).")
|
|
args = parser.parse_args()
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
dry_run = not args.apply
|
|
models = ["contact", "opportunity"] if args.object == "both" else [args.object]
|
|
only_norm = normalize_name(args.only) if args.only else None
|
|
accounts = sync_engine.parse_accounts_csv()
|
|
brand = next((a for a in accounts if a["location_id"] == BRAND_LOCATION_ID), None)
|
|
if not brand:
|
|
raise SystemExit(f"Cuenta de Marca {BRAND_LOCATION_ID} no encontrada en el CSV.")
|
|
targets = select_targets(args, accounts)
|
|
|
|
if args.run_id:
|
|
script_audit.create_run(
|
|
args.run_id, "create_missing_brand_fields.py",
|
|
arguments=" ".join(sys.argv[1:]),
|
|
locations=[t["location_id"] for t in targets],
|
|
)
|
|
|
|
print(f"Modo: {'DRY-RUN' if dry_run else 'APPLY'}")
|
|
print(f"Objetos: {', '.join(models)}" + (f" | solo: {args.only!r}" if only_norm else ""))
|
|
print(f"Cuentas en scope: {len(targets)}")
|
|
|
|
print(f"\n[BRAND] {brand['nombre']} ({BRAND_LOCATION_ID})")
|
|
brand_defs = {}
|
|
for model in models:
|
|
brand_defs[model] = build_brand_reference(brand, model, only_norm)
|
|
names = [d["name"] for d in brand_defs[model]]
|
|
print(f" {model}: {len(names)} campos de referencia" + (f" → {names}" if only_norm else ""))
|
|
|
|
total_created = total_skipped = 0
|
|
errors_global = []
|
|
for acc in targets:
|
|
if not script_audit.wait_if_paused_or_stopped(args.run_id):
|
|
print("Detención solicitada. Saliendo.")
|
|
break
|
|
all_actions = []
|
|
try:
|
|
for model in models:
|
|
all_actions.extend(plan_account_model(acc, model, brand_defs[model]))
|
|
except Exception as exc:
|
|
print(f" [{acc['nombre']}] ERROR plan: {exc}")
|
|
errors_global.append({"account": acc["nombre"], "error": str(exc)})
|
|
continue
|
|
print_plan(acc, all_actions)
|
|
try:
|
|
stats, snap_path = apply_actions(acc, all_actions, dry_run=dry_run, run_id=args.run_id)
|
|
total_created += stats["created"]
|
|
total_skipped += stats["skipped"]
|
|
errors_global.extend(stats["errors"])
|
|
print(f" snapshot: {snap_path}")
|
|
if not dry_run:
|
|
print(f" ✓ created={stats['created']} skipped={stats['skipped']} errors={len(stats['errors'])}")
|
|
except Exception as exc:
|
|
print(f" ✗ apply falló: {exc}")
|
|
errors_global.append({"account": acc["nombre"], "error": str(exc)})
|
|
|
|
print("\n" + "=" * 60)
|
|
if dry_run:
|
|
print("DRY-RUN — no se creó nada. Revisa el plan y vuelve a correr con --apply.")
|
|
print(f"Cuentas procesadas : {len(targets)}")
|
|
print(f"Campos creados : {total_created}")
|
|
print(f"Acciones saltadas : {total_skipped}")
|
|
print(f"Errores : {len(errors_global)}")
|
|
for e in errors_global[:20]:
|
|
print(f" - {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|