198 lines
7.4 KiB
Python
198 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""enable_duplicate_opportunity.py
|
|
|
|
Activa el ajuste `settings.allowDuplicateOpportunity = true` en una location de GHL
|
|
(por defecto la cuenta de Marca, GbKkBpCmKu2QmloKFHy3) para que un contacto pueda
|
|
tener multiples oportunidades en el mismo pipeline (necesario para replicar
|
|
multi-empeno Sucursal -> Marca).
|
|
|
|
POR QUE EXISTE:
|
|
La replicacion Sucursal->Marca de multi-empenos fallaba con
|
|
`400 "Can not create duplicate opportunity for the contact"` porque Marca tenia
|
|
el flag en false (las sucursales ya lo tienen en true). Ver memoria
|
|
opp_multiplicity_replication_gap.
|
|
|
|
TOKEN:
|
|
Escribir en /locations/{id} requiere un token de AGENCIA con scope locations.write
|
|
(el token PIT por-location del CSV da 401 en este PUT). Se lee de la variable de
|
|
entorno GHL_AGENCY_TOKEN (definida en .env, gitignored). NUNCA se imprime.
|
|
|
|
SEGURIDAD:
|
|
- El PUT envia SOLO {"settings": {"allowDuplicateOpportunity": <bool>}}; GHL hace
|
|
merge y preserva el resto de settings (crmSettings, saasSettings, etc.).
|
|
- Snapshot del settings actual en generated/migrations/ antes de mutar.
|
|
- Registro en script_audit (object_type 'location'). OJO: el rollback automatico
|
|
del dashboard (rollback_run) NO revierte settings de location -- el rollback es
|
|
manual con --disable (PUT allowDuplicateOpportunity=false) desde el snapshot.
|
|
|
|
Uso:
|
|
python scripts/enable_duplicate_opportunity.py # DRY-RUN (no escribe)
|
|
python scripts/enable_duplicate_opportunity.py --apply --run-id <uuid>
|
|
python scripts/enable_duplicate_opportunity.py --location <id> [--apply]
|
|
python scripts/enable_duplicate_opportunity.py --disable --apply --run-id <uuid> # rollback
|
|
python scripts/enable_duplicate_opportunity.py --json
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import uuid
|
|
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)
|
|
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(os.path.join(ROOT_DIR, ".env"))
|
|
except Exception:
|
|
pass
|
|
|
|
import requests # noqa: E402
|
|
|
|
import script_audit # noqa: E402
|
|
from paths import MIGRATIONS_DIR # noqa: E402
|
|
|
|
BASE_URL = "https://services.leadconnectorhq.com"
|
|
BRAND_LOCATION_ID = "GbKkBpCmKu2QmloKFHy3"
|
|
SETTING_KEY = "allowDuplicateOpportunity"
|
|
|
|
|
|
def _headers(token, *, write=False):
|
|
h = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Version": "2021-07-28",
|
|
"Accept": "application/json",
|
|
"User-Agent": "Mozilla/5.0", # GHL rechaza sin User-Agent (403)
|
|
}
|
|
if write:
|
|
h["Content-Type"] = "application/json"
|
|
return h
|
|
|
|
|
|
def get_location(token, location_id):
|
|
r = requests.get(f"{BASE_URL}/locations/{location_id}", headers=_headers(token), timeout=15)
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"GET /locations/{location_id} -> HTTP {r.status_code}: {r.text[:200]}")
|
|
return r.json().get("location", {})
|
|
|
|
|
|
def put_setting(token, location_id, value):
|
|
body = {"settings": {SETTING_KEY: value}}
|
|
r = requests.put(
|
|
f"{BASE_URL}/locations/{location_id}",
|
|
headers=_headers(token, write=True),
|
|
json=body,
|
|
timeout=15,
|
|
)
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"PUT /locations/{location_id} -> HTTP {r.status_code}: {r.text[:300]}")
|
|
return r.json()
|
|
|
|
|
|
def snapshot_settings(location_id, name, settings):
|
|
os.makedirs(str(MIGRATIONS_DIR), exist_ok=True)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
path = os.path.join(str(MIGRATIONS_DIR), f"enable_dup_opp_{location_id}_{ts}.json")
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump({"location_id": location_id, "name": name, "settings_before": settings}, f,
|
|
ensure_ascii=False, indent=2)
|
|
return path
|
|
|
|
|
|
def run(location_id=BRAND_LOCATION_ID, target_value=True, dry_run=True, run_id=None, log=print):
|
|
token = os.getenv("GHL_AGENCY_TOKEN")
|
|
if not token:
|
|
raise RuntimeError("Falta GHL_AGENCY_TOKEN en el entorno (.env). Es el token de agencia con locations.write.")
|
|
|
|
log(f"[{datetime.now().strftime('%H:%M:%S')}] === enable_duplicate_opportunity (target={target_value}) ===")
|
|
log(f"Modo: {'APPLY' if not dry_run else 'DRY-RUN (no escribe)'} | location={location_id}")
|
|
|
|
loc = get_location(token, location_id)
|
|
name = loc.get("name", "?")
|
|
settings = loc.get("settings", {})
|
|
current = settings.get(SETTING_KEY, None)
|
|
log(f"Location: {name} | {SETTING_KEY} actual = {current}")
|
|
|
|
snap = snapshot_settings(location_id, name, settings)
|
|
log(f"Snapshot settings -> {snap}")
|
|
|
|
result = {
|
|
"location_id": location_id, "name": name,
|
|
"before": current, "target": target_value,
|
|
"dry_run": dry_run, "changed": False, "snapshot": snap,
|
|
}
|
|
|
|
if current is target_value:
|
|
log(f"No-op: {SETTING_KEY} ya es {target_value}.")
|
|
return result
|
|
|
|
if dry_run:
|
|
log(f"DRY-RUN: cambiaria {SETTING_KEY} {current} -> {target_value} (PUT no enviado).")
|
|
return result
|
|
|
|
change_id = None
|
|
if run_id:
|
|
change_id = script_audit.record_change(
|
|
run_id, location_id, "location", location_id,
|
|
SETTING_KEY, SETTING_KEY, {"value": current}, {"value": target_value},
|
|
)
|
|
try:
|
|
put_setting(token, location_id, target_value)
|
|
if change_id:
|
|
script_audit.mark_change(change_id, "applied")
|
|
except Exception as exc:
|
|
if change_id:
|
|
script_audit.mark_change(change_id, "failed", str(exc))
|
|
raise
|
|
|
|
# Verificacion
|
|
after = get_location(token, location_id).get("settings", {}).get(SETTING_KEY)
|
|
result["after"] = after
|
|
result["changed"] = (after is target_value)
|
|
log(f"Verificacion: {SETTING_KEY} = {after} -> {'OK' if result['changed'] else 'FALLO'}")
|
|
return result
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description="Activa allowDuplicateOpportunity en una location de GHL.")
|
|
p.add_argument("--location", default=BRAND_LOCATION_ID, help="location_id (default: Marca).")
|
|
p.add_argument("--apply", action="store_true", help="Aplica el cambio (sin esto, dry-run).")
|
|
p.add_argument("--disable", action="store_true", help="Pone el flag en FALSE (rollback) en vez de TRUE.")
|
|
p.add_argument("--run-id", default=None, help="run_id de script_audit (recomendado con --apply).")
|
|
p.add_argument("--json", action="store_true", help="Imprime el resultado como JSON.")
|
|
args = p.parse_args()
|
|
|
|
dry_run = not args.apply
|
|
run_id = args.run_id
|
|
if args.apply and not run_id:
|
|
run_id = uuid.uuid4().hex
|
|
script_audit.create_run(run_id, "enable_duplicate_opportunity",
|
|
arguments=f"location={args.location} disable={args.disable}",
|
|
locations=[args.location])
|
|
print(f"run_id generado: {run_id}")
|
|
|
|
try:
|
|
res = run(location_id=args.location, target_value=(not args.disable),
|
|
dry_run=dry_run, run_id=run_id)
|
|
if args.apply and run_id:
|
|
script_audit.update_run_status(run_id, "completed")
|
|
except Exception as e:
|
|
print(f"ERROR: {e}")
|
|
if args.apply and run_id:
|
|
try:
|
|
script_audit.update_run_status(run_id, "failed", str(e))
|
|
except Exception:
|
|
pass
|
|
sys.exit(2)
|
|
|
|
if args.json:
|
|
print(json.dumps(res, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|