250 lines
10 KiB
Python
250 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
find_test_contacts.py
|
|
|
|
Busca y clasifica contactos de test, pruebas, demo o E3 en la base de datos local SQLite.
|
|
También detecta si estos contactos tienen oportunidades asociadas.
|
|
|
|
Uso:
|
|
python scripts/find_test_contacts.py --all
|
|
python scripts/find_test_contacts.py --location GbKkBpCmKu2QmloKFHy3
|
|
python scripts/find_test_contacts.py --all --include-main --csv report.csv
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import sqlite3
|
|
import json
|
|
import csv
|
|
|
|
# Agregar directorio raÃz al path para importar contact_classifier
|
|
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)
|
|
|
|
# Single source of truth: patrones y logica de clasificacion viven en contact_classifier.
|
|
from contact_classifier import ( # noqa: E402
|
|
classify_contact,
|
|
normalize_text,
|
|
SPANISH_PATTERNS,
|
|
ENGLISH_PATTERNS,
|
|
E3_PATTERNS,
|
|
GENERIC_PATTERNS,
|
|
)
|
|
|
|
from paths import DB_PATH
|
|
BRAND_LOCATION_ID = "GbKkBpCmKu2QmloKFHy3"
|
|
|
|
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 main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Busca y clasifica contactos de test, pruebas, demo o E3 en SQLite."
|
|
)
|
|
parser.add_argument("--all", action="store_true", help="Buscar en todas las sucursales.")
|
|
parser.add_argument("--location", type=str, help="Filtrar por una ubicación (locationId) especÃfica.")
|
|
|
|
parser.add_argument("--include-main", action="store_true", help="Incluir la cuenta de Marca Principal en la búsqueda global.")
|
|
parser.add_argument("--csv", type=str, help="Exportar resultados a un archivo CSV (ruta).")
|
|
parser.add_argument("--limit", type=int, default=200, help="LÃmite máximo de resultados a imprimir en pantalla. Default: 200.")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not os.path.exists(DB_PATH):
|
|
safe_print(f"Error: La base de datos local no existe en {DB_PATH}.")
|
|
safe_print("Por favor, ejecuta la sincronización global primero desde el dashboard.")
|
|
sys.exit(1)
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
try:
|
|
# Construir consulta SQL
|
|
sql_contacts = """
|
|
SELECT c.*, a.nombre as branch_name, a.type as branch_type
|
|
FROM contacts c
|
|
JOIN accounts a ON c.location_id = a.location_id
|
|
"""
|
|
params = []
|
|
conditions = []
|
|
|
|
if args.location:
|
|
conditions.append("c.location_id = ?")
|
|
params.append(args.location)
|
|
else:
|
|
# Buscar en todos. Si no se incluye la principal, omitir BRAND_LOCATION_ID
|
|
if not args.include_main:
|
|
conditions.append("c.location_id != ?")
|
|
params.append(BRAND_LOCATION_ID)
|
|
|
|
if conditions:
|
|
sql_contacts += " WHERE " + " AND ".join(conditions)
|
|
|
|
sql_contacts += " ORDER BY c.date_added DESC"
|
|
|
|
# Obtener todos los contactos candidatos
|
|
contacts_rows = conn.execute(sql_contacts, params).fetchall()
|
|
|
|
# Obtener todas las oportunidades para mapear si tienen
|
|
opps_rows = conn.execute("""
|
|
SELECT o.*, p.name as pipeline_name
|
|
FROM opportunities o
|
|
LEFT JOIN pipelines p ON o.pipeline_id = p.id AND o.location_id = p.location_id
|
|
""").fetchall()
|
|
|
|
# Agrupar oportunidades por contact_id y location_id
|
|
opps_by_contact = {}
|
|
for row in opps_rows:
|
|
key = (row["contact_id"], row["location_id"])
|
|
opps_by_contact.setdefault(key, []).append(dict(row))
|
|
|
|
matched_contacts = []
|
|
stats_by_class = {}
|
|
stats_by_branch = {}
|
|
|
|
for row in contacts_rows:
|
|
tags = []
|
|
if row["tags"]:
|
|
try:
|
|
tags = json.loads(row["tags"])
|
|
except Exception:
|
|
tags = []
|
|
|
|
reasons, classes = classify_contact(row["first_name"], row["last_name"], row["email"], tags)
|
|
|
|
if reasons:
|
|
c_dict = dict(row)
|
|
c_dict["tags_parsed"] = tags
|
|
c_dict["reasons"] = reasons
|
|
c_dict["classifications"] = classes
|
|
|
|
# Obtener oportunidades
|
|
key = (row["id"], row["location_id"])
|
|
c_dict["opportunities"] = opps_by_contact.get(key, [])
|
|
|
|
matched_contacts.append(c_dict)
|
|
|
|
# EstadÃsticas
|
|
for cls in classes:
|
|
stats_by_class[cls] = stats_by_class.get(cls, 0) + 1
|
|
|
|
branch_key = f"{row['branch_name']} ({row['location_id']})"
|
|
stats_by_branch[branch_key] = stats_by_branch.get(branch_key, 0) + 1
|
|
|
|
# Imprimir reporte
|
|
W = 100
|
|
safe_print("=" * W)
|
|
safe_print("REPORTE: BÚSQUEDA DE CONTACTOS DE TEST / PRUEBAS / E3".center(W))
|
|
safe_print("=" * W)
|
|
|
|
safe_print(f" Total contactos analizados en SQLite: {len(contacts_rows)}")
|
|
safe_print(f" Total contactos de test detectados: {len(matched_contacts)}")
|
|
|
|
if matched_contacts:
|
|
# Desglose por clasificación
|
|
safe_print("\n" + "─" * W)
|
|
safe_print(" Desglose por clasificación:")
|
|
for cls, count in sorted(stats_by_class.items(), key=lambda x: -x[1]):
|
|
safe_print(f" - {cls:<25}: {count:>4} contactos")
|
|
|
|
# Desglose por sucursal
|
|
safe_print("\n" + "─" * W)
|
|
safe_print(" Top Sucursales con registros de test:")
|
|
for branch, count in sorted(stats_by_branch.items(), key=lambda x: -x[1])[:10]:
|
|
safe_print(f" - {branch:<55}: {count:>4} contactos")
|
|
if len(stats_by_branch) > 10:
|
|
safe_print(f" ... y otras {len(stats_by_branch) - 10} sucursales más.")
|
|
|
|
# Listado de contactos (hasta el lÃmite)
|
|
safe_print("\n" + "=" * W)
|
|
safe_print(f"LISTADO DE COINCIDENCIAS (Mostrando primeros {min(len(matched_contacts), args.limit)})".center(W))
|
|
safe_print("=" * W)
|
|
|
|
for i, c in enumerate(matched_contacts[:args.limit], 1):
|
|
name = f"{c['first_name'] or ''} {c['last_name'] or ''}".strip() or "Sin Nombre"
|
|
email = c["email"] or "—"
|
|
phone = c["phone"] or "—"
|
|
reasons_str = ", ".join(c["reasons"])
|
|
class_str = "/".join(c["classifications"])
|
|
|
|
safe_print(f"\n{i:03d}. {name} | Sucursal: {c['branch_name']}")
|
|
safe_print(f" ID GHL: {c['id']}")
|
|
safe_print(f" Clase/Motivo: [{class_str}] — {reasons_str}")
|
|
safe_print(f" Email: {email} | Tel: {phone}")
|
|
if c["tags_parsed"]:
|
|
safe_print(f" Etiquetas: {', '.join(c['tags_parsed'])}")
|
|
|
|
# Mostrar oportunidades vinculadas si tiene
|
|
if c["opportunities"]:
|
|
safe_print(f" Oportunidades vinculadas ({len(c['opportunities'])}):")
|
|
for o in c["opportunities"]:
|
|
val = f"${(o.get('monetary_value') or 0):,.0f}"
|
|
safe_print(f" • [{o.get('status', 'open').upper()}] {o.get('name') or 'Sin Nombre'} | Valor: {val} | Etapa: {o.get('pipeline_stage_id')}")
|
|
|
|
if len(matched_contacts) > args.limit:
|
|
safe_print(f"\n... [{len(matched_contacts) - args.limit} contactos de prueba omitidos en consola por lÃmite. Use --limit para aumentar, o exporte a CSV]")
|
|
|
|
# Exportar a CSV si se solicita
|
|
if args.csv:
|
|
csv_path = args.csv
|
|
# Si no es absoluta, resolverla contra el directorio actual o raÃz
|
|
if not os.path.isabs(csv_path):
|
|
csv_path = os.path.join(os.getcwd(), csv_path)
|
|
|
|
try:
|
|
with open(csv_path, mode='w', encoding='utf-8', newline='') as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow([
|
|
"ID Contacto", "Nombre", "Email", "Telefono", "Location ID",
|
|
"Sucursal", "Tipo Sucursal", "Clasificaciones", "Detalle Motivos",
|
|
"Etiquetas", "Cant Opps", "Detalle Opps", "Fecha Creacion"
|
|
])
|
|
|
|
for c in matched_contacts:
|
|
name = f"{c['first_name'] or ''} {c['last_name'] or ''}".strip()
|
|
opps_info = []
|
|
for o in c["opportunities"]:
|
|
opps_info.append(f"[{o.get('status','open').upper()}] {o.get('name')} (${o.get('monetary_value',0)})")
|
|
opps_str = " | ".join(opps_info)
|
|
|
|
writer.writerow([
|
|
c["id"],
|
|
name,
|
|
c["email"] or "",
|
|
c["phone"] or "",
|
|
c["location_id"],
|
|
c["branch_name"],
|
|
c["branch_type"],
|
|
",".join(c["classifications"]),
|
|
"; ".join(c["reasons"]),
|
|
",".join(c["tags_parsed"]),
|
|
len(c["opportunities"]),
|
|
opps_str,
|
|
c["date_added"] or ""
|
|
])
|
|
safe_print(f"\n[ÉXITO] Reporte exportado correctamente a CSV: {csv_path}")
|
|
except Exception as e:
|
|
safe_print(f"\n[ERROR] No se pudo exportar a CSV: {e}")
|
|
else:
|
|
safe_print("\n ¡Excelente! No se encontraron contactos con nomenclaturas de test, prueba o E3.")
|
|
|
|
safe_print("\n" + "=" * W)
|
|
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|