204 lines
8.0 KiB
Python
204 lines
8.0 KiB
Python
"""Resolver de Canal de Origen + tag canonico para un contacto.
|
|
|
|
Combina señales (source nativo de GHL, tags existentes, Canal de Origen previo,
|
|
Fuente de Prospecto) para producir UNA decisión consistente:
|
|
|
|
- `decided_tag`: tag canonico que debe quedar (facebook-ads | sucursal | formulario).
|
|
- `canal_value`: valor para custom field "Canal de Origen" (FACEBOOK | SUCURSAL | FORMULARIO).
|
|
- `tags_to_add` / `tags_to_remove`: ajustes necesarios sobre las etiquetas actuales.
|
|
- `confident` / `review`: si la decisión es segura o requiere revisión humana.
|
|
- `reason`: explicación textual para auditoría.
|
|
|
|
Reglas de prioridad (de mayor a menor):
|
|
1. `source` nativo de GHL si es interpretable.
|
|
2. Único tag target ya presente, salvo que source Y canal previo lo contradigan.
|
|
3. Canal de Origen previo si existe y es del catálogo.
|
|
4. Fuente de Prospecto (SUCURSAL fuerza sucursal; LEAD DIGITAL favorece formulario sobre facebook salvo evidencia).
|
|
"""
|
|
|
|
TARGET_TAGS = ("facebook-ads", "sucursal", "formulario")
|
|
|
|
CANAL_BY_TAG = {
|
|
"facebook-ads": "FACEBOOK",
|
|
"sucursal": "SUCURSAL",
|
|
"formulario": "FORMULARIO",
|
|
}
|
|
|
|
TAG_BY_CANAL = {v: k for k, v in CANAL_BY_TAG.items()}
|
|
|
|
FUENTE_BY_TAG = {
|
|
"facebook-ads": "LEAD DIGITAL",
|
|
"sucursal": "SUCURSAL",
|
|
"formulario": "LEAD DIGITAL",
|
|
}
|
|
|
|
DIGITAL_KEYWORDS = ("facebook", "ads", "instagram", "fb.com", "ig ", " ig", "ig:", "meta")
|
|
FORM_KEYWORDS = (
|
|
"form", "formulario", "web", "landing", "page", "survey", "encuesta",
|
|
"qr", "chat", "widget", "inbound", "tiktok", "whatsapp", "wa.me", "api",
|
|
"sms", "lead-form", "leadform",
|
|
)
|
|
SUCURSAL_KEYWORDS = ("sucursal", "manual", "tienda", "offline", "store", "walk-in", "walkin")
|
|
|
|
|
|
def classify_source(source):
|
|
s = (source or "").strip().lower()
|
|
if not s:
|
|
return None
|
|
if any(w in s for w in DIGITAL_KEYWORDS):
|
|
return "facebook-ads"
|
|
if any(w in s for w in FORM_KEYWORDS):
|
|
return "formulario"
|
|
if any(w in s for w in SUCURSAL_KEYWORDS):
|
|
return "sucursal"
|
|
return None
|
|
|
|
|
|
def canal_value_to_tag(canal_value):
|
|
return TAG_BY_CANAL.get((canal_value or "").strip().upper())
|
|
|
|
|
|
def opportunity_canal_to_tag(opp_canal_value):
|
|
"""Mapea el campo 'Canal de Origen de la Oportunidad' (Facebook|Formulario|Sucursal|...)."""
|
|
v = (opp_canal_value or "").strip().lower()
|
|
if not v:
|
|
return None
|
|
if "facebook" in v or "ig" in v or "instagram" in v:
|
|
return "facebook-ads"
|
|
if "formulario" in v or "form" in v or "web" in v or "whatsapp" in v:
|
|
return "formulario"
|
|
if "sucursal" in v or "tienda" in v or "manual" in v:
|
|
return "sucursal"
|
|
return None
|
|
|
|
|
|
def decide(tags, fuente_value, canal_value, source, opportunity_canal_value=None):
|
|
tags_lower = {str(t).strip().lower() for t in tags or []}
|
|
present_targets = tags_lower & set(TARGET_TAGS)
|
|
source_tag = classify_source(source)
|
|
prior_canal_tag = canal_value_to_tag(canal_value)
|
|
opp_tag = opportunity_canal_to_tag(opportunity_canal_value)
|
|
fuente_norm = (fuente_value or "").strip().upper()
|
|
|
|
decided_tag = None
|
|
confident = False
|
|
review = False
|
|
reason = ""
|
|
|
|
if len(present_targets) == 1:
|
|
only_tag = next(iter(present_targets))
|
|
contradicts_source = source_tag and source_tag != only_tag
|
|
contradicts_canal = prior_canal_tag and prior_canal_tag != only_tag
|
|
if contradicts_source and contradicts_canal:
|
|
decided_tag = only_tag
|
|
confident = False
|
|
review = True
|
|
reason = (
|
|
f"tag único='{only_tag}', source='{source}' sugiere '{source_tag}', "
|
|
f"canal previo='{canal_value}' -> conflicto, requiere revisión"
|
|
)
|
|
else:
|
|
decided_tag = only_tag
|
|
confident = True
|
|
reason = f"tag único='{only_tag}'"
|
|
if contradicts_source:
|
|
reason += f" (source='{source}' sugiere '{source_tag}', se respeta tag)"
|
|
elif contradicts_canal:
|
|
reason += f" (canal previo='{canal_value}' difiere, se actualiza canal)"
|
|
|
|
elif len(present_targets) >= 2:
|
|
if source_tag and source_tag in present_targets:
|
|
decided_tag = source_tag
|
|
confident = True
|
|
reason = f"mix {sorted(present_targets)} -> source='{source}' apunta a '{source_tag}'"
|
|
elif source_tag:
|
|
decided_tag = source_tag
|
|
confident = True
|
|
reason = (
|
|
f"mix {sorted(present_targets)} pero source='{source}' sugiere '{source_tag}' "
|
|
"(fuera del mix); se respeta source"
|
|
)
|
|
elif prior_canal_tag and prior_canal_tag in present_targets:
|
|
decided_tag = prior_canal_tag
|
|
confident = True
|
|
reason = f"mix {sorted(present_targets)} -> canal previo '{canal_value}' apunta a '{prior_canal_tag}'"
|
|
elif prior_canal_tag:
|
|
decided_tag = prior_canal_tag
|
|
confident = True
|
|
reason = (
|
|
f"mix {sorted(present_targets)} -> canal previo '{canal_value}' "
|
|
"(no incluido en mix actual)"
|
|
)
|
|
elif opp_tag and opp_tag in present_targets:
|
|
decided_tag = opp_tag
|
|
confident = True
|
|
reason = (
|
|
f"mix {sorted(present_targets)} -> oportunidad canal='{opportunity_canal_value}' "
|
|
f"-> '{opp_tag}'"
|
|
)
|
|
elif opp_tag:
|
|
decided_tag = opp_tag
|
|
confident = True
|
|
reason = (
|
|
f"mix {sorted(present_targets)} -> oportunidad canal='{opportunity_canal_value}' "
|
|
f"(ajeno al mix) -> '{opp_tag}'"
|
|
)
|
|
elif fuente_norm == "SUCURSAL" and "sucursal" in present_targets:
|
|
decided_tag = "sucursal"
|
|
confident = True
|
|
reason = f"mix {sorted(present_targets)} -> fuente=SUCURSAL"
|
|
elif fuente_norm == "LEAD DIGITAL":
|
|
if "formulario" in present_targets:
|
|
decided_tag = "formulario"
|
|
elif "facebook-ads" in present_targets:
|
|
decided_tag = "facebook-ads"
|
|
if decided_tag:
|
|
confident = True
|
|
reason = f"mix {sorted(present_targets)} -> fuente=LEAD DIGITAL -> '{decided_tag}'"
|
|
if not decided_tag:
|
|
review = True
|
|
reason = f"mix {sorted(present_targets)} sin source/canal/fuente que resuelva"
|
|
|
|
else:
|
|
if source_tag:
|
|
decided_tag = source_tag
|
|
confident = True
|
|
reason = f"sin tag target, source='{source}' -> '{source_tag}'"
|
|
elif prior_canal_tag:
|
|
decided_tag = prior_canal_tag
|
|
confident = True
|
|
reason = f"sin tag target, canal previo '{canal_value}' -> '{prior_canal_tag}'"
|
|
elif opp_tag:
|
|
decided_tag = opp_tag
|
|
confident = True
|
|
reason = f"sin tag target, oportunidad canal='{opportunity_canal_value}' -> '{opp_tag}'"
|
|
elif fuente_norm == "SUCURSAL":
|
|
decided_tag = "sucursal"
|
|
confident = True
|
|
reason = "sin tag target, fuente=SUCURSAL"
|
|
elif fuente_norm == "LEAD DIGITAL":
|
|
decided_tag = "formulario"
|
|
confident = True
|
|
reason = "sin tag target, fuente=LEAD DIGITAL -> default 'formulario'"
|
|
else:
|
|
review = True
|
|
reason = "sin tag target, sin source claro, sin canal previo, sin fuente"
|
|
|
|
canal_value_out = CANAL_BY_TAG[decided_tag] if decided_tag else None
|
|
tags_to_add = [decided_tag] if decided_tag and decided_tag not in tags_lower else []
|
|
tags_to_remove = sorted(t for t in present_targets if decided_tag and t != decided_tag)
|
|
|
|
return {
|
|
"decided_tag": decided_tag,
|
|
"canal_value": canal_value_out,
|
|
"fuente_value": FUENTE_BY_TAG[decided_tag] if decided_tag else None,
|
|
"tags_to_add": tags_to_add,
|
|
"tags_to_remove": tags_to_remove,
|
|
"confident": confident,
|
|
"review": review,
|
|
"reason": reason,
|
|
"present_targets": sorted(present_targets),
|
|
"source_tag": source_tag,
|
|
"prior_canal_tag": prior_canal_tag,
|
|
}
|