Files
dashboard-leads/backend/services.py

2734 lines
138 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# backend/services.py
"""Servicios: traen datos crudos (cacheados) y arman la respuesta del módulo Leads."""
from data_manager_v2 import DataManager
import leads_logic as L
from cache_manager import cache_get_or_set, cache_invalidate
_DM = None
def get_dm() -> DataManager:
global _DM
if _DM is None:
_DM = DataManager()
return _DM
# ── Prefijos de cache que dependen del mapa de campañas (leyendas). Al cambiar
# una leyenda se invalidan estos para que TODOS los apartados se reconstruyan
# en su proxima peticion (perezoso: solo lo que se pida, ~2-5s), sin esperar
# los 15 min ni recargar todo. NO se toca la logica, solo se refresca el cache.
_PREF_CAMPANIAS = (
"campanias_map", "campanias_sc", "leads_raw", "leads_dash", "leads_filtros",
"leads_alertas", "vendedores_dash", "roas_dash", "otros_general",
"matriz_always", "matriz_webform", "matriz_asignados", "matriz_plantillas",
"tiempo_resp", "importe_pauta_periodo", "resultados_pauta_periodo",
"resultados_pauta", "leads_x_codigo", "tel_es_copito", "programa_dash",
)
def _invalidar_por_campanias():
"""Invalida el cache de campañas (Supabase) y de todos los apartados que
dependen de las leyendas, para que el cambio se vea de inmediato."""
from data_manager_v2 import invalidar_campanias_cache
invalidar_campanias_cache()
for p in _PREF_CAMPANIAS:
cache_invalidate(p)
# ── Datos crudos cacheados (consultas pesadas, una sola vez) ────
def _leads_crudos():
return cache_get_or_set("leads_raw", ("GLOBAL",),
lambda: L.procesar_leads(get_dm().traer_leads_chatwoot()))
def _cursos_crudos():
return cache_get_or_set("cursos_raw", ("GLOBAL",), lambda: get_dm().traer_cursos())
def _matriculas_crudas():
return cache_get_or_set("mat_raw", ("GLOBAL",), lambda: get_dm().traer_matriculas())
def _pauta_cruda():
return cache_get_or_set("pauta_raw", ("GLOBAL",), lambda: get_dm().traer_pauta_cursos())
def _programa_pautas():
"""{num_indice: [pautas]} desde la tabla programa_pautas (muchos-a-muchos)."""
return cache_get_or_set("programa_pautas", ("GLOBAL",),
lambda: get_dm().traer_programa_pautas())
def _pautas_de_indice():
"""{num_indice: [pautas]} combinando:
- la pauta de basebi_programacion (1 por num_indice), y
- las pautas extra de programa_pautas (varias).
Sin duplicados. Un num_indice puede terminar con varias pautas que se SUMAN."""
def _load():
pm = _pauta_cruda() # {num_indice: {pauta,...}}
extra = _programa_pautas() # {num_indice: [pautas]}
out = {}
for ni, info in pm.items():
p = info.get("pauta")
if p:
out.setdefault(str(ni).strip(), []).append(str(p).strip())
for ni, pautas in extra.items():
lst = out.setdefault(str(ni).strip(), [])
for p in pautas:
if str(p).strip() not in lst:
lst.append(str(p).strip())
return out
return cache_get_or_set("pautas_de_indice", ("GLOBAL",), _load)
def _campanias_map():
return cache_get_or_set("campanias_map", ("GLOBAL",),
lambda: get_dm().traer_campanias_map())
def _clasif_meta():
"""JSON de clasificacion (sede/programa) de GitHub, MISMO que Ventas."""
return cache_get_or_set("clasif_meta", ("GLOBAL",),
lambda: get_dm().traer_clasificacion_meta())
def clasificar_sede_meta(dsc_programa):
"""Sede segun el JSON (clasificacion_sedes). Default LIMA. Igual que Ventas."""
up = str(dsc_programa or "").upper()
data = _clasif_meta().get("clasificacion_sedes", {})
for sede, cfg in data.items():
if sede == "DEFAULT":
continue
for pat in cfg.get("patrones", []):
if str(pat).upper() in up:
return sede
return data.get("DEFAULT", "LIMA")
def clasificar_prog_meta(dsc_programa):
"""Programa (TEAC/TERC/SEMINARIOS/OTROS) segun clasificacion_filtro_programa
del JSON. Respeta el ORDEN del JSON (primero que coincide gana). Igual que Ventas."""
up = str(dsc_programa or "").upper()
data = _clasif_meta().get("clasificacion_filtro_programa", {})
for cat, cfg in data.items():
if cat == "DEFAULT":
continue
for pat in cfg.get("patrones", []):
if str(pat).upper() in up:
return cat
return data.get("DEFAULT", "OTROS")
def meta_de_curso(dsc_programa):
"""Meta de inscritos segun sede+programa (JSON), como acordado:
LIMA TEAC/TERC=22, provincia TEAC/TERC=18, SEMINARIOS/OTROS=15."""
sede = clasificar_sede_meta(dsc_programa)
prog = clasificar_prog_meta(dsc_programa)
if prog in ("TEAC", "TERC"):
return 22 if sede == "LIMA" else 18
return 15
# Conecta los clasificadores del JSON a leads_logic (metas + tipo de matricula).
L.set_clasificadores(clasificar_sede_meta, clasificar_prog_meta)
def _conjunto_pauta():
return cache_get_or_set("conjunto_pauta", ("GLOBAL",),
lambda: get_dm().traer_conjunto_pauta())
def _meta_filas():
"""Filas crudas del CSV de Meta: [{conjunto, importe, inicio}, ...]."""
return cache_get_or_set("meta_filas", ("GLOBAL",),
lambda: get_dm().traer_meta_importe())
def _importe_por_pauta():
"""{pauta: importe_gastado_total} = suma del importe de TODAS las filas de los
conjuntos ligados a esa pauta (conjunto_pauta × Meta CSV). Sin filtro de fecha."""
def _load():
conj_pauta = _conjunto_pauta() # {pauta: [conjunto,...]}
imp_norm = {}
for f in _meta_filas():
k = " ".join(str(f.get("conjunto")).split()).upper()
imp_norm[k] = imp_norm.get(k, 0.0) + float(f.get("importe") or 0.0)
out = {}
for pauta, conjuntos in conj_pauta.items():
total = sum(imp_norm.get(" ".join(str(co).split()).upper(), 0.0) for co in conjuntos)
out[str(pauta).strip()] = round(total, 2)
return out
return cache_get_or_set("importe_pauta", ("GLOBAL",), _load)
# Canales que cuentan como Copito (ademas de origen_base=POSTGRE).
_CANALES_COPITO = {"PAUTA_WSP", "PAUTA_WSP_FACE", "COPITO"}
def _tel_es_copito():
"""{telefono: True/False} = si el telefono es COPITO segun su FILA MAS ANTIGUA
en la cartera. Copito = (origen_base=POSTGRE) OR (canal en _CANALES_COPITO).
Se calcula una vez sobre toda la cartera (cacheado)."""
def _load():
mejor = {} # tel -> (fecha_mas_antigua, es_copito)
for r in _cartera_rows_cache():
tel = str(r.get("telefono") or "").strip()
if not tel:
continue
f = L._to_date(r.get("fecha_creada"))
if f is None:
continue
if tel not in mejor or f < mejor[tel][0]:
canal = " ".join(str(r.get("canal") or "").upper().split())
ob = " ".join(str(r.get("origen_base") or "").upper().split())
es_cop = (ob == "POSTGRE") or (canal in _CANALES_COPITO)
mejor[tel] = (f, es_cop)
return {t: v[1] for t, v in mejor.items()}
return cache_get_or_set("tel_es_copito", ("GLOBAL",), _load)
def _leads_por_codigo_periodo(ano, mes, dia):
"""{codigo(pauta): {"rec": set(tels recibidos del periodo),
"proc": set(tels procesados del periodo)}}.
Recibido = lead unico con ese codigo cuya fecha cae en el periodo.
Procesado = de esos, con asesor. Mismo criterio que 'Detalle por Curso'."""
def _load():
from collections import defaultdict
out = defaultdict(lambda: {"rec": set(), "proc": set(), "cont": set()})
for l in _leads_crudos():
if not l.get("es_unico"):
continue
if not L._en_periodo(l.get("fecha"), ano, mes, dia):
continue
cod = str(l.get("codigo", "-")).strip()
tel = l.get("telefono")
if not tel:
continue
out[cod]["rec"].add(tel)
if l.get("asesor"):
out[cod]["proc"].add(tel)
ult = l.get("ultima") or "NO TRABAJADO"
if ult != "NO TRABAJADO":
out[cod].setdefault("cont", set()).add(tel)
return dict(out)
return cache_get_or_set("leads_x_codigo", (ano, mes, dia), _load)
def _resultados_por_pauta():
"""{pauta: resultados_totales} = suma de Resultados de TODAS las filas de los
conjuntos ligados a esa pauta. Sin filtro de fecha (analogo a _importe_por_pauta)."""
def _load():
conj_pauta = _conjunto_pauta()
res_norm = {}
for f in _meta_filas():
k = " ".join(str(f.get("conjunto")).split()).upper()
res_norm[k] = res_norm.get(k, 0.0) + float(f.get("resultados") or 0.0)
out = {}
for pauta, conjuntos in conj_pauta.items():
total = sum(res_norm.get(" ".join(str(co).split()).upper(), 0.0) for co in conjuntos)
out[str(pauta).strip()] = int(total)
return out
return cache_get_or_set("resultados_pauta", ("GLOBAL",), _load)
def _importe_por_pauta_periodo(ano, mes, dia):
"""{pauta: importe} sumando solo las filas del CSV cuyo 'Inicio del informe'
cae dentro del periodo filtrado (año/mes/día). Respeta el filtro de arriba."""
def _load():
conj_pauta = _conjunto_pauta()
imp_norm = {}
sin_filtro = (ano in ("TODOS", None) and mes in ("TODOS", None) and dia in ("TODOS", None))
for f in _meta_filas():
fx = L._to_date(f.get("inicio"))
if sin_filtro:
pass # sin filtro: cuenta todo (incluye filas sin fecha) = total
elif not L._en_periodo(fx, ano, mes, dia):
continue
k = " ".join(str(f.get("conjunto")).split()).upper()
imp_norm[k] = imp_norm.get(k, 0.0) + float(f.get("importe") or 0.0)
out = {}
for pauta, conjuntos in conj_pauta.items():
total = sum(imp_norm.get(" ".join(str(co).split()).upper(), 0.0) for co in conjuntos)
out[str(pauta).strip()] = round(total, 2)
return out
return cache_get_or_set("importe_pauta_periodo", (ano, mes, dia), _load)
def _resultados_por_pauta_periodo(ano, mes, dia):
"""{pauta: resultados} sumando solo las filas del CSV cuyo 'Inicio del informe'
cae en el periodo filtrado. Analogo a _importe_por_pauta_periodo pero con Resultados."""
def _load():
conj_pauta = _conjunto_pauta()
res_norm = {}
sin_filtro = (ano in ("TODOS", None) and mes in ("TODOS", None) and dia in ("TODOS", None))
for f in _meta_filas():
if not sin_filtro and not L._en_periodo(L._to_date(f.get("inicio")), ano, mes, dia):
continue
k = " ".join(str(f.get("conjunto")).split()).upper()
res_norm[k] = res_norm.get(k, 0.0) + float(f.get("resultados") or 0.0)
out = {}
for pauta, conjuntos in conj_pauta.items():
total = sum(res_norm.get(" ".join(str(co).split()).upper(), 0.0) for co in conjuntos)
out[str(pauta).strip()] = int(total)
return out
return cache_get_or_set("resultados_pauta_periodo", (ano, mes, dia), _load)
def _campanias_sede_cargo():
return cache_get_or_set("campanias_sc", ("GLOBAL",),
lambda: get_dm().traer_campanias_sede_cargo())
def matriz_always(ano="TODOS", mes="TODOS", dia="TODOS", sede="TODOS", programa="TODOS"):
"""Matriz de gasto Meta Ads de pautas ALWAYS (con gasto pero SIN num_indice).
Agrupa Sede -> Programa (TEAC/TERC/SEMINARIOS) con Importe, Resultados,
Leads Nuevos (fecha_creada en el mes) y Leads Nuevos Asignados (con asesor).
Sede/programa salen de la tabla campanias. Pautas sin campanias NO se cuentan."""
def _load():
conj_pauta = _conjunto_pauta() # {pauta: [conjuntos]}
camp = _campanias_sede_cargo() # {pauta: {sede,cargo}}
pauta_map = _pauta_cruda() # {num_indice: {pauta,...}}
con_indice = {str(v.get("pauta")).strip() for v in pauta_map.values() if v.get("pauta")}
# Leads nuevos (unicos) por pauta = codigo, cuya fecha_creada cae en el mes filtrado
leads = _leads_crudos()
leads_nuevos = {} # codigo -> nº leads unicos del mes
leads_nuevos_asig = {} # codigo -> nº leads unicos del mes con asesor
tel_lead_mes = {} # telefono -> (codigo, fecha_creada) del lead unico del mes
for l in leads:
if not l.get("es_unico"):
continue
if not L._en_periodo(l.get("fecha"), ano, mes, dia):
continue
cod = str(l.get("codigo") or "").strip()
if not cod:
continue
leads_nuevos[cod] = leads_nuevos.get(cod, 0) + 1
if str(l.get("asesor") or "").strip():
leads_nuevos_asig[cod] = leads_nuevos_asig.get(cod, 0) + 1
tel = str(l.get("telefono") or "").strip()
if tel:
tel_lead_mes[tel] = (cod, l.get("fecha"))
# MATRICULAS por pauta: matriculas cuyo telefono = un lead nuevo del mes, y con
# fch_matricula >= fecha_creada del lead. Cuenta CADA matricula (una vez c/u),
# atribuida a la pauta del lead. Cruce por telefono (tel_1, o tel_2 si vacio).
matriculas_pauta = {} # codigo -> nº matriculas
venta_pauta = {} # codigo -> suma INV_NETA en soles (DOL ×3.34)
_vistas = set() # (num_matricula) para no contar 2 veces
for m in _matriculas_crudas():
t1 = str(m.get("dsc_telefono_1") or "").strip()
t2 = str(m.get("dsc_telefono_2") or "").strip()
ph = t1 if t1 else t2
if ph not in tel_lead_mes:
continue
cod, f_lead = tel_lead_mes[ph]
fm = L._to_date(m.get("fch_matricula"))
if not fm or not f_lead or fm < f_lead:
continue # matricula anterior al ingreso del lead -> no cuenta
nm = str(m.get("num_matricula"))
if nm in _vistas:
continue
_vistas.add(nm)
matriculas_pauta[cod] = matriculas_pauta.get(cod, 0) + 1
# INV_NETA en soles: si moneda DOL, ×3.34 (igual que tabla por Sede)
try:
inv = float(m.get("INV_NETA", 0) or 0)
except (TypeError, ValueError):
inv = 0.0
if str(m.get("cod_moneda", "")).strip().upper() == "DOL":
inv *= 3.34
venta_pauta[cod] = venta_pauta.get(cod, 0.0) + inv
# Importe y Resultados por conjunto, filtrando por el MES de arriba
imp = {}; res = {}
sin_filtro = (ano in ("TODOS", None) and mes in ("TODOS", None) and dia in ("TODOS", None))
for f in _meta_filas():
if not sin_filtro and not L._en_periodo(L._to_date(f.get("inicio")), ano, mes, dia):
continue
k = " ".join(str(f.get("conjunto")).split()).upper()
imp[k] = imp.get(k, 0.0) + float(f.get("importe") or 0.0)
res[k] = res.get(k, 0.0) + float(f.get("resultados") or 0.0)
# Conjuntos BRANDING/COPEAIRE: su importe SÍ suma, pero sus resultados NO.
conj_brand = _conjuntos_branding()
# Acumular por Sede -> Programa -> Pauta (solo pautas ALWAYS que esten en campanias)
from collections import defaultdict
grupos = defaultdict(lambda: defaultdict(lambda: {"importe": 0.0, "resultados": 0.0,
"nuevos": 0, "nuevos_asig": 0,
"matriculas": 0, "venta": 0.0, "pautas": []}))
for pauta, conjuntos in conj_pauta.items():
p = str(pauta).strip()
if p in con_indice: # tiene num_indice -> NO es always
continue
info = camp.get(p)
if not info: # no esta en campanias -> no se cuenta
continue
se = str(info.get("sede") or "").upper()
gr = L.grupo_programa_lead(info.get("cargo")) # TEAC/TERC/SEMINARIOS
ti = sum(imp.get(" ".join(str(c).split()).upper(), 0.0) for c in conjuntos)
# Resultados: NO contar los conjuntos BRANDING/COPEAIRE.
tr = sum(res.get(" ".join(str(c).split()).upper(), 0.0) for c in conjuntos
if " ".join(str(c).split()).upper() not in conj_brand)
if ti == 0 and tr == 0:
continue # sin gasto en el mes -> no aparece
# filtros de arriba
if sede not in ("TODOS", None) and se != str(sede).upper():
continue
if programa not in ("TODOS", None) and gr != str(programa).upper():
continue
nv = leads_nuevos.get(p, 0)
na = leads_nuevos_asig.get(p, 0)
mt = matriculas_pauta.get(p, 0)
vv = venta_pauta.get(p, 0.0)
g = grupos[se][gr]
g["importe"] += ti
g["resultados"] += tr
g["nuevos"] += nv
g["nuevos_asig"] += na
g["matriculas"] += mt
g["venta"] += vv
g["pautas"].append({"pauta": p, "importe": round(ti, 2), "resultados": int(tr),
"nuevos": nv, "nuevos_asig": na, "matriculas": mt,
"venta": round(vv, 0)})
# Armar filas: Sede -> subfilas por programa -> pautas (solo con gasto)
filas = []
for se in sorted(grupos.keys()):
subs = []
s_imp = s_res = 0.0; s_nv = s_na = s_mt = 0; s_vv = 0.0
for gr in ("TEAC", "TERC", "SEMINARIOS"):
if gr in grupos[se]:
d = grupos[se][gr]
pautas = sorted(d["pautas"], key=lambda x: -x["importe"])
subs.append({"programa": gr, "importe": round(d["importe"], 2),
"resultados": int(d["resultados"]),
"nuevos": d["nuevos"], "nuevos_asig": d["nuevos_asig"],
"matriculas": d["matriculas"], "venta": round(d["venta"], 0),
"pautas": pautas})
s_imp += d["importe"]; s_res += d["resultados"]
s_nv += d["nuevos"]; s_na += d["nuevos_asig"]; s_mt += d["matriculas"]; s_vv += d["venta"]
filas.append({"sede": se, "importe": round(s_imp, 2), "resultados": int(s_res),
"nuevos": s_nv, "nuevos_asig": s_na, "matriculas": s_mt,
"venta": round(s_vv, 0), "subfilas": subs})
tot_i = round(sum(f["importe"] for f in filas), 2)
tot_r = int(sum(f["resultados"] for f in filas))
tot_nv = sum(f["nuevos"] for f in filas)
tot_na = sum(f["nuevos_asig"] for f in filas)
tot_mt = sum(f["matriculas"] for f in filas)
tot_vv = round(sum(f["venta"] for f in filas), 0)
return {"filas": filas, "total": {"importe": tot_i, "resultados": tot_r,
"nuevos": tot_nv, "nuevos_asig": tot_na,
"matriculas": tot_mt, "venta": tot_vv}}
return cache_get_or_set("matriz_always", (ano, mes, dia, sede, programa), _load)
def matriz_web_formulario(ano="TODOS", mes="TODOS", dia="TODOS", sede="TODOS", programa="TODOS"):
"""Matriz de leads con canal WEB_FORMULARIO (cartera_junta), agrupados Sede -> Programa.
Valores: Leads Recibidos (todos), Leads Procesados (con asesor), Matriculas y
Valor Venta (cruce por telefono, fch_matricula >= fecha_creada del lead)."""
def _load():
rows = _cartera_rows_cache()
# 1) leads WEB_FORMULARIO cuya fila es el ORIGEN del telefono en TODA la cartera
# (es_origen=SI) y con fecha_creada en el periodo. Asi NO cuenta si el telefono
# ya vino antes por otro canal (igual criterio que las demas tablas).
vistos = {} # telefono -> {fecha, sede, prog, asesor}
for r in rows:
if str(r.get("canal") or "").strip().upper() != "WEB_FORMULARIO":
continue
if str(r.get("es_origen") or "").strip().upper() != "SI":
continue # solo el origen real del telefono
tel = str(r.get("telefono") or "").strip()
if not tel:
continue
f = L._to_date(r.get("fecha_creada"))
if not L._en_periodo(f, ano, mes, dia):
continue
# sede y programa YA vienen normalizados en cartera_junta. Reducir a 4 grupos.
se = " ".join(str(r.get("sede") or "").upper().split()) or "SIN SEDE"
pr = " ".join(str(r.get("programa") or "").upper().split())
if pr == "TEAC":
gr = "TEAC"
elif pr == "TERC":
gr = "TERC"
elif pr in ("SEMINARIOS", "VRF", "CO2", "DIPLOMADO", "VENTILACIÓN", "VENTILACION"):
gr = "SEMINARIOS"
else:
gr = "OTROS"
vistos[tel] = {"fecha": f, "sede": se, "prog": gr,
"asesor": str(r.get("asesor") or "").strip()}
# 2) matriculas por telefono (num_matricula, fecha, inv en soles)
from collections import defaultdict
mat_por_tel = defaultdict(list)
for m in _matriculas_crudas():
t1 = str(m.get("dsc_telefono_1") or "").strip()
t2 = str(m.get("dsc_telefono_2") or "").strip()
ph = t1 if t1 else t2
if not ph:
continue
fm = L._to_date(m.get("fch_matricula"))
try:
inv = float(m.get("INV_NETA", 0) or 0)
except (TypeError, ValueError):
inv = 0.0
if str(m.get("cod_moneda", "")).strip().upper() == "DOL":
inv *= 3.34
mat_por_tel[ph].append((str(m.get("num_matricula")), fm, inv))
# 3) agrupar Sede -> Programa
grupos = defaultdict(lambda: defaultdict(lambda: {"recibidos": 0, "procesados": 0,
"matriculas": 0, "venta": 0.0}))
for tel, info in vistos.items():
se = info["sede"]; gr = info["prog"]; f_lead = info["fecha"]
if sede not in ("TODOS", None) and se != str(sede).upper():
continue
if programa not in ("TODOS", None) and gr != str(programa).upper():
continue
g = grupos[se][gr]
g["recibidos"] += 1
if info["asesor"]:
g["procesados"] += 1
for (nm, fm, inv) in mat_por_tel.get(tel, []):
if fm and f_lead and fm >= f_lead:
g["matriculas"] += 1
g["venta"] += inv
# 4) filas
filas = []
for se in sorted(grupos.keys()):
subs = []
s_rec = s_proc = s_mt = 0; s_vv = 0.0
for gr in ("TEAC", "TERC", "SEMINARIOS", "OTROS"):
if gr in grupos[se]:
d = grupos[se][gr]
subs.append({"programa": gr, "recibidos": d["recibidos"],
"procesados": d["procesados"], "matriculas": d["matriculas"],
"venta": round(d["venta"], 0)})
s_rec += d["recibidos"]; s_proc += d["procesados"]
s_mt += d["matriculas"]; s_vv += d["venta"]
filas.append({"sede": se, "recibidos": s_rec, "procesados": s_proc,
"matriculas": s_mt, "venta": round(s_vv, 0), "subfilas": subs})
tot = {"recibidos": sum(f["recibidos"] for f in filas),
"procesados": sum(f["procesados"] for f in filas),
"matriculas": sum(f["matriculas"] for f in filas),
"venta": round(sum(f["venta"] for f in filas), 0)}
return {"filas": filas, "total": tot}
return cache_get_or_set("matriz_webform", (ano, mes, dia, sede, programa), _load)
def _asignacion_respuesta_crudos():
return cache_get_or_set("asig_resp", ("GLOBAL",),
lambda: get_dm().traer_asignacion_respuesta())
def _leads_asignados_crudos():
# UNIFICADO: usa la MISMA consulta/cache que el grafico (asignacion+respuesta+etiquetas),
# asi la tarjeta, la matriz de asignados y el grafico nunca se desfasan.
return _asignacion_respuesta_crudos()
def _actividad_asesores_crudos():
return cache_get_or_set("act_asesores", ("GLOBAL",),
lambda: get_dm().traer_actividad_asesores())
def matriz_asignados(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Matriz pivot: filas = asesor (user_name) con sus telefonos; columnas = dias del
mes; valor = cantidad de telefonos asignados a ese asesor ese dia.
Fecha = created_at del mensaje de asignacion (ya con -5h)."""
def _load():
import calendar
from collections import defaultdict
rows = _leads_asignados_crudos()
# dias del mes filtrado (columnas)
if mes not in ("TODOS", None) and ano not in ("TODOS", None):
ndias = calendar.monthrange(int(ano), int(mes))[1]
dias_cols = list(range(1, ndias + 1))
else:
dias_cols = list(range(1, 32))
# asesores excluidos de esta matriz
EXCLUIR = {"SIN ASESOR", "COPITO RIVERA"}
# asesor -> dia -> [telefonos]
por = defaultdict(lambda: defaultdict(list))
for r in rows:
f = L._to_date(r.get("created_at"))
if not L._en_periodo(f, ano, mes, dia):
continue
ase = str(r.get("user_name") or "Sin Asesor").strip() or "Sin Asesor"
if ase.upper() in EXCLUIR:
continue
tel = str(r.get("telefono") or "").strip()
por[ase][f.day].append(tel)
filas = []
tot_por_dia = defaultdict(int)
for ase in sorted(por.keys()):
pordia = {d: len(por[ase].get(d, [])) for d in dias_cols}
total = sum(pordia.values())
# telefonos por dia (para expandir con +)
tels_dia = {d: sorted(por[ase].get(d, [])) for d in dias_cols if por[ase].get(d)}
for d, n in pordia.items():
tot_por_dia[d] += n
filas.append({"asesor": ase, "por_dia": pordia, "total": total, "tels_dia": tels_dia})
filas.sort(key=lambda x: -x["total"])
total_gral = sum(f["total"] for f in filas)
return {"dias": dias_cols, "filas": filas,
"total_por_dia": {d: tot_por_dia.get(d, 0) for d in dias_cols},
"total": total_gral}
return cache_get_or_set("matriz_asignados", (ano, mes, dia), _load)
def _plantillas_crudas():
return cache_get_or_set("plantillas_raw", ("GLOBAL",),
lambda: get_dm().traer_plantillas())
def matriz_plantillas(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Matriz de plantillas COBRADAS por plantilla. Columnas:
ENVIADAS = Mas_de_24h=SI, created_at_peru en el periodo filtrado
RESPONDIDAS= de esas, con respuesta del contacto (siguiente_mensaje_1)
ASIGNADAS = de esas, con asesor Y respondida por el contacto
MATRICULAS = matriculas de esos telefonos (fch_matricula >= envio mas antiguo). Sin filtro de mes.
VENTA = INV_NETA (DOL x3.34) de esas matriculas."""
from datetime import timedelta, datetime
from collections import defaultdict
def _load():
rows = _plantillas_crudas()
UMBRAL = timedelta(hours=24)
def _lleno(s):
return s is not None and str(s).strip() != ""
enviadas = defaultdict(int)
respondidas = defaultdict(int)
asignadas = defaultdict(int)
prim_por_tel = {} # telefono -> (fecha_envio, plantilla) la cobrada mas antigua DEL PERIODO
for r in rows:
plantilla = r.get("plantilla")
envio = r.get("created_at_peru")
f = L._to_date(envio)
# ENVIADAS/RESP/ASIG respetan el filtro de mes (por created_at_peru)
if not L._en_periodo(f, ano, mes, dia):
continue
anterior = r.get("hora_anterior_contacto")
sig = r.get("siguiente_mensaje_1")
es_si = False
if plantilla == "descuento_egresados_seminarios":
es_si = True
elif anterior is not None and (envio - anterior) > UMBRAL:
es_si = True
if not es_si:
continue
enviadas[plantilla] += 1
if _lleno(sig):
respondidas[plantilla] += 1
if _lleno(r.get("user_name")) and _lleno(sig):
asignadas[plantilla] += 1
tel = str(r.get("telefono") or "").strip()
if tel and envio is not None:
if tel not in prim_por_tel or envio < prim_por_tel[tel][0]:
prim_por_tel[tel] = (envio, plantilla)
# MATRICULAS + VENTA de los telefonos filtrados (sin filtro de mes en la matricula)
mat_por_tel = defaultdict(list)
for m in _matriculas_crudas():
t1 = str(m.get("dsc_telefono_1") or "").strip()
t2 = str(m.get("dsc_telefono_2") or "").strip()
ph = t1 if t1 else t2
if not ph:
continue
fm = m.get("fch_matricula")
if isinstance(fm, datetime): fm = fm.date()
try:
inv = float(m.get("INV_NETA", 0) or 0)
except (TypeError, ValueError):
inv = 0.0
if str(m.get("cod_moneda", "")).strip().upper() == "DOL":
inv *= 3.34
mat_por_tel[ph].append((str(m.get("num_matricula")), fm, inv))
matriculas = defaultdict(int)
venta = defaultdict(float)
_vistas = set()
for tel, (fenvio, plantilla) in prim_por_tel.items():
fenvio_d = fenvio.date() if hasattr(fenvio, "date") else fenvio
for (nm, fm, inv) in mat_por_tel.get(tel, []):
if not fm or nm in _vistas:
continue
if fm >= fenvio_d:
_vistas.add(nm)
matriculas[plantilla] += 1
venta[plantilla] += inv
filas = []
for p in sorted(enviadas.keys(), key=lambda x: -enviadas[x]):
filas.append({"plantilla": p, "enviadas": enviadas[p],
"respondidas": respondidas.get(p, 0),
"asignadas": asignadas.get(p, 0),
"matriculas": matriculas.get(p, 0),
"venta": round(venta.get(p, 0.0), 0)})
tot = {"enviadas": sum(enviadas.values()), "respondidas": sum(respondidas.values()),
"asignadas": sum(asignadas.values()), "matriculas": sum(matriculas.values()),
"venta": round(sum(venta.values()), 0)}
return {"filas": filas, "total": tot}
return cache_get_or_set("matriz_plantillas", (ano, mes, dia), _load)
def _norm_ni(x):
"""Normaliza num_indice a texto (quita el '.0' si viniera como float)."""
s = str(x or "").strip()
return s[:-2] if s.endswith(".0") else s
def _mat_por_indice():
"""{num_indice: [(telefono_resuelto, fecha_matricula), ...]} desde las matrículas.
Teléfono resuelto = dsc_telefono_1 si existe; si no, dsc_telefono_2."""
def _load():
from collections import defaultdict
out = defaultdict(list)
for m in _matriculas_crudas():
ni = _norm_ni(m.get("num_indice"))
t1 = str(m.get("dsc_telefono_1") or "").strip()
t2 = str(m.get("dsc_telefono_2") or "").strip()
fm = L._to_date(m.get("fch_matricula"))
out[ni].append((t1 if t1 else t2, fm))
return out
return cache_get_or_set("mat_por_indice", ("GLOBAL",), _load)
def _cartera_rows_cache():
"""Filas crudas de cartera_junta cacheadas (se bajan de Supabase UNA sola vez).
Evita re-descargar las 86k filas en cada llamada/cambio de mes."""
return cache_get_or_set("cartera_rows", ("GLOBAL",),
lambda: get_dm().traer_cartera_rows())
def _cartera_idx():
"""Devuelve {'all':..., 'asig':...}, cada uno {(sede,programa):{telefono:fecha_min}}.
'all' = todos los teléfonos; 'asig' = solo los que tienen asesor (no vacío).
Se baja la cartera UNA vez y queda cacheado."""
def _load():
from collections import defaultdict
CANAL_PAUTA = {"COPITO", "PAUTA_WSP", "PAUTA_WSP_FACE"}
CANAL_WEB = {"WEB_FORMULARIO", "WEB_WHATSAPP", "WHATSAPP WEB"}
def _clase(canal):
if canal in CANAL_PAUTA: return "PAUTA"
if canal in CANAL_WEB: return "WEB"
return "OTROS"
# PASO 1: DEDUP por (sede, prog, telefono) -> quedarse con la fila mas antigua.
# Se guarda la fecha, si tiene asesor, y el CANAL de esa fila ganadora.
# (Asi el canal de PAUTA/OTROS se decide DESPUES de deduplicar, como pide el usuario.)
ganador = {} # (sede, prog, tel) -> {"f":fecha, "asesor":bool, "canal":canal}
tels = set()
tels_canal = {} # telefono -> canal de su fila ganadora GLOBAL (para matriculas)
gan_global = {} # telefono -> fecha ganadora global (para decidir canal de matriculas)
origen = {}
for r in _cartera_rows_cache():
tel = str(r.get("telefono") or "").strip()
if not tel:
continue
tels.add(tel)
sede = " ".join(str(r.get("sede") or "").upper().split())
prog = " ".join(str(r.get("programa") or "").upper().split())
canal = " ".join(str(r.get("canal") or "").upper().split())
f = L._to_date(r.get("fecha_creada"))
if str(r.get("es_origen") or "").upper() == "SI" and f is not None:
origen[tel] = f
if f is None:
continue
con_asesor = bool(str(r.get("asesor") or "").strip())
# ganador por sede+prog+tel (para la tabla)
k = (sede, prog, tel)
g = ganador.get(k)
if g is None or f < g["f"]:
ganador[k] = {"f": f, "asesor": con_asesor, "canal": canal}
# ganador global por tel (para clasificar matriculas por canal)
if tel not in gan_global or f < gan_global[tel]:
gan_global[tel] = f
tels_canal[tel] = canal
# PASO 2: armar indices con la fila ganadora ya deduplicada
idx_all = defaultdict(dict); idx_asig = defaultdict(dict)
# por clase (PAUTA/WEB/OTROS): {clase: {"all":..,"asig":..}}
CLASES = ("PAUTA", "WEB", "OTROS")
idx_cl = {cl: {"all": defaultdict(dict), "asig": defaultdict(dict)} for cl in CLASES}
tels_cl = {cl: set() for cl in CLASES}
for (sede, prog, tel), g in ganador.items():
f = g["f"]; cl = _clase(g["canal"])
idx_all[(sede, prog)][tel] = f
if g["asesor"]: idx_asig[(sede, prog)][tel] = f
idx_cl[cl]["all"][(sede, prog)][tel] = f
if g["asesor"]: idx_cl[cl]["asig"][(sede, prog)][tel] = f
# telefonos por clase (segun ganador GLOBAL) para matriculas
for tel, canal in tels_canal.items():
tels_cl[_clase(canal)].add(tel)
out = {"all": idx_all, "asig": idx_asig, "tels": tels, "origen": origen}
for cl in CLASES:
out[f"all_{cl.lower()}"] = idx_cl[cl]["all"]
out[f"asig_{cl.lower()}"] = idx_cl[cl]["asig"]
out[f"tels_{cl.lower()}"] = tels_cl[cl]
return out
return cache_get_or_set("cartera_idx", ("GLOBAL",), _load)
# ── Respuesta del dashboard de Leads (con filtros) ──────────────
def leads_dashboard(ano="TODOS", mes="TODOS", dia="TODOS", programa="TODOS", sede="TODOS"):
def _load():
leads = _leads_crudos()
cursos = _cursos_crudos()
mats = _matriculas_crudas()
# Filtro por PROGRAMA (4 grupos: TEAC / TERC / SEMINARIOS / OTROS).
# Recalcula toda la pantalla filtrando las listas base ANTES de calcular.
# - leads: grupo por 'cargo' del código de campaña
# - cursos y matrículas: grupo por dsc_programa
if programa not in ("TODOS", None):
g = str(programa).upper()
leads = [l for l in leads if L.grupo_programa_lead(l.get("cargo")) == g]
cursos = [c for c in cursos if L.grupo_programa_curso(c.get("dsc_programa")) == g]
mats = [m for m in mats if L.grupo_programa_curso(m.get("dsc_programa")) == g]
# Filtro por SEDE (recalcula toda la pantalla). Se filtran las listas base
# ANTES de calcular, para no tocar las funciones de cálculo existentes.
# - leads: sede POR CÓDIGO de campaña (NO sede_act; coincide con conteo real)
# - cursos y matrículas: sede por clasificar_sede(dsc_programa)
if sede not in ("TODOS", None):
s = str(sede).upper()
sede_cod = L.sede_por_codigo_map()
leads = [l for l in leads
if sede_cod.get(str(l.get("codigo") or "").strip()) == s]
cursos = [c for c in cursos if L.clasificar_sede(c.get("dsc_programa")) == s]
mats = [m for m in mats if L.clasificar_sede(m.get("dsc_programa")) == s]
kpis = L.kpis_leads(leads, cursos, mats, ano, mes, dia)
tabla = L.tabla_estado_objecion(leads, ano, mes, dia)
por_dia = L.matriculas_por_dia(mats, cursos, ano, mes, dia)
leads_dia = L.leads_por_dia(leads, ano, mes, dia)
pauta = L.tabla_pauta(leads, mats, ano, mes, dia,
_importe_por_pauta_periodo(ano, mes, dia),
_resultados_por_pauta_periodo(ano, mes, dia),
_campanias_sede_cargo(),
codigos_branding=_codigos_branding())
ci = _cartera_idx()
cartera_canal = {
"PAUTA": {"all": ci.get("all_pauta"), "asig": ci.get("asig_pauta"), "tels": ci.get("tels_pauta")},
"WEB": {"all": ci.get("all_web"), "asig": ci.get("asig_web"), "tels": ci.get("tels_web")},
"OTROS": {"all": ci.get("all_otros"), "asig": ci.get("asig_otros"), "tels": ci.get("tels_otros")},
}
matriz = L.matriz_cursos(cursos, leads, ano, mes, dia, _pauta_cruda(),
_campanias_map(), ci.get("all"), ci.get("asig"),
_mat_por_indice(), ci.get("tels"), ci.get("origen"),
_importe_por_pauta(),
_importe_por_pauta_periodo(ano, mes, dia),
cartera_canal, _conjunto_pauta(), _pautas_de_indice(),
_resultados_por_pauta())
return {"kpis": kpis, "estado_objecion": tabla, "matriculas_por_dia": por_dia,
"leads_por_dia": leads_dia, "tabla_pauta": pauta, "matriz_cursos": matriz}
return cache_get_or_set("leads_dash", (ano, mes, dia, programa, sede), _load)
def _asesores():
"""{'map': {alias_norm: correcto}, 'lista': [los 8 vendedores]}."""
return cache_get_or_set("asesores", ("GLOBAL",), lambda: get_dm().traer_asesores())
def _norm_asesor(nombre, mapa):
"""Normaliza un nombre de asesor a su 'correcto' via el diccionario."""
k = " ".join(str(nombre or "").upper().split())
return mapa.get(k) # None si no es uno de los vendedores
def _norm_tel_pe(v):
"""Normaliza telefono igual que la cartera: quita simbolos y el '51' inicial
de celulares peruanos (rango 51900000000-51999999999). Asi cruzan con cartera_junta."""
s = str(v or "")
for x in ("+51", "+", " ", "-", "(", ")"):
s = s.replace(x, "")
s = s.strip()
if s.isdigit() and 51900000000 <= int(s) <= 51999999999:
s = s[2:]
return s
def tiempo_respuesta_dashboard(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Tiempo de respuesta LABORAL (minutos) del asesor tras la asignacion.
- Turno detectado por dia/asesor segun donde hay MAS actividad:
tarde (13-18) -> Turno 1 (09-18, refrigerio 13-14 excluido).
noche (18-22) -> Turno 2 (09-13 + 18-22, sin refrigerio).
sabado: manana(09-13) o tarde(14-18), 4h, donde haya mas actividad.
sin actividad ese dia -> no trabajo (se salta).
- El tiempo cuenta SOLO minutos dentro del turno; si la asignacion cae fuera,
el reloj arranca al inicio del siguiente bloque laboral (saltando dias sin
actividad). Leads asignados en refrigerio (13-14) de turno1 se excluyen.
- Serie dia a dia por FECHA DE ASIGNACION: promedio de minutos de los leads
asignados ese dia (solo respondidos). Filtro ano/mes/dia por fecha asignacion.
- No respondidos -> se marcan aparte, NO entran en el promedio."""
from datetime import datetime, timedelta, time as _time
def _load():
from collections import defaultdict
ase = _asesores(); amap = ase["map"]; lista = ase["lista"]
# ── 1) Detectar turno por (asesor_norm, fecha) segun actividad ──
# Mapa user_id -> nombre normalizado (de asignacion_respuesta).
uid2vend = {}
for r in _asignacion_respuesta_crudos():
uid = r.get("user_id")
v = _norm_asesor(r.get("user_name"), amap)
if uid is not None and v:
uid2vend[uid] = v
# Contar actividad por (vend, fecha, franja): tarde=13-18, noche=18-22,
# sab_manana=9-13, sab_tarde=14-18.
act = defaultdict(lambda: defaultdict(int)) # (vend,fecha) -> franja -> conteo
for a in _actividad_asesores_crudos():
uid = a.get("user_id"); ts = a.get("created_at")
if uid not in uid2vend or not isinstance(ts, datetime):
continue
v = uid2vend[uid]; d = ts.date(); h = ts.hour + ts.minute / 60.0
key = (v, d)
if d.weekday() == 6: # domingo: no trabaja
continue
if d.weekday() == 5: # sabado
if 9 <= h < 13: act[key]["sab_manana"] += 1
elif 14 <= h < 18: act[key]["sab_tarde"] += 1
else: # lun-vie: 3 franjas
if 9 <= h < 13: act[key]["manana"] += 1
elif 14 <= h < 18: act[key]["tarde"] += 1 # tarde = 14-18 (T1)
elif 18 <= h < 22: act[key]["noche"] += 1 # noche = 18-22 (T2)
def _turno_dia(v, d):
"""Devuelve lista de bloques [(h_ini,h_fin), ...] del turno de ese dia,
o None si no trabajo. Reglas lun-vie:
Turno 1 = actividad en manana (9-13) Y en tarde (14-18).
Turno 2 = actividad en manana (9-13) Y en noche (18-22).
Si cumple ambos, gana el que tenga mas actividad (tarde vs noche).
Si no cumple ninguno completo -> None (no trabajo)."""
if d.weekday() == 6:
return None
fr = act.get((v, d))
if not fr:
return None # sin actividad -> no trabajo
if d.weekday() == 5: # sabado (4h) -> regla SIN cambios
m = fr.get("sab_manana", 0); t = fr.get("sab_tarde", 0)
if m == 0 and t == 0:
return None
return [(9, 13)] if m >= t else [(14, 18)]
# lun-vie: se exige la manana + (tarde o noche)
man = fr.get("manana", 0); t = fr.get("tarde", 0); n = fr.get("noche", 0)
cumple_t1 = man > 0 and t > 0
cumple_t2 = man > 0 and n > 0
if cumple_t1 and cumple_t2:
return [(9, 13), (18, 22)] if n > t else [(9, 13), (14, 18)]
if cumple_t2:
return [(9, 13), (18, 22)] # Turno 2
if cumple_t1:
return [(9, 13), (14, 18)] # Turno 1
return None # no cumple ningun turno completo
def _dt(d, hh):
return datetime(d.year, d.month, d.day) + timedelta(hours=hh)
def _en_refrigerio_t1(v, ts):
"""True si ts cae 13-15 y ese dia el asesor es turno 1 (lun-vie).
Estos leads NO se excluyen: cuentan como respondidos, pero su tiempo
(diferencia directa T2-T1) se reduce 45 min si supera 45."""
d = ts.date()
if d.weekday() >= 5:
return False
bloques = _turno_dia(v, d)
# Turno 1 = bloques [(9,13),(14,18)] (contiene la tarde 14-18)
es_t1 = bloques == [(9, 13), (14, 18)]
h = ts.hour + ts.minute / 60.0
return es_t1 and (13 <= h < 15)
def _minutos_laborales(v, t1, t2):
"""Minutos de turno entre t1 (asignacion) y t2 (respuesta), saltando
huecos y dias sin actividad. Devuelve None si no se puede (raro)."""
if not isinstance(t1, datetime) or not isinstance(t2, datetime) or t2 <= t1:
return 0
total = 0.0
cur = t1
saltos = 0
while cur < t2 and saltos < 60: # tope 60 dias por seguridad
d = cur.date()
bloques = _turno_dia(v, d)
if not bloques:
# dia sin trabajo -> saltar al dia siguiente 00:00
cur = _dt(d, 24); saltos += 1; continue
avanzo = False
for (hi, hf) in bloques:
ini = _dt(d, hi); fin = _dt(d, hf)
if t2 <= ini: # la respuesta es antes de este bloque
break
if cur >= fin: # ya paso este bloque
continue
a = max(cur, ini)
b = min(t2, fin)
if b > a:
total += (b - a).total_seconds() / 60.0
cur = fin # avanzar al fin del bloque
avanzo = True
if cur >= t2:
break
# pasar al dia siguiente
if cur.date() == d:
cur = _dt(d, 24); saltos += 1
return round(total)
# ── 2) Por lead: minutos laborales, agrupado por FECHA DE ASIGNACION ──
por_dia_sum = defaultdict(float) # (vend,dia) -> suma minutos
por_dia_cnt = defaultdict(int) # (vend,dia) -> nº respondidos
no_resp = defaultdict(int) # (vend,dia) -> nº sin responder
for r in _asignacion_respuesta_crudos():
v = _norm_asesor(r.get("user_name"), amap)
if not v:
continue
t1 = r.get("created_at")
if not isinstance(t1, datetime):
continue
if not L._en_periodo(t1.date(), ano, mes, dia):
continue # filtro por fecha de asignacion
t2 = r.get("respuesta_fecha")
if not isinstance(t2, datetime):
no_resp[(v, t1.day)] += 1 # no respondio
continue
if _en_refrigerio_t1(v, t1):
# asignado 13-15 en turno1: tiempo = diferencia DIRECTA (T2-T1),
# y si supera 45 min se le restan 45 (nunca negativo).
dif = (t2 - t1).total_seconds() / 60.0
mins = round(dif - 45) if dif > 45 else round(dif)
if mins < 0:
mins = 0
else:
mins = _minutos_laborales(v, t1, t2)
por_dia_sum[(v, t1.day)] += mins
por_dia_cnt[(v, t1.day)] += 1
# ── 2b) Igual pero agrupado por FECHA DE RESPUESTA (dia en que respondio).
# Solo leads respondidos. Filtro por fecha de respuesta (t2).
# Si el tiempo laboral supera 960 min (2 dias laborales) -> NO entra en
# el promedio; se cuenta aparte como "respuesta tardia".
LIMITE_TARDIA = 960 # 2 dias laborales (8h/dia)
pr_sum = defaultdict(float) # (vend,dia_respuesta) -> suma minutos (validos)
pr_cnt = defaultdict(int) # (vend,dia_respuesta) -> nº validos (<=960)
pr_tardia = defaultdict(int) # (vend,dia_respuesta) -> nº tardios (>960)
for r in _asignacion_respuesta_crudos():
v = _norm_asesor(r.get("user_name"), amap)
if not v:
continue
t1 = r.get("created_at"); t2 = r.get("respuesta_fecha")
if not isinstance(t1, datetime) or not isinstance(t2, datetime):
continue # sin respuesta -> no entra en este grafico
if not L._en_periodo(t2.date(), ano, mes, dia):
continue # filtro por FECHA DE RESPUESTA
if _en_refrigerio_t1(v, t1):
dif = (t2 - t1).total_seconds() / 60.0
mins = round(dif - 45) if dif > 45 else round(dif)
if mins < 0:
mins = 0
else:
mins = _minutos_laborales(v, t1, t2)
if mins > LIMITE_TARDIA:
pr_tardia[(v, t2.day)] += 1 # respuesta tardia -> fuera del promedio
continue
pr_sum[(v, t2.day)] += mins
pr_cnt[(v, t2.day)] += 1
# ── 3) Serie dia a dia (promedio min) por vendedor y total ──
def _serie(vends):
out = []
for dnum in range(1, 32):
s = sum(por_dia_sum.get((v, dnum), 0) for v in vends)
c = sum(por_dia_cnt.get((v, dnum), 0) for v in vends)
nr = sum(no_resp.get((v, dnum), 0) for v in vends)
out.append({"dia": dnum,
"promedio_min": round(s / c) if c else 0,
"respondidos": c, "no_respondidos": nr})
return out
def _serie_resp(vends):
out = []
for dnum in range(1, 32):
s = sum(pr_sum.get((v, dnum), 0) for v in vends)
c = sum(pr_cnt.get((v, dnum), 0) for v in vends)
td = sum(pr_tardia.get((v, dnum), 0) for v in vends)
out.append({"dia": dnum,
"promedio_min": round(s / c) if c else 0,
"respondidos": c, "tardios": td})
return out
serie_por_vend = {v: _serie([v]) for v in lista}
serie_total = _serie(lista)
serie_resp_por_vend = {v: _serie_resp([v]) for v in lista}
serie_resp_total = _serie_resp(lista)
return {"serie_por_vend": serie_por_vend, "serie_total": serie_total,
"serie_resp_por_vend": serie_resp_por_vend,
"serie_resp_total": serie_resp_total,
"vendedores": lista}
return cache_get_or_set("tiempo_resp", (ano, mes, dia), _load)
def tiempo_respuesta_detalle(dia_click, modo="asignacion", vendedor="TODOS",
ano="TODOS", mes="TODOS"):
"""Detalle de los leads que forman el promedio de UN día del gráfico
'Tiempo de Respuesta por Día'. Reproduce la MISMA lógica de turnos/minutos.
modo: 'asignacion' (agrupa por día de created_at) o 'respuesta' (por respuesta_fecha).
Devuelve, por asesor, la lista de leads con T1, T2 y minutos, y el promedio."""
from datetime import datetime, timedelta
def _load():
from collections import defaultdict
ase = _asesores(); amap = ase["map"]; lista = ase["lista"]
uid2vend = {}
for r in _asignacion_respuesta_crudos():
uid = r.get("user_id"); v = _norm_asesor(r.get("user_name"), amap)
if uid is not None and v:
uid2vend[uid] = v
act = defaultdict(lambda: defaultdict(int))
for a in _actividad_asesores_crudos():
uid = a.get("user_id"); ts = a.get("created_at")
if uid not in uid2vend or not isinstance(ts, datetime):
continue
v = uid2vend[uid]; d = ts.date(); h = ts.hour + ts.minute / 60.0
key = (v, d)
if d.weekday() == 6:
continue
if d.weekday() == 5:
if 9 <= h < 13: act[key]["sab_manana"] += 1
elif 14 <= h < 18: act[key]["sab_tarde"] += 1
else: # lun-vie: 3 franjas
if 9 <= h < 13: act[key]["manana"] += 1
elif 14 <= h < 18: act[key]["tarde"] += 1
elif 18 <= h < 22: act[key]["noche"] += 1
def _turno_dia(v, d):
# Turno 1 = manana(9-13) Y tarde(14-18). Turno 2 = manana(9-13) Y noche(18-22).
if d.weekday() == 6: return None
fr = act.get((v, d))
if not fr: return None
if d.weekday() == 5:
m = fr.get("sab_manana", 0); t = fr.get("sab_tarde", 0)
if m == 0 and t == 0: return None
return [(9, 13)] if m >= t else [(14, 18)]
man = fr.get("manana", 0); t = fr.get("tarde", 0); n = fr.get("noche", 0)
cumple_t1 = man > 0 and t > 0
cumple_t2 = man > 0 and n > 0
if cumple_t1 and cumple_t2:
return [(9, 13), (18, 22)] if n > t else [(9, 13), (14, 18)]
if cumple_t2: return [(9, 13), (18, 22)]
if cumple_t1: return [(9, 13), (14, 18)]
return None
def _dt(d, hh):
return datetime(d.year, d.month, d.day) + timedelta(hours=hh)
def _en_refrigerio_t1(v, ts):
d = ts.date()
if d.weekday() >= 5: return False
es_t1 = _turno_dia(v, d) == [(9, 13), (14, 18)]
h = ts.hour + ts.minute / 60.0
return es_t1 and (13 <= h < 15)
def _minutos_laborales(v, t1, t2):
if not isinstance(t1, datetime) or not isinstance(t2, datetime) or t2 <= t1:
return 0
total = 0.0; cur = t1; saltos = 0
while cur < t2 and saltos < 60:
d = cur.date(); bloques = _turno_dia(v, d)
if not bloques:
cur = _dt(d, 24); saltos += 1; continue
for (hi, hf) in bloques:
ini = _dt(d, hi); fin = _dt(d, hf)
if t2 <= ini: break
if cur >= fin: continue
a = max(cur, ini); b = min(t2, fin)
if b > a: total += (b - a).total_seconds() / 60.0
cur = fin
if cur >= t2: break
if cur.date() == d:
cur = _dt(d, 24); saltos += 1
return round(total)
LIMITE_TARDIA = 960
d_click = int(dia_click)
es_resp = str(modo).lower().startswith("resp")
por_asesor = defaultdict(list) # vend -> [ {t1,t2,mins} ]
for r in _asignacion_respuesta_crudos():
v = _norm_asesor(r.get("user_name"), amap)
if not v: continue
if vendedor not in ("TODOS", None, "") and v != vendedor:
continue
t1 = r.get("created_at"); t2 = r.get("respuesta_fecha")
if not isinstance(t1, datetime): continue
if es_resp:
# agrupar por fecha de respuesta; solo respondidos, <=960 min
if not isinstance(t2, datetime): continue
if not L._en_periodo(t2.date(), ano, mes, "TODOS"): continue
if t2.day != d_click: continue
if _en_refrigerio_t1(v, t1):
dif = (t2 - t1).total_seconds() / 60.0
mins = round(dif - 45) if dif > 45 else round(dif)
if mins < 0: mins = 0
else:
mins = _minutos_laborales(v, t1, t2)
if mins > LIMITE_TARDIA: continue # tardío -> fuera del promedio
else:
# agrupar por fecha de asignación; solo respondidos entran al promedio
if not L._en_periodo(t1.date(), ano, mes, "TODOS"): continue
if t1.day != d_click: continue
if not isinstance(t2, datetime): continue # sin responder: no promedia
if _en_refrigerio_t1(v, t1):
dif = (t2 - t1).total_seconds() / 60.0
mins = round(dif - 45) if dif > 45 else round(dif)
if mins < 0: mins = 0
else:
mins = _minutos_laborales(v, t1, t2)
por_asesor[v].append({
"telefono": str(r.get("telefono") or "").strip(),
"t1": t1.strftime("%d/%m/%Y %H:%M"),
"t2": t2.strftime("%d/%m/%Y %H:%M"),
"minutos": int(mins),
})
# Armar salida por asesor con su promedio; y total del día
asesores = []
tot_sum = 0; tot_cnt = 0
for v in sorted(por_asesor.keys()):
leads = sorted(por_asesor[v], key=lambda x: x["t1"])
s = sum(x["minutos"] for x in leads); c = len(leads)
tot_sum += s; tot_cnt += c
asesores.append({"asesor": v, "leads": leads,
"cantidad": c, "promedio_min": round(s / c) if c else 0})
return {"dia": d_click, "modo": ("respuesta" if es_resp else "asignacion"),
"asesores": asesores,
"total": {"cantidad": tot_cnt, "promedio_min": round(tot_sum / tot_cnt) if tot_cnt else 0}}
return cache_get_or_set("tiempo_resp_det", (dia_click, modo, vendedor, ano, mes), _load)
def vendedores_dashboard(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Por vendedor (los 8 de alias_normalizacion). Filtro Año/Mes/Día:
- Cartera (total/copito/otros): filtra por fecha_creada de cartera_junta.
- Matriculas: se cruzan por TELEFONO con los leads del mes (es_origen=SI,
fecha_creada en el periodo); la matricula cuenta en CUALQUIER fecha.
- Leads Procesados/Contactados: filtran por fecha del lead (mensaje campaña)."""
def _load():
from collections import defaultdict
ase = _asesores()
amap = ase["map"]; lista = ase["lista"]
# 1) CARTERA por vendedor. Fila ORIGEN de cada telefono (es_origen=SI) cuya
# fecha_creada cae en el periodo filtrado. Cada telefono cuenta para UN solo asesor.
# Ademas guardamos {vendedor: set(telefonos del periodo)} para cruzar matriculas.
# COPITO/OTROS: se clasifica el telefono con SU FILA ORIGEN
# (Copito = origen_base=POSTGRE O canal en PAUTA_WSP/PAUTA_WSP_FACE/COPITO).
cart_total = defaultdict(int); cart_copito = defaultdict(int); cart_otros = defaultdict(int)
tels_por_vend = defaultdict(set) # vendedor -> telefonos (es_origen=SI del periodo)
cont_tels = defaultdict(set) # vendedor -> telefonos ya contados (sin repetir)
for r in _cartera_rows_cache():
if str(r.get("es_origen") or "").strip().upper() != "SI":
continue # solo la fila origen de cada telefono
f = L._to_date(r.get("fecha_creada"))
if not L._en_periodo(f, ano, mes, dia):
continue # filtro por fecha_creada de la cartera
v = _norm_asesor(r.get("asesor"), amap)
if not v: # el asesor del origen no es uno de los 8
continue
tel = str(r.get("telefono") or "").strip()
if tel and tel in cont_tels[v]:
continue # ese telefono ya se conto para este vendedor
canal = " ".join(str(r.get("canal") or "").upper().split())
ob = " ".join(str(r.get("origen_base") or "").upper().split())
cart_total[v] += 1
if (ob == "POSTGRE") or (canal in _CANALES_COPITO):
cart_copito[v] += 1
else:
cart_otros[v] += 1
if tel:
cont_tels[v].add(tel)
tels_por_vend[v].add(tel)
# 2) MATRICULAS por vendedor. Se CRUZAN por TELEFONO con los leads del periodo
# (tels_por_vend). La matricula cuenta en CUALQUIER fecha (no se filtra por
# fch_matricula). Telefono de matricula: dsc_telefono_1, o _2 si el 1 esta vacio.
# Un telefono puede tener varias matriculas -> se cuentan todas (no unico).
# MATRICULAS: por dsc_vendedor + fch_matricula en el periodo (NO se cruza
# con la cartera por telefono). Filtro Año/Mes/Día sobre fch_matricula.
mat_total = defaultdict(int); mat_curso = defaultdict(int); mat_retir = defaultdict(int)
for m in _matriculas_crudas():
vend = _norm_asesor(m.get("dsc_vendedor"), amap)
if not vend:
continue
estado = str(m.get("estado_matricula") or "").strip().upper()
if estado == "ANU":
continue # anuladas no cuentan
fm = L._to_date(m.get("fch_matricula"))
if not L._en_periodo(fm, ano, mes, dia):
continue # filtro por fch_matricula
mat_total[vend] += 1
if estado in ("ALU", "PRE"):
mat_curso[vend] += 1
elif estado == "RET":
mat_retir[vend] += 1
# 3) LEADS Procesados/Contactados por vendedor (leads de Chatwoot). Filtro por
# fecha del lead (mensaje de campaña). Procesados = lead unico con asesor;
# Contactados = de esos, ultima_etiqueta != "NO TRABAJADO".
lead_proc = defaultdict(int); lead_cont = defaultdict(int)
etiq = defaultdict(int) # clave: (vendedor, ultima_etiqueta) -> nº leads procesados
for l in _leads_crudos():
if not l.get("es_unico"):
continue
if not L._en_periodo(l.get("fecha"), ano, mes, dia):
continue # filtro por fecha del lead
v = _norm_asesor(l.get("asesor"), amap)
if not v:
continue
lead_proc[v] += 1 # tiene asesor de los 8 => procesado
ult = l.get("ultima") or "NO TRABAJADO"
etiq[(v, ult)] += 1 # para la matriz Estado/Objeción
# NOTA: "Contactados" ya NO se cuenta aqui; ahora sale de los ASIGNADOS
# (bloque 3b), usando su ultima etiqueta.
# 3b) LEADS ASIGNADOS por vendedor (mensajes "Asignado a..." de Chatwoot).
# Se cuenta por telefono (1 asignacion por telefono, la mas reciente).
# Filtro por fecha de asignacion (created_at). Asesor via user_name normalizado.
# CONTACTADOS = de los asignados, cuya ULTIMA ETIQUETA != "NO TRABAJADO".
lead_asig = defaultdict(int)
etiq_asig = defaultdict(int) # (vendedor, ultima_etiqueta) -> nº asignados (para matriz)
for a in _leads_asignados_crudos():
f = L._to_date(a.get("created_at"))
if not L._en_periodo(f, ano, mes, dia):
continue # filtro por fecha de asignacion
v = _norm_asesor(a.get("user_name"), amap)
if not v:
continue # el asesor asignado no es uno de los 8
lead_asig[v] += 1
ult = L.ultima_etiqueta(a.get("etiquetas")) or "NO TRABAJADO"
etiq_asig[(v, ult)] += 1 # para la matriz Estado/Objeción (asignados)
if ult != "NO TRABAJADO":
lead_cont[v] += 1 # contactado = asignado con ultima etiqueta trabajada
# SERIE DIARIA: Leads Asignados (por fecha de asignacion) y Matriculas (por fch_matricula),
# por vendedor y por dia del mes. clave: (vendedor, dia) -> conteo.
serie_proc = defaultdict(int) # (vendedor, dia) -> leads asignados
serie_mat = defaultdict(int) # (vendedor, dia) -> matriculas
for a in _leads_asignados_crudos():
f = L._to_date(a.get("created_at"))
if not L._en_periodo(f, ano, mes, dia):
continue
v = _norm_asesor(a.get("user_name"), amap)
if v and f:
serie_proc[(v, f.day)] += 1
for m in _matriculas_crudas():
vend = _norm_asesor(m.get("dsc_vendedor"), amap)
if not vend:
continue
estado = str(m.get("estado_matricula") or "").strip().upper()
if estado == "ANU":
continue
fm = L._to_date(m.get("fch_matricula"))
if fm and L._en_periodo(fm, ano, mes, dia):
serie_mat[(vend, fm.day)] += 1
def _serie(vendedores):
"""Lista 1..31 con asignados y matriculas del conjunto de vendedores."""
out = []
for d in range(1, 32):
p = sum(serie_proc.get((v, d), 0) for v in vendedores)
mm = sum(serie_mat.get((v, d), 0) for v in vendedores)
out.append({"dia": d, "asignados": p, "matriculas": mm})
return out
serie_por_vend = {v: _serie([v]) for v in lista}
serie_total = _serie(lista)
# Matriz Estado/Objeción (Ultima_Etiqueta) por vendedor y total.
# Ahora se basa en los LEADS ASIGNADOS (etiq_asig), no en los de campaña.
etiquetas_all = sorted({e for (_, e) in etiq_asig.keys()}, key=lambda x: x.lower())
def _matriz_etiq(vendedores):
out = []
for e in etiquetas_all:
c = sum(etiq_asig.get((v, e), 0) for v in vendedores)
if c > 0:
out.append({"estado": e, "cantidad": c})
out.sort(key=lambda x: -x["cantidad"]) # mayor a menor por cantidad
return out
matriz_etiq_por_vend = {v: _matriz_etiq([v]) for v in lista}
matriz_etiq_total = _matriz_etiq(lista)
# 4) MATRIZ por TIPO DE PROGRAMA (TEAC/TERC/SEMINARIOS/OTROS).
# Filtro por fch_matricula (periodo) y dsc_vendedor. Total y Retiradas (RET).
# Se guarda por (vendedor, grupo) para poder filtrar por vendedor en el front.
GRUPOS_PROG = ["TEAC", "TERC", "SEMINARIOS", "OTROS"]
prog_tot = defaultdict(int); prog_ret = defaultdict(int) # clave: (vendedor, grupo)
for m in _matriculas_crudas():
vend = _norm_asesor(m.get("dsc_vendedor"), amap)
if not vend:
continue
estado = str(m.get("estado_matricula") or "").strip().upper()
if estado == "ANU":
continue
fm = L._to_date(m.get("fch_matricula"))
if not L._en_periodo(fm, ano, mes, dia):
continue # filtro por fch_matricula
grupo = L.grupo_programa_curso(m.get("dsc_programa"))
prog_tot[(vend, grupo)] += 1
if estado == "RET":
prog_ret[(vend, grupo)] += 1
def _matriz_prog(vendedores):
"""Suma total/retiradas por grupo para el conjunto de vendedores dado."""
out = []
for g in GRUPOS_PROG:
t = sum(prog_tot.get((v, g), 0) for v in vendedores)
r = sum(prog_ret.get((v, g), 0) for v in vendedores)
out.append({"programa": g, "total": t, "retiradas": r,
"pct_retiradas": round(r / t * 100, 1) if t else 0})
return out
# matriz por programa para CADA vendedor y para el TOTAL (todos)
matriz_prog_por_vend = {v: _matriz_prog([v]) for v in lista}
matriz_prog_total = _matriz_prog(lista)
filas = []
for v in lista:
proc = lead_proc.get(v, 0); cont = lead_cont.get(v, 0)
asig = lead_asig.get(v, 0)
filas.append({
"vendedor": v,
"cartera_total": cart_total.get(v, 0),
"cartera_copito": cart_copito.get(v, 0),
"cartera_otros": cart_otros.get(v, 0),
"mat_total": mat_total.get(v, 0),
"mat_curso": mat_curso.get(v, 0),
"mat_retiradas": mat_retir.get(v, 0),
"pct_retiradas": round(mat_retir.get(v, 0) / mat_total.get(v, 0) * 100, 1)
if mat_total.get(v, 0) else 0,
"leads_procesados": proc,
"leads_contactados": cont,
# % Contactados = contactados / ASIGNADOS (ambos de la misma fuente)
"pct_contactados": round(cont / asig * 100, 1) if asig else 0,
"leads_asignados": asig,
})
tot = {"cartera_total": sum(f["cartera_total"] for f in filas),
"cartera_copito": sum(f["cartera_copito"] for f in filas),
"cartera_otros": sum(f["cartera_otros"] for f in filas),
"mat_total": sum(f["mat_total"] for f in filas),
"mat_curso": sum(f["mat_curso"] for f in filas),
"mat_retiradas": sum(f["mat_retiradas"] for f in filas),
"leads_procesados": sum(f["leads_procesados"] for f in filas),
"leads_contactados": sum(f["leads_contactados"] for f in filas),
"leads_asignados": sum(f["leads_asignados"] for f in filas)}
tot["pct_retiradas"] = (round(tot["mat_retiradas"] / tot["mat_total"] * 100, 1)
if tot["mat_total"] else 0)
tot["pct_contactados"] = (round(tot["leads_contactados"] / tot["leads_asignados"] * 100, 1)
if tot["leads_asignados"] else 0)
# Tiempo de respuesta laboral (min) dia a dia por asesor (por fecha asignacion)
tr = tiempo_respuesta_dashboard(ano, mes, dia)
return {"filas": filas, "total": tot, "vendedores": lista,
"matriz_prog": matriz_prog_por_vend, "matriz_prog_total": matriz_prog_total,
"matriz_etiq": matriz_etiq_por_vend, "matriz_etiq_total": matriz_etiq_total,
"serie_dia": serie_por_vend, "serie_dia_total": serie_total,
"tiempo_resp": tr.get("serie_por_vend", {}),
"tiempo_resp_total": tr.get("serie_total", []),
"tiempo_resp_r": tr.get("serie_resp_por_vend", {}),
"tiempo_resp_r_total": tr.get("serie_resp_total", [])}
return cache_get_or_set("vendedores_dash", (ano, mes, dia), _load)
def _conjunto_sede_prog():
"""{conjunto_UPPER: {sede, programa}} asignaciones manuales (tabla Supabase)."""
return cache_get_or_set("conjunto_sede_prog", ("GLOBAL",),
lambda: get_dm().traer_conjunto_sede_programa())
# Programas cuyos RESULTADOS no deben sumar (son de otro tipo: alcance/marca).
_PROG_SIN_RESULTADO = {"BRANDING", "COPEAIRE"}
def _conjuntos_branding():
"""Set de CONJUNTOS (UPPER) asignados manualmente como BRANDING/COPEAIRE.
Sus resultados NO deben contarse (su importe sí)."""
def _load():
asign = _conjunto_sede_prog() # {conjunto_UPPER: {sede, programa}}
return {k for k, v in asign.items()
if str(v.get("programa") or "").strip().upper() in _PROG_SIN_RESULTADO}
return cache_get_or_set("conjuntos_branding", ("GLOBAL",), _load)
def _codigos_branding():
"""Set de codigos(pauta) cuyos conjuntos estan asignados manualmente como
BRANDING/COPEAIRE. Sirve para NO contar sus resultados en las tablas."""
def _load():
conj_pauta = _conjunto_pauta() # {pauta: [conjuntos]}
asign = _conjunto_sede_prog() # {conjunto_UPPER: {sede, programa}}
# conjuntos_branding (UPPER)
conj_brand = {k for k, v in asign.items()
if str(v.get("programa") or "").strip().upper() in _PROG_SIN_RESULTADO}
cods = set()
for pauta, conjuntos in conj_pauta.items():
for co in conjuntos:
if " ".join(str(co).split()).upper() in conj_brand:
cods.add(str(pauta).strip())
break
return cods
return cache_get_or_set("codigos_branding", ("GLOBAL",), _load)
def conjuntos_sin_sede():
"""Lista los conjuntos del Meta CSV que NO estan ligados a ninguna sede
(ni por pauta->campania, ni por asignacion manual). Para el apartado de
asignacion. Sin filtro de mes. Incluye la sede/programa YA asignados (si tiene).
Devuelve [{conjunto, importe, resultados, sede, programa}]."""
def _load():
SEDES = {"LIMA", "PIURA", "AREQUIPA", "TRUJILLO"}
conj_pauta = _conjunto_pauta()
camp = _campanias_sede_cargo()
manual = _conjunto_sede_prog()
# conjuntos que SÍ tienen sede via pauta -> campania
con_sede = set()
for pauta, conjuntos in conj_pauta.items():
sp = camp.get(str(pauta).strip())
sede = " ".join(str((sp or {}).get("sede") or "").upper().split())
if sp and sede in SEDES:
for co in conjuntos:
con_sede.add(" ".join(str(co).split()).upper())
agg = {}
for f in _meta_filas():
co = " ".join(str(f.get("conjunto")).split())
if not co:
continue
k = co.upper()
if k in con_sede:
continue # ya tiene sede por campania -> no aparece aqui
d = agg.setdefault(k, {"conjunto": co, "importe": 0.0, "resultados": 0,
"sede": manual.get(k, {}).get("sede", ""),
"programa": manual.get(k, {}).get("programa", "")})
d["importe"] += float(f.get("importe") or 0.0)
d["resultados"] += int(float(f.get("resultados") or 0.0))
out = sorted(agg.values(), key=lambda x: -x["importe"])
for d in out:
d["importe"] = round(d["importe"], 2)
return out
return cache_get_or_set("conjuntos_sin_sede", ("GLOBAL",), _load)
def guardar_conjunto_sede(cambios):
"""Guarda asignaciones conjunto->sede/programa e invalida cachés dependientes."""
r = get_dm().guardar_conjunto_sede_programa(cambios)
for pref in ("conjunto_sede_prog", "conjuntos_sin_sede", "roas_dash"):
cache_invalidate(prefix=pref)
return r
def borrar_conjunto_sede(conjunto):
"""Borra la asignacion de un conjunto e invalida cachés dependientes."""
r = get_dm().borrar_conjunto_sede_programa(conjunto)
for pref in ("conjunto_sede_prog", "conjuntos_sin_sede", "roas_dash"):
cache_invalidate(prefix=pref)
return r
def roas_dashboard(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Matriz ROAS por SEDE (LIMA, PIURA, AREQUIPA, TRUJILLO): Importe Gastado y
Resultados del Google Sheet (Meta Ads), filtrado por Año/Mes (Inicio del informe).
Cadena: conjunto -> pauta (conjunto_pauta) -> sede (campanias). Conjuntos sin
pauta/sede se omiten por ahora."""
def _load():
SEDES = ["LIMA", "PIURA", "AREQUIPA", "TRUJILLO", "OTROS"] # OTROS = 5ta fila (manual)
CAMP_SEDES = {"LIMA", "PIURA", "AREQUIPA", "TRUJILLO"} # sedes validas por campania
conj_pauta = _conjunto_pauta() # {pauta: [conjunto,...]}
camp = _campanias_sede_cargo() # {pauta: {sede, cargo}}
PROGS_OK = {"TEAC", "TERC", "SEMINARIOS", "OTROS"}
# Programas cuyos RESULTADOS NO deben sumar (son de otro tipo: alcance/marca).
# Su importe SÍ suma; sus resultados se muestran solo en el detalle, y el
# subtotal del programa aparece en 0 para no inflar los totales.
PROG_SIN_RESULTADO = {"BRANDING", "COPEAIRE"}
# conjunto (normalizado) -> (sede, programa) via su pauta -> campania
conj_sede = {} # conjunto -> sede
conj_prog = {} # conjunto -> programa (grupo)
for pauta, conjuntos in conj_pauta.items():
sp = camp.get(str(pauta).strip())
if not sp:
continue
sede = " ".join(str(sp.get("sede") or "").upper().split())
if sede not in CAMP_SEDES:
continue
prog = L.grupo_programa_lead(sp.get("cargo")) # TEAC/TERC/SEMINARIOS
for co in conjuntos:
k = " ".join(str(co).split()).upper()
conj_sede[k] = sede
conj_prog[k] = prog
# + asignaciones MANUALES (sede + programa). Prioridad sobre la campania.
for k, v in _conjunto_sede_prog().items():
sede = v.get("sede")
if sede in SEDES:
conj_sede[k] = sede
prog = str(v.get("programa") or "").strip().upper()
if prog: # acepta CUALQUIER programa asignado manualmente (no solo los fijos)
conj_prog[k] = prog
# conjunto(UPPER) -> [pautas] (para cruzar con leads por codigo)
pautas_de_conj = {}
for pauta, conjuntos in conj_pauta.items():
for co in conjuntos:
k = " ".join(str(co).split()).upper()
pautas_de_conj.setdefault(k, []).append(str(pauta).strip())
# leads (recibidos/procesados del periodo) por codigo de pauta
lxc = _leads_por_codigo_periodo(ano, mes, dia)
# sumar importe y resultados por sede y por (sede, programa), filtrando por Año/Mes
imp = {s: 0.0 for s in SEDES}
res = {s: 0 for s in SEDES}
imp_pg = {} # (sede, prog) -> importe
res_pg = {} # (sede, prog) -> resultados
# tels de leads por nivel (sets para no duplicar telefonos)
rec_s = {s: set() for s in SEDES}; proc_s = {s: set() for s in SEDES} # por sede
rec_pg = {}; proc_pg = {} # (sede,prog)
# conjuntos por (sede, prog): nombre_conjunto -> {importe, resultados}
conj_det = {} # (sede, prog) -> { conjunto_original: {importe, resultados, rec, proc} }
sin_filtro = (ano in ("TODOS", None) and mes in ("TODOS", None) and dia in ("TODOS", None))
for f in _meta_filas():
if not sin_filtro and not L._en_periodo(L._to_date(f.get("inicio")), ano, mes, dia):
continue
co = " ".join(str(f.get("conjunto")).split())
k = co.upper()
sede = conj_sede.get(k)
if not sede:
continue # conjunto sin pauta/sede -> se omite
i = float(f.get("importe") or 0.0)
rr = int(float(f.get("resultados") or 0.0))
prog = conj_prog.get(k, "SIN PROGRAMA")
# BRANDING/COPEAIRE: su importe suma, pero sus resultados NO cuentan en
# los totales (sede y general). Solo se ven en el detalle del conjunto.
rr_sede = 0 if prog in PROG_SIN_RESULTADO else rr
imp[sede] += i
res[sede] += rr_sede
imp_pg[(sede, prog)] = imp_pg.get((sede, prog), 0.0) + i
res_pg[(sede, prog)] = res_pg.get((sede, prog), 0) + rr_sede
dd = conj_det.setdefault((sede, prog), {})
d = dd.setdefault(co, {"conjunto": co, "importe": 0.0, "resultados": 0,
"rec": 0, "proc": 0})
d["importe"] += i
d["resultados"] += rr
# ── Leads Recibidos/Procesados del mes por conjunto (via su pauta -> codigo).
# Se hace UNA vez por conjunto (no por fila del CSV). Los sets evitan
# duplicar telefonos al agregar por programa y por sede.
for (sede, prog), dd in conj_det.items():
for co, d in dd.items():
k = co.upper()
rec = set(); proc = set()
for pa in pautas_de_conj.get(k, []):
lp = lxc.get(pa)
if lp:
rec |= lp["rec"]; proc |= lp["proc"]
d["rec"] = len(rec); d["proc"] = len(proc)
rec_pg.setdefault((sede, prog), set()).update(rec)
proc_pg.setdefault((sede, prog), set()).update(proc)
rec_s[sede].update(rec); proc_s[sede].update(proc)
ORDEN_PROG = ["TEAC", "TERC", "SEMINARIOS", "OTROS", "SIN PROGRAMA"]
def _conjuntos(sede, prog):
"""Lista de conjuntos de anuncio de un (sede, programa), ordenada por importe."""
det = list(conj_det.get((sede, prog), {}).values())
det.sort(key=lambda x: -x["importe"])
for d in det:
d["importe"] = round(d["importe"], 2)
return det
def _sub(sede):
"""Desglose por programa de una sede (solo los que tienen algo).
Primero los fijos en su orden, luego CUALQUIER otro programa asignado
(ej. BRANDING), y al final SIN PROGRAMA."""
progs_de_sede = [p for (s, p) in imp_pg.keys() if s == sede]
extras = sorted(set(progs_de_sede) - set(ORDEN_PROG))
orden = [p for p in ORDEN_PROG if p != "SIN PROGRAMA"] + extras + ["SIN PROGRAMA"]
out = []
for p in orden:
i = imp_pg.get((sede, p), 0.0); r = res_pg.get((sede, p), 0)
if i or r:
out.append({"programa": p, "importe": round(i, 2), "resultados": r,
"rec": len(rec_pg.get((sede, p), set())),
"proc": len(proc_pg.get((sede, p), set())),
"conjuntos": _conjuntos(sede, p)})
return out
# Las 4 sedes reales siempre se muestran. OTROS solo si tiene algo (>0).
filas = []
for s in SEDES:
if s == "OTROS" and imp[s] == 0 and res[s] == 0:
continue
filas.append({"sede": s, "importe": round(imp[s], 2), "resultados": res[s],
"rec": len(rec_s[s]), "proc": len(proc_s[s]),
"programas": _sub(s)})
tot_rec = set(); tot_proc = set()
for s in SEDES:
tot_rec |= rec_s[s]; tot_proc |= proc_s[s]
tot = {"importe": round(sum(imp.values()), 2), "resultados": sum(res.values()),
"rec": len(tot_rec), "proc": len(tot_proc)}
return {"filas": filas, "total": tot}
return cache_get_or_set("roas_dash", (ano, mes, dia), _load)
_CURSOS_EXCLUIDOS = {"1121", "1154"}
def _curso_excluido(c, pmap):
"""Mismo criterio que 'Detalle por Curso': excluye cursos en la lista fija,
o con contar=NO / estado=SUSPENDIDO en basebi."""
ni = str(c.get("num_indice", "")).strip()
if ni in _CURSOS_EXCLUIDOS:
return True
info = pmap.get(ni)
if not info:
return False
return info.get("contar") == "NO" or info.get("estado") == "SUSPENDIDO"
def programa_opciones(ano="TODOS", mes="TODOS"):
"""Lista de programas (dsc_det_programa) cuyos cursos tienen fch_inicio en el
periodo. Para el selector del apartado PROGRAMA. Ordenada por fch_inicio ASC."""
def _load():
from datetime import date
pmap = _pauta_cruda()
prog_fecha = {} # nombre -> fecha_inicio mas temprana
for c in _cursos_crudos():
f = L._to_date(c.get("fch_inicio"))
if not L._en_periodo(f, ano, mes, "TODOS"):
continue
if _curso_excluido(c, pmap): # mismo filtro que 'Detalle por Curso'
continue
nom = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()
if not nom:
continue
if nom not in prog_fecha or (f and f < prog_fecha[nom]):
prog_fecha[nom] = f or date.max
# ordenar por fecha de inicio ascendente
return [n for n, _ in sorted(prog_fecha.items(), key=lambda x: (x[1] or date.max, x[0]))]
return cache_get_or_set("programa_opciones", (ano, mes), _load)
def programa_dashboard(ano="TODOS", mes="TODOS", programa="TODOS"):
"""Datos agregados de UN programa (dsc_det_programa) en el periodo:
Cartera Total/Copito/Otros, Leads Recibidos/Procesados/Contactados y %.
- Filtro por fch_inicio del curso para elegir los cursos del programa.
- Cartera: teléfonos origen (es_origen=SI) cuya sede+programa (via pauta->
campania) coincide con los cursos del programa, en el periodo.
- Leads: por codigo de pauta de esos cursos (recibidos/procesados/contactados
del periodo), sin duplicar telefonos."""
def _load():
prog_sel = str(programa or "").strip().upper()
pmap = _pauta_cruda()
# 1) cursos del programa (dsc_det_programa) en el periodo -> sus pautas.
# Aplica el MISMO filtro de exclusion que 'Detalle por Curso'.
cods = set()
cursos_prog = [] # [(fch_inicio, [pautas del curso])] para la cartera
for c in _cursos_crudos():
nom = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip().upper()
if prog_sel != "TODOS" and nom != prog_sel:
continue
f = L._to_date(c.get("fch_inicio"))
if not L._en_periodo(f, ano, mes, "TODOS"):
continue
if _curso_excluido(c, pmap):
continue
ni = _norm_ni(c.get("num_indice"))
cds_curso = set()
for p in _pautas_de_indice().get(ni, []):
cds_curso.add(str(p).strip())
info = _pauta_cruda().get(ni)
if info and info.get("pauta"):
cds_curso.add(str(info.get("pauta")).strip())
cods |= cds_curso
cursos_prog.append((f, list(cds_curso)))
# 2) LEADS por esas pautas (recibidos/procesados/contactados TOTAL/historico,
# sin filtro de mes -> igual que la columna "L. Recibidos" de Detalle por Curso).
lxc = _leads_por_codigo_periodo("TODOS", "TODOS", "TODOS")
rec = set(); proc = set(); cont = set()
for cd in cods:
lp = lxc.get(cd)
if lp:
rec |= lp["rec"]; proc |= lp["proc"]; cont |= lp.get("cont", set())
# 2b) MATRIZ ESTADO/OBJECION: por ultima etiqueta de los leads (procesados)
# de las pautas del programa. Telefonos unicos por etiqueta.
from collections import defaultdict as _dd
etiq_tels = _dd(set)
etiq_ase_tels = _dd(lambda: _dd(set)) # estado -> asesor -> {telefonos}
cods_set0 = set(cods)
for l in _leads_crudos():
if not l.get("es_unico") or not l.get("asesor"):
continue
if str(l.get("codigo", "-")).strip() not in cods_set0:
continue
tel = l.get("telefono")
if not tel:
continue
ult = l.get("ultima") or "NO TRABAJADO"
ase = l.get("asesor") or "SIN ASESOR"
etiq_tels[ult].add(tel)
etiq_ase_tels[ult][ase].add(tel)
matriz_etiq = sorted(
[{"estado": k, "cantidad": len(v),
"asesores": sorted(
[{"asesor": a, "cantidad": len(t), "telefonos": sorted(t)}
for a, t in etiq_ase_tels[k].items()],
key=lambda x: -x["cantidad"])}
for k, v in etiq_tels.items()],
key=lambda x: -x["cantidad"])
# 3) CARTERA TOTAL: MISMA logica que 'Detalle por Curso'. Por cada curso,
# telefonos con la sede+programa de sus pautas y fecha_creada <= fch_inicio.
# Union entre cursos del programa (sin duplicar telefono).
# COPITO/OTROS: se clasifica cada telefono por su FILA MAS ANTIGUA DENTRO
# de esta seleccion (Copito = origen_base=POSTGRE O canal en
# PAUTA_WSP/PAUTA_WSP_FACE/COPITO).
cmap = _campanias_map() # {codigo: (sede, programa)}
# {(sede,prog): fch_corte maximo} de las pautas de los cursos del programa.
sp_corte = {}
for (fch, cds_curso) in cursos_prog:
if not fch:
continue
for cd in cds_curso:
sp = cmap.get(str(cd).upper())
if not sp:
continue
if sp not in sp_corte or fch > sp_corte[sp]:
sp_corte[sp] = fch
# recorrer cartera cruda; fila mas antigua por telefono dentro de la
# seleccion (sede+prog de una pauta + fecha_creada <= fch_inicio del curso).
mejor = {} # tel -> (fecha, es_copito)
for r in _cartera_rows_cache():
sede = " ".join(str(r.get("sede") or "").upper().split())
prog = " ".join(str(r.get("programa") or "").upper().split())
corte = sp_corte.get((sede, prog))
if corte is None:
continue
f = L._to_date(r.get("fecha_creada"))
if f is None or f > corte:
continue
tel = str(r.get("telefono") or "").strip()
if not tel:
continue
if tel not in mejor or f < mejor[tel][0]:
canal = " ".join(str(r.get("canal") or "").upper().split())
ob = " ".join(str(r.get("origen_base") or "").upper().split())
es_cop = (ob == "POSTGRE") or (canal in _CANALES_COPITO)
mejor[tel] = (f, es_cop)
cart_total = len(mejor)
cart_cop = sum(1 for (f, ec) in mejor.values() if ec)
cart_otros = cart_total - cart_cop
# 4) IMPORTE (gasto Meta) y RESULTADOS totales de las pautas del programa.
imp_map = _importe_por_pauta(); res_map = _resultados_por_pauta()
importe = round(sum(float(imp_map.get(cd, 0.0)) for cd in cods), 2)
resultados = int(sum(int(res_map.get(cd, 0)) for cd in cods))
tiene_pauta = len(cods) > 0
# 4b) V. VENTA INICIAL = suma de INV_NETA (en soles) de las matriculas del
# programa (dsc_promocion == programa). Igual que Rentabilidad: DOL -> soles
# con el imp_tc del comprobante (fallback 3.45 si viene 0/invalido).
# ROAS = venta_inicial (soles) / (importe pauta en soles = importe * 3.5).
# cartera para clasificar matriculas (nuevo/antiguo/no iden)
_ci = _cartera_idx()
cart_origen = _ci.get("origen", {}) # tel -> fecha origen del lead
cart_tels = _ci.get("tels", set()) # telefonos que estan en cartera
venta_inicial = 0.0
mat_por_fecha = {} # date -> cantidad de matriculas (no ANU)
mat_no_iden = 0; mat_nuevos = 0; mat_antiguos = 0
dias_lista = [] # dias lead->matricula (promedio general)
dias_nuevos = [] # dias de los <= 45
dias_antiguos = [] # dias de los > 45
for m in _matriculas_crudas():
nomp = str(m.get("dsc_promocion") or m.get("dsc_programa") or "").strip().upper()
if nomp != prog_sel:
continue
if str(m.get("estado_matricula") or "").strip().upper() == "ANU":
continue
try:
inv = float(m.get("INV_NETA", 0.0) or 0.0)
except Exception:
inv = 0.0
if str(m.get("cod_moneda") or "").strip().upper() == "DOL":
try:
tc = float(m.get("imp_tc", 0) or 0)
except Exception:
tc = 0
if tc < 2:
tc = 3.45 # TC por defecto, igual que Rentabilidad
inv *= tc
venta_inicial += inv
# serie de matriculas por dia (fch_matricula)
fm = L._to_date(m.get("fch_matricula"))
if fm:
mat_por_fecha[fm] = mat_por_fecha.get(fm, 0) + 1
# telefono de la matricula (tel_1 o tel_2)
ph = str(m.get("dsc_telefono_1") or "").strip() or str(m.get("dsc_telefono_2") or "").strip()
if not ph or ph not in cart_tels:
mat_no_iden += 1
continue
forig = cart_origen.get(ph)
if fm and forig:
d = (fm - forig).days
if 0 <= d <= 45:
mat_nuevos += 1
dias_nuevos.append(d)
else:
mat_antiguos += 1
if d > 45:
dias_antiguos.append(d)
if d >= 0:
dias_lista.append(d) # promedio general: TODOS los que tienen origen
else:
mat_antiguos += 1
venta_inicial = round(venta_inicial, 2) # se muestra al 100%
tiempo_mat_prom = round(sum(dias_lista) / len(dias_lista)) if dias_lista else 0
prom_nuevos = round(sum(dias_nuevos) / len(dias_nuevos)) if dias_nuevos else 0
prom_antiguos = round(sum(dias_antiguos) / len(dias_antiguos)) if dias_antiguos else 0
importe_soles = round(importe * 3.5, 2) # importe Meta (dolares) -> soles con 3.5
# ROAS usa el 85% de la venta inicial (descuento estimado de retiros/mora)
roas = round((venta_inicial * 0.85) / importe_soles, 2) if importe_soles else 0.0
# 5) SERIE DIARIA (por fecha del lead) para el grafico: recibidos/procesados/
# contactados por FECHA COMPLETA (dd/mm/yyyy). NO filtra por mes: muestra
# TODO el historial del programa. Solo aparecen dias con al menos 1 lead.
cods_set = set(cods)
por_fecha = {} # date -> {"rec":set,"proc":set,"cont":set}
for l in _leads_crudos():
if not l.get("es_unico"):
continue
if str(l.get("codigo", "-")).strip() not in cods_set:
continue
fx = l.get("fecha")
if not fx:
continue
tel = l.get("telefono")
if not tel:
continue
d = por_fecha.setdefault(fx, {"rec": set(), "proc": set(), "cont": set()})
d["rec"].add(tel)
if l.get("asesor"):
d["proc"].add(tel)
if (l.get("ultima") or "NO TRABAJADO") != "NO TRABAJADO":
d["cont"].add(tel)
_MES_ES = ["", "Enero", "Febrero", "Marzo", "Abril", "Mayo", "Junio",
"Julio", "Agosto", "Setiembre", "Octubre", "Noviembre", "Diciembre"]
serie = [{"fecha": f.strftime("%d/%m/%Y"),
"dia": f.day,
"mes": f"{_MES_ES[f.month]} {f.year}",
"recibidos": len(v["rec"]),
"procesados": len(v["proc"]),
"contactados": len(v["cont"])}
for f, v in sorted(por_fecha.items())]
# 5b) LEADS SIN RESPONDER por dia: de los telefonos del programa (rec),
# los que estan asignados pero NO tienen respuesta del asesor.
# Se agrupa por dia de la FECHA DE ASIGNACION (created_at).
from datetime import datetime as _dt
sin_resp_dia = {} # date -> set(tel)
for a in _asignacion_respuesta_crudos():
tel = str(a.get("telefono") or "").strip()
if tel not in rec: # solo leads de este programa
continue
t2 = a.get("respuesta_fecha")
if isinstance(t2, _dt):
continue # tiene respuesta -> no cuenta
t1 = a.get("created_at")
if not isinstance(t1, _dt):
continue
sin_resp_dia.setdefault(t1.date(), set()).add(tel)
serie_sin_resp = [{"fecha": f.strftime("%d/%m/%Y"), "dia": f.day,
"mes": f"{_MES_ES[f.month]} {f.year}",
"cantidad": len(v)}
for f, v in sorted(sin_resp_dia.items()) if len(v) >= 1]
# 5c) SERIE de MATRICULAS por dia (fch_matricula) del programa.
serie_mat = [{"fecha": f.strftime("%d/%m/%Y"), "dia": f.day,
"mes": f"{_MES_ES[f.month]} {f.year}", "cantidad": n}
for f, n in sorted(mat_por_fecha.items()) if n >= 1]
n_rec = len(rec); n_proc = len(proc); n_cont = len(cont)
return {
"programa": programa,
"tiene_pauta": tiene_pauta,
"serie": serie,
"serie_sin_resp": serie_sin_resp,
"serie_mat": serie_mat,
"matriz_etiq": matriz_etiq,
"importe": importe,
"resultados": resultados,
"cpl": round(importe / resultados, 2) if resultados else 0.0, # CPL = importe / resultados
"venta_inicial": venta_inicial, # V. Venta Inicial (soles)
"roas": roas, # Venta inicial / (importe * 3.5)
"mat_no_iden": mat_no_iden,
"mat_nuevos": mat_nuevos,
"mat_antiguos": mat_antiguos,
"tiempo_mat_prom": tiempo_mat_prom, # dias promedio lead->matricula
"prom_nuevos": prom_nuevos, # dias prom de los <=45
"prom_antiguos": prom_antiguos, # dias prom de los >45
"cartera_total": cart_total,
"cartera_copito": cart_cop,
"cartera_otros": cart_otros,
"leads_recibidos": n_rec,
"leads_procesados": n_proc,
"pct_procesados": round(n_proc / n_rec * 100, 1) if n_rec else 0.0,
"leads_contactados": n_cont,
"pct_contactados": round(n_cont / n_proc * 100, 1) if n_proc else 0.0,
}
return cache_get_or_set("programa_dash", (ano, mes, programa), _load)
def otros_general(ano="TODOS", mes="TODOS", dia="TODOS"):
"""Solo las 4 matrices de la pagina 'Otros General' (sin KPIs/graficos/tabla_pauta).
Mas liviano que leads_dashboard -> filtrar por mes es mas rapido."""
def _load():
return {
"matriz_always": matriz_always(ano, mes, dia, "TODOS", "TODOS"),
"matriz_webform": matriz_web_formulario(ano, mes, dia, "TODOS", "TODOS"),
"matriz_asignados": matriz_asignados(ano, mes, dia),
"matriz_plantillas": matriz_plantillas(ano, mes, dia),
}
return cache_get_or_set("otros_general", (ano, mes, dia), _load)
def conjuntos_sin_pauta():
"""Lista de conjuntos de anuncios sin pauta (para el desplegable del pop-up)."""
return get_dm().traer_conjuntos_sin_pauta()
def _pautas_de_curso_ni(ni):
"""Pautas (codigos) de un num_indice = pauta de basebi + programa_pautas."""
cds = set()
for p in _pautas_de_indice().get(ni, []):
cds.add(str(p).strip())
info = _pauta_cruda().get(ni)
if info and info.get("pauta"):
cds.add(str(info.get("pauta")).strip())
return cds
def alertas_dashboard():
"""ALERTAS: cursos que inician en los proximos 30 dias (con pauta, importe>0).
Estado (CRITICO/ADVERTENCIA/BUEN RUMBO) segun MATRICULAS vs meta ESPERADA por
dias que faltan. CPL alto / pocos leads son SENALES (no cambian el estado).
Pares = cursos mismo sede+programa con importe>0 (info real desde ~diciembre)."""
def _load():
import datetime as _dt
from collections import defaultdict as _dd
HOY = _dt.date.today()
LIM = HOY + _dt.timedelta(days=30)
imp_map = _importe_por_pauta()
res_map = _resultados_por_pauta()
lxc = _leads_por_codigo_periodo("TODOS", "TODOS", "TODOS")
mat_idx = _mat_por_indice()
pmap = _pauta_cruda()
# telefono -> set(num_indice donde se matriculo) para el analisis "este vs otros"
tel_mats = _dd(set)
for ni_m, filas in mat_idx.items():
for (tel, _f) in filas:
if tel:
tel_mats[str(tel).strip()].add(ni_m)
def datos(c):
ni = _norm_ni(c.get("num_indice"))
cds = _pautas_de_curso_ni(ni)
importe = round(sum(float(imp_map.get(cd, 0.0)) for cd in cds), 2)
resultados = int(sum(int(res_map.get(cd, 0)) for cd in cds))
rec = set(); proc = set()
for cd in cds:
lp = lxc.get(cd)
if lp:
rec |= lp["rec"]; proc |= lp["proc"]
nom = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()
info = pmap.get(ni) or {}
susp = (str(c.get("cod_estado", "")).strip().upper() == "SUS"
or info.get("estado") == "SUSPENDIDO")
return {"ni": ni, "nom": nom,
"sede": clasificar_sede_meta(nom), "prog": clasificar_prog_meta(nom),
"fch": L._to_date(c.get("fch_inicio")),
"importe": importe, "resultados": resultados,
"rec": rec, "proc": proc, "suspendido": susp,
"mats": len(mat_idx.get(ni, [])), "meta": meta_de_curso(nom)}
cursos = []
susp_ini = [] # cursos SUSPENDIDOS que ya iniciaron (se muestran marcados)
for c in _cursos_crudos():
if _curso_excluido(c, pmap): # MISMO filtro que 'Detalle por Curso'
# Los suspendidos SÍ se listan en "Ya iniciados" (con etiqueta),
# pero NO entran en cursos/pares/próximos (no afectan promedios).
d = datos(c)
if d["fch"] and d["suspendido"]:
susp_ini.append(d)
continue
d = datos(c)
if d["fch"]:
cursos.append(d)
# pares por (sede,prog) SOLO con importe>0
pares = _dd(list)
for d in cursos:
if d["importe"] > 0:
pares[(d["sede"], d["prog"])].append(d)
proximos = [d for d in cursos if HOY <= d["fch"] <= LIM and d["importe"] > 0]
proximos.sort(key=lambda x: x["fch"])
# Programas que YA INICIARON: desde ayer hasta hace 30 días (con importe>0).
INI_MIN = HOY - _dt.timedelta(days=30)
INI_MAX = HOY - _dt.timedelta(days=1)
iniciados_src = [d for d in cursos if INI_MIN <= d["fch"] <= INI_MAX and d["importe"] > 0]
# Suspendidos que ya iniciaron en el rango (se muestran marcados, sin exigir importe).
iniciados_src += [d for d in susp_ini if INI_MIN <= d["fch"] <= INI_MAX]
iniciados_src.sort(key=lambda x: x["fch"], reverse=True) # más recientes primero
def _armar_item(d, ya_inicio=False):
grp = [p for p in pares.get((d["sede"], d["prog"]), []) if p["ni"] != d["ni"]]
sum_imp = sum(p["importe"] for p in grp)
sum_res = sum(p["resultados"] for p in grp)
cpl_prom = round(sum_imp / sum_res, 2) if sum_res > 0 else 0.0
imp_prom = round(sum_imp / len(grp), 2) if grp else 0.0 # importe prom por curso par
res_prom = round(sum_res / len(grp), 0) if grp else 0.0 # resultados prom por curso par
leads_prom = round(sum(len(p["rec"]) for p in grp) / len(grp), 1) if grp else 0.0
proc_prom = round(sum(len(p["proc"]) for p in grp) / len(grp), 0) if grp else 0.0
leads = len(d["rec"]); proc = len(d["proc"])
cpl = round(d["importe"] / d["resultados"], 2) if d["resultados"] > 0 else 0.0
dias = (d["fch"] - HOY).days
dias_desde = (HOY - d["fch"]).days # días transcurridos desde el inicio (>=1)
# meta esperada por dias que faltan (si ya inició, se espera el 100%)
if ya_inicio:
pct_esp = 1.0
else:
pct_esp = 0.30 if dias > 20 else (0.60 if dias >= 8 else 0.85)
esperado = round(d["meta"] * pct_esp)
avance = (d["mats"] / esperado) if esperado > 0 else 1.0
if d["meta"] > 0 and d["mats"] >= d["meta"]:
nivel = "ok"
elif avance >= 1.0:
nivel = "ok"
elif avance >= 0.60:
nivel = "advertencia"
else:
nivel = "critico"
# senales CPL/leads (rojo)
cpl_rojo = cpl_prom > 0 and cpl > 0 and (cpl - cpl_prom) / cpl_prom > 0.75
leads_rojo = leads_prom > 0 and (leads - leads_prom) / leads_prom < -0.50
# conversion Meta -> base y atencion
conv = round(leads / d["resultados"] * 100, 1) if d["resultados"] > 0 else 0.0
aten = round(proc / leads * 100, 0) if leads > 0 else 0.0
# matriculados de ESTE curso vs OTROS (cruce por telefono de los recibidos)
este = otros = 0
for tel in d["rec"]:
ms = tel_mats.get(tel)
if not ms:
continue
if d["ni"] in ms:
este += 1
else:
otros += 1
def est_cpl():
if cpl_prom <= 0 or cpl <= 0: return "verde"
dif = (cpl - cpl_prom) / cpl_prom
return "rojo" if dif > 0.75 else ("amarillo" if dif > 0.25 else "verde")
def est_leads():
if leads_prom <= 0: return "verde"
dif = (leads - leads_prom) / leads_prom
return "rojo" if dif < -0.50 else ("amarillo" if dif < -0.20 else "verde")
est_mat = {"ok": "verde", "advertencia": "amarillo", "critico": "rojo"}[nivel]
def est_vs(actual, prom):
# Mas es mejor: verde >=prom, naranja 50-99% del prom, rojo <50%.
if prom <= 0: return "verde"
r = actual / prom
return "verde" if r >= 1.0 else ("amarillo" if r >= 0.50 else "rojo")
def est_gasto(actual, prom):
# Importe: menos es mejor. verde <=100%, naranja 101-150%, rojo >150%.
if prom <= 0: return "verde"
r = actual / prom
return "verde" if r <= 1.0 else ("amarillo" if r <= 1.5 else "rojo")
metricas = [
{"l": "Importe", "v": f"$ {d['importe']:.2f}", "sub": f"prom $ {imp_prom:.2f}", "est": est_gasto(d['importe'], imp_prom)},
{"l": "CPL", "v": f"$ {cpl:.2f}", "sub": f"prom $ {cpl_prom:.2f}", "est": est_cpl()},
{"l": "Resultados", "v": f"{d['resultados']}", "sub": f"prom {res_prom:.0f}",
"est": est_vs(d['resultados'], res_prom)},
{"l": "Leads recibidos", "v": f"{leads}", "sub": f"prom {leads_prom:.0f}", "est": est_leads()},
{"l": "Leads procesados", "v": f"{proc}", "sub": f"prom {proc_prom:.0f}",
"est": est_vs(proc, proc_prom)},
{"l": "Matrículas", "v": f"{d['mats']} / {d['meta']}",
"sub": (f"cerró con {d['mats']} de {d['meta']}" if ya_inicio else f"esperado ~{esperado} a {dias}d"),
"est": est_mat},
]
# analisis narrado: cada punto lleva su estado (verde/amarillo/rojo)
# para agruparlo luego en Óptimo / En observación / Crítico.
puntos = [] # lista de {"txt":..., "est":...}
# Matriculas
if ya_inicio:
puntos.append({"est": est_mat,
"txt": f"El curso inició hace {dias_desde} días y cerró con {d['mats']} de {d['meta']} matrículas."})
else:
puntos.append({"est": est_mat,
"txt": f"Faltan {dias} días para el inicio y lleva {d['mats']} de {d['meta']} matrículas "
f"(se esperaría ~{esperado} a esta altura)."})
# CPL
if cpl_prom > 0:
if cpl > cpl_prom:
puntos.append({"est": est_cpl(),
"txt": f"El CPL es $ {cpl:.2f} siendo el promedio $ {cpl_prom:.2f}: está costando más "
f"conseguir cada resultado que en cursos similares — la pauta se encareció."})
else:
puntos.append({"est": est_cpl(),
"txt": f"El CPL es $ {cpl:.2f}, por debajo del promedio $ {cpl_prom:.2f} — la pauta rinde bien."})
# Importe (punto propio)
if imp_prom > 0:
pct_imp = d["importe"] / imp_prom * 100
puntos.append({"est": est_gasto(d["importe"], imp_prom),
"txt": f"Llevas invertido $ {d['importe']:.2f} de un promedio de $ {imp_prom:.2f} por curso similar "
f"({pct_imp:.0f}% del gasto habitual)."})
# Resultados (punto propio)
if res_prom > 0:
pct_res = d["resultados"] / res_prom * 100
puntos.append({"est": est_vs(d["resultados"], res_prom),
"txt": f"Tienes {d['resultados']} resultados de un promedio de {res_prom:.0f} ({pct_res:.0f}%)."
+ (" Aún no alcanzas el promedio de resultados, tienes margen para invertir más."
if pct_res < 100 else " Ya superaste el promedio de resultados.")})
# Recibidos vs promedio
if leads_prom > 0:
dif_pct = (leads - leads_prom) / leads_prom * 100
p = f"Tenemos {leads} leads recibidos; antes recibíamos en promedio {leads_prom:.0f}."
if leads < leads_prom:
p += (f" Es {abs(dif_pct):.0f}% menos de lo habitual: está entrando poca gente a la base, "
"hay que reforzar la captación (pauta/anuncio).")
else:
p += " Estamos por encima del promedio de captación."
puntos.append({"est": est_leads(), "txt": p})
# Procesados vs promedio
if proc_prom > 0:
p = f"De esos, {proc} pasaron a un asesor (procesados); antes en promedio {proc_prom:.0f}."
if proc < proc_prom:
dp = (proc - proc_prom) / proc_prom * 100
p += (f" Es {abs(dp):.0f}% menos: llegan menos interesados a la etapa de asesor, "
"consecuencia de los pocos recibidos.")
else:
p += " Buen paso a asesor frente al promedio."
puntos.append({"est": est_vs(proc, proc_prom), "txt": p})
# (el cruce por telefono se muestra aparte, ver "cruce" abajo)
# Aviso por baja conversion: muchos resultados de Meta pero pocos llegan
# al WhatsApp/base (conv < 30%). Indica fuga entre el anuncio y el chat.
conv_baja = ""
if d["resultados"] >= 20 and conv < 30:
conv_baja = (f"Atención: de {d['resultados']} resultados solo {leads} llegaron a nuestra base "
f"({conv:.0f}%). Se está perdiendo mucha conversión entre resultados y recibidos: "
f"de muchos resultados, pocos llegan a nuestra base de datos. Si bien vamos bien en "
f"CPL e importe, tenemos pocas adiciones a nuestra base. ")
# recomendacion
if ya_inicio:
# Curso ya iniciado: la lectura es de cierre/aprendizaje, no de acción urgente.
if nivel == "critico":
reco = (f"El curso cerró con {d['mats']} de {d['meta']} matrículas, por debajo de la meta. "
+ conv_baja
+ ("El CPL quedó alto: revisar el anuncio para próximos cursos similares. " if cpl_rojo else "")
+ ("Entró poca gente a la base: reforzar captación en el siguiente. " if leads_rojo else "")
+ "Usar este resultado para ajustar la pauta del próximo inicio.")
elif nivel == "advertencia":
reco = (conv_baja + "Cerró cerca de la meta; hay margen de mejora en captación para el próximo curso.")
else:
reco = (conv_baja + "Cerró en buen nivel de matrículas.") if conv_baja else "Cerró en buen nivel de matrículas."
elif nivel == "critico":
reco = (f"Con {d['mats']} matrículas y {dias} días para el inicio va por debajo del ritmo esperado. "
+ conv_baja
+ ("Revisar la segmentación/mensaje del anuncio o aumentar inversión. " if leads_rojo else "")
+ ("El CPL está alto, revisar el anuncio. " if cpl_rojo else "")
+ "Priorizar la captación y el seguimiento de los leads pendientes.")
elif nivel == "advertencia":
reco = (conv_baja
+ "Va encaminado pero necesita un empujón: reforzar seguimiento de leads y vigilar el CPL.")
else:
reco = (conv_baja + "Sin acciones urgentes, va a buen ritmo.") if conv_baja else "Sin acciones urgentes, va a buen ritmo."
señal = ""
if cpl_rojo: señal += "CPL alto"
if leads_rojo: señal = (señal + " · " if señal else "") + "pocos leads"
# Agrupa los PUNTOS de analisis segun su estado (verde/amarillo/rojo).
g_opt = [x["txt"] for x in puntos if x["est"] == "verde"]
g_med = [x["txt"] for x in puntos if x["est"] == "amarillo"]
g_mal = [x["txt"] for x in puntos if x["est"] == "rojo"]
grupos = [
{"titulo": "Óptimo", "color": "verde", "items": g_opt},
{"titulo": "En observación", "color": "amarillo", "items": g_med},
{"titulo": "Crítico", "color": "rojo", "items": g_mal},
]
# Cruce por telefono, separado del resto.
cruce = f"{este} se matricularon a ESTE curso" + (f" y {otros} en OTROS programas." if otros else ".")
return {
"id": d["ni"], "curso": d["nom"], "dias_inicio": dias, "dias_desde": dias_desde,
"ya_inicio": ya_inicio, "suspendido": bool(d.get("suspendido")), "nivel": nivel,
"sede": d["sede"], "programa": d["prog"],
"resumen": señal or ("Cerca de la meta" if nivel != "critico" else "Requiere atención"),
"metricas": metricas, "puntos": puntos, "reco": reco,
"grupos": grupos, "cruce": cruce,
}
items = [_armar_item(d, ya_inicio=False) for d in proximos]
iniciados = [_armar_item(d, ya_inicio=True) for d in iniciados_src]
return {"items": items, "iniciados": iniciados}
# NO cachear si sale vacio por data aun no lista (evita quedar pegado en []).
# Si hay cursos cargados y aun asi no hay proximos, se cachea normal.
res = cache_get_or_set("alertas_dash", ("GLOBAL",), _load)
if not res.get("items") and _cursos_crudos():
from cache_manager import cache_invalidate
cache_invalidate("alertas_dash")
res = _load()
return res
def programas_ocultos():
"""Programas con contar=NO en basebi_programacion (ocultos de la matriz).
Cada item: {num_indice, dsc_det_programa, fch_inicio, pauta}."""
pm = _pauta_cruda()
ocultos_ni = {ni for ni, info in pm.items() if str(info.get("contar") or "").upper() == "NO"}
prog_de_ni = {}
fch_de_ni = {}
for c in _cursos_crudos():
ni = _norm_ni(c.get("num_indice"))
prog_de_ni[ni] = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()
f = L._to_date(c.get("fch_inicio"))
fch_de_ni[ni] = f.strftime("%d/%m/%Y") if f else ""
filas = []
for ni in ocultos_ni:
filas.append({"num_indice": ni,
"dsc_det_programa": prog_de_ni.get(ni, "(sin nombre)"),
"fch_inicio": fch_de_ni.get(ni, ""),
"pauta": (pm.get(ni) or {}).get("pauta") or ""})
filas.sort(key=lambda x: x["dsc_det_programa"])
return {"filas": filas}
def encender_programas(num_indices):
"""Pone contar=SI a los num_indice dados en basebi_programacion (los reactiva)."""
from cache_manager import cache_invalidate
dm = get_dm()
for ni in num_indices:
dm.guardar_pauta_basebi(ni, "", "SI") # pauta vacia -> no toca pauta, solo contar
for pref in ("pauta_raw", "leads_dash", "otros_general", "leads_filtros"):
cache_invalidate(pref)
return {"encendidos": len(num_indices)}
def programas_disponibles():
"""Lista de dsc_det_programa de cursos con fch_inicio >= 2026-01-01 (para el
desplegable de Programa en la Leyenda). Cada item: {num_indice, dsc_det_programa}."""
from datetime import date
corte = date(2026, 1, 1)
vistos = {}
for c in _cursos_crudos():
f = L._to_date(c.get("fch_inicio"))
if not f or f < corte:
continue
ni = _norm_ni(c.get("num_indice"))
nombre = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()
if not nombre:
continue
finicio = f.strftime("%d/%m/%Y")
label = f"{nombre} · {finicio}"
vistos[ni] = {"num_indice": ni, "dsc_det_programa": nombre,
"fch_inicio": finicio, "label": label}
filas = sorted(vistos.values(), key=lambda x: x["label"])
return {"programas": filas}
# ── LEYENDA (campanias de Supabase): listar / agregar / borrar ──
def campanias_listar():
return get_dm().listar_campanias()
def campania_agregar(datos):
res = get_dm().agregar_campania(datos)
_invalidar_por_campanias() # la leyenda nueva se refleja de inmediato en todo el dashboard
return res
def campania_editar(cid, datos):
res = get_dm().editar_campania(cid, datos)
_invalidar_por_campanias() # el cambio se refleja de inmediato en todo el dashboard
return res
def campania_borrar(cid):
res = get_dm().borrar_campania(cid)
_invalidar_por_campanias()
return res
def leyenda_anuncios():
"""Tabla: todos los conjuntos del Sheet con su pauta (si tiene) y el programa
(dsc_det_programa) del num_indice que usa esa pauta. Cacheado (usa insumos ya
en memoria) para que abrir la Leyenda sea rapido."""
def _load():
conj_pauta = _conjunto_pauta() # cacheado
pauta_de_conj = {}
for pa, cjs in conj_pauta.items():
for c in cjs:
pauta_de_conj[" ".join(str(c).split())] = str(pa).strip()
# pauta -> [num_indices] usando TODAS las vinculaciones (basebi + programa_pautas)
ni_de_pauta = {}
for ni, pautas in _pautas_de_indice().items():
for p in pautas:
ni_de_pauta.setdefault(str(p).strip(), []).append(str(ni).strip())
prog_de_ni = {}
for c in _cursos_crudos(): # cacheado
prog_de_ni[_norm_ni(c.get("num_indice"))] = str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()
def _programa(pauta):
if not pauta:
return ""
nis = ni_de_pauta.get(str(pauta).strip(), [])
nombres = [prog_de_ni.get(n, "") for n in nis if prog_de_ni.get(n)]
return " | ".join(sorted(set(nombres)))
filas = []
vistos = set()
for f in _meta_filas(): # cacheado (CSV de Meta)
co = " ".join(str(f.get("conjunto")).split())
if not co or co in vistos:
continue
vistos.add(co)
pa = pauta_de_conj.get(co, "")
filas.append({"conjunto": co, "pauta": pa, "programa": _programa(pa)})
filas.sort(key=lambda x: x["conjunto"])
return {"filas": filas}
return cache_get_or_set("leyenda_anuncios", ("GLOBAL",), _load)
def guardar_leyenda(cambios):
"""cambios = [{conjunto, pauta, num_indice?}].
- conjunto -> pauta: upsert en conjunto_pauta.
- si viene num_indice: conecta ese num_indice a la pauta en basebi_programacion.
Invalida solo cachés dependientes de pauta/conjunto (no la cartera)."""
from cache_manager import cache_invalidate
dm = get_dm()
for c in cambios:
conjunto = c.get("conjunto")
pauta = (c.get("pauta") or "").strip()
ni = c.get("num_indice")
if conjunto:
dm.upsert_conjunto_pauta(conjunto, pauta)
# si eligió un programa (num_indice), AGREGAR el vínculo num_indice<->pauta
# en programa_pautas (muchos-a-muchos, se SUMA, no reemplaza).
if ni and pauta:
dm.agregar_programa_pauta(str(ni).strip(), pauta)
# Invalida TODO lo que depende de pauta/conjunto (incluye ROAS, Programa, etc.)
for pref in ("conjunto_pauta", "pauta_raw", "programa_pautas", "pautas_de_indice",
"meta_filas", "importe_pauta", "importe_pauta_periodo",
"resultados_pauta_periodo", "resultados_pauta",
"leads_dash", "otros_general", "matriz_always", "matriz_webform",
"matriz_asignados", "matriz_plantillas", "leyenda_anuncios",
"roas_dash", "programa_dash", "programa_opciones",
"leads_x_codigo", "leads_filtros", "vendedores_dash"):
cache_invalidate(pref)
return {"guardados": len(cambios)}
def uso_de_pauta(pauta, excluir_num_indice=None):
"""Verifica si una pauta ya esta en uso. Devuelve los cursos (dsc_det_programa)
y conjuntos que la usan (para la advertencia antes de guardar)."""
dm = get_dm()
nis = dm.num_indices_de_pauta(pauta)
excl = str(excluir_num_indice or "").strip()
nis = [n for n in nis if n != excl]
# dsc_det_programa de esos num_indice (desde cursos de SQL, ya cacheados)
cursos = _cursos_crudos()
detalle = []
for c in cursos:
ni = _norm_ni(c.get("num_indice"))
if ni in nis:
detalle.append({"num_indice": ni,
"dsc_det_programa": str(c.get("dsc_det_programa") or c.get("dsc_programa") or "").strip()})
conjuntos = _conjunto_pauta().get(str(pauta).strip(), [])
return {"en_uso": bool(nis) or bool(conjuntos),
"cursos": detalle, "num_indices": nis, "conjuntos": conjuntos}
def guardar_edicion_curso(num_indice, pauta, conjunto=None, contar=None):
"""Guarda pauta y contar (SI/NO) en basebi_programacion (upsert) y, si se eligió,
vincula el conjunto a esa pauta en conjunto_pauta. Invalida SOLO los cachés que
dependen de pauta/conjunto (NO la cartera, para que la recarga sea rápida)."""
from cache_manager import cache_invalidate
dm = get_dm()
res = dm.guardar_pauta_basebi(num_indice, pauta, contar)
if conjunto:
dm.guardar_conjunto_pauta(conjunto, pauta)
# invalidar solo lo afectado por pauta/conjunto (la cartera NO se re-baja)
for pref in ("pauta_raw", "pautas_de_indice", "conjunto_pauta", "meta_filas",
"importe_pauta", "importe_pauta_periodo", "resultados_pauta_periodo",
"leads_dash", "otros_general", "matriz_always", "matriz_plantillas",
"leads_filtros", "roas_dash"):
cache_invalidate(pref)
return res
def _tipo_de_curso(c):
"""Categoría de tipo de programa para el filtro (PROGRAMAS/PROVINCIA/SEMINARIOS/OTROS)."""
dscp = c.get("dsc_programa", "")
otros = L.clasificar_tipo_programa(dscp)
if otros == "OTROS":
return "OTROS"
up = str(dscp).upper()
cat = "TEAC" if ("AIRE ACONDICIONADO" in up or "TEAC" in up) else (
"TERC" if ("REFRIGERACION COMERCIAL" in up or "REFRIGERACIÓN COMERCIAL" in up or "TERC" in up) else "SEM")
sede = L.clasificar_sede(dscp)
return L.tipo_programa_cohorte(sede, cat)
# ── Listas para los filtros (años, tipos de programa) ───────────
def opciones_filtros():
def _load():
cursos = _cursos_crudos()
anos = sorted({L._to_date(c.get("fch_inicio")).year for c in cursos
if L._to_date(c.get("fch_inicio"))}, reverse=True)
# Grupos de programa fijos para el filtro PROGRAMA
tipos = ["TEAC", "TERC", "SEMINARIOS", "OTROS"]
return {"anos": anos or [2026], "tipos": tipos}
return cache_get_or_set("leads_filtros", ("GLOBAL",), _load)
# ── Alertas: valores de programa/sede no identificados en cartera_junta ──
def alertas():
return cache_get_or_set("leads_alertas", ("GLOBAL",),
lambda: get_dm().traer_no_identificados())
# ── Diagnóstico de cursos (para cuadrar ocupabilidad) ───────────
def debug_cursos(ano, mes):
cursos = _cursos_crudos()
periodo = [c for c in cursos if L._en_periodo(L._to_date(c.get("fch_inicio")), ano, mes, "TODOS")]
detalle = []
tot_insc = 0
tot_meta = 0
for c in periodo:
insc = int(c.get("Inscritos_Totales", 0) or 0)
meta = L.meta_curso(c.get("dsc_programa"))
tp = L.tipo_programa_curso(c.get("dsc_programa"))
tot_insc += insc
tot_meta += meta
detalle.append({
"num_indice": c.get("num_indice"),
"dsc_programa": c.get("dsc_programa"),
"fch_inicio": str(c.get("fch_inicio"))[:10],
"cod_estado": c.get("cod_estado"),
"inscritos": insc,
"tipo": tp,
"meta": meta,
})
return {
"num_cursos": len(periodo),
"suma_inscritos": tot_insc,
"suma_meta": tot_meta,
"ocupabilidad": round((tot_insc / tot_meta * 100), 2) if tot_meta else 0,
"cursos": detalle,
}
# ── Precarga / refresco ─────────────────────────────────────────
def _periodos_actual_anterior():
"""[(ano, mes), ...] = mes actual y mes anterior (segun fecha de hoy)."""
from datetime import date
hoy = date.today()
ano, mes = hoy.year, hoy.month
if mes == 1:
prev = (ano - 1, 12)
else:
prev = (ano, mes - 1)
return [(str(ano), str(mes)), (str(prev[0]), str(prev[1]))]
import os as _os
_LOG_CARGA = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "carga_tiempos.log")
def _log_carga(linea, reset=False):
"""Escribe una linea en carga_tiempos.log (lo lee la ventana de tiempos)."""
try:
with open(_LOG_CARGA, "w" if reset else "a", encoding="utf-8") as f:
f.write(linea + "\n")
except Exception:
pass
import threading as _thr
_TIEMPOS = {} # etiqueta -> segundos (para el resumen final)
_TIEMPOS_LOCK = _thr.Lock()
_HILOS_LISTOS = set() # {"leads","otros"} — cuando estan los 2, se imprime resumen
def _cron(etiqueta, base, fn, apartado=None):
"""Ejecuta fn() midiendo el tiempo. Imprime y ademas guarda en el log:
paso, base de datos y segundos. Se ve en la ventana del BACKEND y en la
ventana de TIEMPOS al iniciar, para saber que carga cada cosa.
Si 'apartado' se indica, acumula el tiempo para el resumen final."""
import time as _t
t0 = _t.perf_counter()
r = fn()
dt = _t.perf_counter() - t0
n = ""
try:
if isinstance(r, (list, dict)):
n = f"({len(r)} filas)"
except Exception:
pass
linea = f" [carga] {etiqueta:32} {base:14} {dt:6.2f}s {n}"
print(linea, flush=True)
_log_carga(linea)
with _TIEMPOS_LOCK:
_TIEMPOS[apartado or etiqueta] = _TIEMPOS.get(apartado or etiqueta, 0.0) + dt
return r
def _hilo_termino(nombre):
"""Marca un hilo como listo. Cuando ambos ("leads" y "otros") terminan,
imprime el resumen final por apartado en el log."""
with _TIEMPOS_LOCK:
_HILOS_LISTOS.add(nombre)
listos = {"leads", "otros"}.issubset(_HILOS_LISTOS)
tiempos = dict(_TIEMPOS)
if not listos:
return
_log_carga("")
_log_carga("=" * 60)
_log_carga(" RESUMEN — tiempo listo para abrir cada apartado")
_log_carga("=" * 60)
orden = ["Leads", "Vendedores", "Otros General", "ROAS"]
for ap in orden:
seg = tiempos.get(ap)
if seg is not None:
estado = "instantaneo al abrir" if seg < 1 else f"listo tras {seg:.1f}s"
_log_carga(f" {ap:16} {seg:6.2f}s -> {estado}")
_log_carga("-" * 60)
_log_carga(" Ya precargados: al hacer clic se muestran AL INSTANTE (cache).")
_log_carga("=" * 60)
def precargar_todo():
"""Precarga en segundo plano: insumos crudos + el dashboard del mes actual y
anterior (con filtros en TODOS), para que el usuario no espere al abrir/filtrar."""
try:
import time as _t
_log_carga("=" * 60, reset=True)
_log_carga(" CARGA DEL DASHBOARD LEADS — " +
_t.strftime("%d/%m/%Y %H:%M:%S"))
_log_carga("=" * 60)
print(">>> Precargando LEADS (hilo 1)...", flush=True)
_log_carga(">>> Hilo 1: LEADS (PostgreSQL + SQL Server)")
_cron("Leads (mensajes)", "PostgreSQL", _leads_crudos)
_cron("Cursos", "SQL Server", _cursos_crudos)
_cron("Matriculas", "SQL Server", _matriculas_crudas)
_cron("Leads asignados/respuesta", "PostgreSQL", _leads_asignados_crudos)
# Dashboard (Leads) del mes actual y anterior (lo mas usado). Otros General
# NO se precarga aqui (sus consultas son pesadas); se calcula al entrar.
p = _periodos_actual_anterior()
a0, m0 = p[0]
for (a, m) in p:
_cron(f"Apartado Leads {a}-{m}", "cache",
lambda a=a, m=m: leads_dashboard(a, m, "TODOS", "TODOS", "TODOS"), apartado="Leads")
# Vendedores y ROAS del periodo actual: se precargan para abrir al instante.
_cron("Apartado Vendedores", "cache",
lambda: vendedores_dashboard(a0, m0, "TODOS"), apartado="Vendedores")
_cron("Apartado ROAS", "cache",
lambda: roas_dashboard(a0, m0, "TODOS"), apartado="ROAS")
_cron("Apartado Alertas", "cache", alertas_dashboard, apartado="Alertas")
_marcar_actualizacion()
print(">>> Hilo 1 (Leads) LISTO.", flush=True)
_log_carga(">>> Hilo 1 (Leads) LISTO.")
_hilo_termino("leads")
except Exception as e:
print(f"[precarga] {e}", flush=True)
_log_carga(f"[precarga] ERROR: {e}")
_hilo_termino("leads")
def precargar_otros_general():
"""Precarga (aparte, mas pesada) las 4 matrices de Otros General del mes actual
y anterior. Se llama en su propio hilo para no bloquear."""
try:
print(">>> Precargando CARTERA + OTROS GENERAL (hilo 2)...", flush=True)
_log_carga(">>> Hilo 2: CARTERA (Supabase) + OTROS GENERAL")
_cron("Plantillas", "Google Sheet", _plantillas_crudas)
_cron("Cartera", "Supabase", _cartera_rows_cache) # la mas pesada
for (a, m) in _periodos_actual_anterior():
_cron(f"Otros General {a}-{m}", "cache",
lambda a=a, m=m: otros_general(a, m, "TODOS"), apartado="Otros General")
print(">>> Hilo 2 (Cartera/Otros) LISTO.", flush=True)
_log_carga(">>> Hilo 2 (Cartera/Otros) LISTO.")
_hilo_termino("otros")
except Exception as e:
print(f"[precarga-otros] {e}", flush=True)
_log_carga(f"[precarga-otros] ERROR: {e}")
_hilo_termino("otros")
# Hora de la última actualización (hora Perú)
_ULTIMA_ACTUALIZACION = None
def _marcar_actualizacion():
global _ULTIMA_ACTUALIZACION
from datetime import datetime, timezone, timedelta
peru = timezone(timedelta(hours=-5))
_ULTIMA_ACTUALIZACION = datetime.now(peru).strftime("%d/%m/%Y %H:%M")
def ultima_actualizacion():
return {"hora": _ULTIMA_ACTUALIZACION}
def refrescar_todo():
from cache_manager import set_forzar
from data_manager_v2 import invalidar_campanias_cache
invalidar_campanias_cache() # tambien las campañas (cache aparte)
set_forzar(True) # recalcula y sobrescribe (no lee cache viejo)
try:
precargar_todo() # leads (mes actual/anterior)
precargar_otros_general() # cartera + Otros General
finally:
set_forzar(False)
_marcar_actualizacion()