# backend/leads_logic.py """ Lógica del módulo LEADS. Traduce las medidas DAX del PBI a Python. Columnas calculadas replicadas: - Cantidad_Veces -> 1ª aparición de un teléfono = lead único - Ultima_Etiqueta / ESTADO/OBJECION -> limpieza de cached_label_list - Tipo_Programa / Sede -> clasificación por dsc_programa Medidas replicadas: - Leads_Totales_Pauta_Unico, Leads_Procesados_Pauta_Unicos, Leads_Procesados_Contactados_Unicos, % Procesados, % Contactados, Total_Matriculas, Cant_inscritos_Mes, Ocupabilidad, Cursos (Inicios / Suspendidos / Ya Iniciados), Matrículas por día, tabla Estado/Objeción. """ from datetime import datetime, date # Etiquetas de sistema que se eliminan para hallar el estado/objeción real (igual que el DAX) ETIQUETAS_SISTEMA = { "atención_humana", "negociación", "supervisor", "grupo_arequipa", "grupo_trujillo", "sin_respuesta", } META_POR_TIPO = { "PROGRAMAS TEAC": 22, "PROGRAMAS TERC": 22, "PROVINCIA TEAC": 18, "PROVINCIA TERC": 18, "SEMINARIOS": 15, } # ── Helpers de fecha ──────────────────────────────────────────── def _to_date(v): if v is None: return None if isinstance(v, datetime): return v.date() if isinstance(v, date): return v s = str(v)[:10] for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y"): try: return datetime.strptime(s, fmt).date() except Exception: pass return None def _en_periodo(d, ano, mes, dia): """True si la fecha cae en el filtro (ano/mes/dia; 'TODOS' = sin filtrar).""" if d is None: return False try: if ano not in ("TODOS", None, "") and str(ano).upper() != "TODOS" and d.year != int(ano): return False if mes not in ("TODOS", None, "") and str(mes).upper() != "TODOS" and d.month != int(mes): return False if dia not in ("TODOS", None, "") and str(dia).upper() != "TODOS" and d.day != int(dia): return False except (ValueError, TypeError): return False return True # ── Clasificadores (igual que columnas calculadas DAX) ────────── def clasificar_tipo_programa(dsc_programa): up = str(dsc_programa or "").upper() otros = ["CERTIFICACIÓN", "CERTIFICACION", "CURSO A MEDIDA", "MASTERCLASS", "TALLER DE REFRIGERACIÓN DOM", "GESTIÓN DE VENTA", "GESTION DE VENTA"] for o in otros: if o.upper() in up: return "OTROS" return None # el tipo PROGRAMAS/PROVINCIA se arma con sede+programa abajo def clasificar_sede(dsc_programa): up = str(dsc_programa or "").upper() if "AREQUIPA" in up: return "AREQUIPA" if "TRUJILLO" in up: return "TRUJILLO" if "PIURA" in up: return "PIURA" return "LIMA" def sede_por_codigo_map(): """{codigo: SEDE} tomado del mapa de campañas (la sede que le corresponde a cada código, SIN reasignar por asesor). Usado por el filtro SEDE de leads, para que coincida con 'sede del código' (no con sede_act).""" from data_manager_v2 import CAMPANIAS out = {} for fila in CAMPANIAS: cod = str(fila[2]).strip() # índice 2 = codigo sede = str(fila[3]).strip().upper() # índice 3 = sede if cod: out[cod] = sede return out # ── Grupo de PROGRAMA para el filtro: TEAC / TERC / SEMINARIOS / OTROS ── def grupo_programa_curso(dsc_programa): """Reduce el tipo detallado a 4 grupos para el filtro PROGRAMA (cursos/matrículas). TEAC = PROGRAMAS TEAC + PROVINCIA TEAC TERC = PROGRAMAS TERC + PROVINCIA TERC SEMINARIOS = SEMINARIOS OTROS = OTROS + CARRERA""" tp = tipo_programa_curso(dsc_programa) if tp in ("PROGRAMAS TEAC", "PROVINCIA TEAC"): return "TEAC" if tp in ("PROGRAMAS TERC", "PROVINCIA TERC"): return "TERC" if tp == "SEMINARIOS": return "SEMINARIOS" return "OTROS" # OTROS + CARRERA def grupo_programa_lead(cargo): """Grupo de programa para un LEAD, según su 'cargo' (del mapa de campañas). TEAC → TEAC, TERC → TERC, resto (VRF/CO2/DIPLOMADO/etc.) → SEMINARIOS.""" c = str(cargo or "").strip().upper() if c == "TEAC": return "TEAC" if c == "TERC": return "TERC" return "SEMINARIOS" def grupo_tipo_cohorte(tp): """Reduce 'PROGRAMAS TEAC'/'PROVINCIA TEAC'/... a los 3 grupos de la tabla por Sede: TEAC = PROGRAMAS/PROVINCIA TEAC, TERC = PROGRAMAS/PROVINCIA TERC, resto = SEMINARIOS.""" t = str(tp or "").upper() if "TEAC" in t: return "TEAC" if "TERC" in t: return "TERC" return "SEMINARIOS" # Réplica del DAX Sede_Act: la sede del lead se decide por el ASESOR; # si el asesor no está en la lista, usa la sede de la campaña. _SEDE_POR_ASESOR = { "almendra peralta": "Arequipa", "juan carlos aguilar": "Piura", "diego lázaro": "Trujillo", "diego lazaro": "Trujillo", "verónica la rosa": "Lima", "veronica la rosa": "Lima", "dayana balabarca": "Lima", "milagros vargas": "Lima", "carmen montoya": "Lima", "diana chávez": "Lima", "diana chavez": "Lima", "copito rivera": "Lima", } def sede_act(asesor, sede_campania): return _SEDE_POR_ASESOR.get(str(asesor or "").strip().lower(), sede_campania) def tipo_programa_cohorte(sede, programa_cat): """Igual al DAX 'Tipo Programa': combina sede + (TEAC/TERC) -> categoría.""" s = str(sede or "").upper() p = str(programa_cat or "").upper() if s == "LIMA" and p == "TEAC": return "PROGRAMAS TEAC" if s == "LIMA" and p == "TERC": return "PROGRAMAS TERC" if s in ("TRUJILLO", "PIURA", "AREQUIPA") and p == "TEAC": return "PROVINCIA TEAC" if s in ("TRUJILLO", "PIURA", "AREQUIPA") and p == "TERC": return "PROVINCIA TERC" return "SEMINARIOS" def tipo_programa_curso(dsc_programa): """Réplica EXACTA del DAX Tipo_Programa (columna calculada de Fact_SQL_Base_Cursos). Usa CONTAINSSTRING en el mismo ORDEN que el PBI (el orden importa).""" p = str(dsc_programa or "") def has(s): # CONTAINSSTRING es sensible a may/min en DAX; comparamos tal cual return s in p # 1) OTROS if (has("CERTIFICACIÓN") or has("CURSO A MEDIDA") or has("MASTERCLASS") or has("TALLER DE REFRIGERACIÓN DOM") or has("GESTIÓN DE VENTA DE EQUIPOS DE AIRE ACONDICIONADO Y EQUIPOS DE REFRIGERACIÓN")): return "OTROS" if (has("CERTIFICACION FUNDAMENTOS DE CHILLER MODULAR INVERT- LG") or has("PIURA - CERTIFICACIÓN MIDEA: TECNOLOGIA INVERTER") or has("AREQUIPA - CERTIFICACIÓN MIDEA - TECNOLOGÍA INVERTER") or has("CERTIFICACIÓN: OPERACIÓN Y MANTENIMIENTO DE REFRIGERADORES MIDEA") or has("TRUJILLO - CERTIFICACIÓN MIDEA: TECNOLOGÍA INVERTER") or has("CERTIFICACIÓN MIDEA: AIRE ACOND INVERTER (INTROD., FUNC., INST. Y MANT.)")): return "OTROS" # 2) CARRERA if has("CARRERA TECNICA DE AIRE ACONDICIONADO Y REFRIGERACION"): return "CARRERA" # 3) PROVINCIA TEAC if (has("TRUJILLO - TECNICO ESPECIALISTA EN AIRE ACONDICIONADO") or has("PIURA - TECNICO ESPECIALISTA EN AIRE ACONDICIONADO") or has("AREQUIPA - TECNICO ESPECIALISTA EN AIRE ACONDICIONADO") or has("TRUJILLO - VIRTUAL TECNICO ESPECIALISTA EN AIRE ACONDICIONADO")): return "PROVINCIA TEAC" # 4) PROVINCIA TERC if (has("AREQUIPA - TECNICO ESPECIALISTA EN REFRIGERACION DOMÉSTICA Y COMERCIAL") or has("PIURA - TECNICO ESPECIALISTA EN REFRIGERACION DOMÉSTICA Y COMERCIAL") or has("TRUJILLO - TECNICO ESPECIALISTA EN REFRIGERACION DOMÉSTICA Y COMERCIAL")): return "PROVINCIA TERC" # 5) SEMINARIOS if (has("CO2") or has("CERTIF") or has("SEM:") or has("MASTERCLASS") or has("DISEÑO") or has("DIPLOMADO") or has("SEM.") or has("DUCTOS") or has("SEMINARIO") or has("SEMINARIOS")): return "SEMINARIOS" # 6) PROGRAMAS TEAC (Lima) if (has("MANTENIMIENTO EN AIRE ACONDICIONADO") or has("TECNICO ESPECIALISTA EN AIRE ACONDICIONADO") or has("TALLER - TECNICO ESPECIALISTA EN AIRE ACONDICIONADO") or has("CONTINUIDAD-TECNICO ESPECIALISTA EN AA (4M)") or has("VIRTUAL TECNICO ESPECIALISTA EN AIRE ACONDICIONADO")): return "PROGRAMAS TEAC" # 7) PROGRAMAS TERC (Lima) if (has("INSTALACION EN REFRIGERACION COMERCIAL") or has("MANTENIMIENTO EN REFRIGERACIÓN COMERCIAL") or has("CONTINUIDAD-TECNICO ESPECIALISTA EN REFRIGERACIÓN COMERCIAL (4M)") or has("TECNICO ESPECIALISTA EN REFRIGERACION COMERCIAL")): return "PROGRAMAS TERC" return "SEMINARIOS" def meta_curso(dsc_programa): """Meta por curso según Tipo_Programa (igual que el DAX Meta_Curso): PROGRAMAS TEAC/TERC=22, PROVINCIA TEAC/TERC=18, resto=15.""" tp = tipo_programa_curso(dsc_programa) if tp in ("PROGRAMAS TEAC", "PROGRAMAS TERC"): return 22 if tp in ("PROVINCIA TEAC", "PROVINCIA TERC"): return 18 return 15 # ── ESTADO/OBJECION (réplica EXACTA del DAX: cadena de SUBSTITUTE) ── # El DAX quita comas y espacios (pega todas las etiquetas), elimina las de # sistema, y luego reduce combinaciones concatenadas a un estado final. _SUST = [ ("atención_humana", ""), ("negociación", ""), ("supervisor", ""), ("grupo_arequipa", ""), ("grupo_trujillo", ""), ("sin_respuesta", ""), ("interesadovendido", "vendido"), ("próximo_inicio", "proxima_fecha"), ("inicio", ""), ("aprobación", ""), ("negociacion", ""), ("no_acepta", ""), ("pendiente", ""), ("revisando_informaciónsólo_consulta", "sólo_consulta"), ("contacto_iniciadovendido", "vendido"), ("revisando_informacióninteresado", "interesado"), ("revisando_informaciónvendido", "vendido"), ("sólo_consultainteresado", "interesado"), ("revisando_informaciónno_califica", "no_califica"), ("interesadopor_pagar", "por_pagar"), ("revisando_informaciónprecio_elevado", "precio_elevado"), ("proxima_fechavendido", "vendido"), ("sólo_consultavendido", "vendido"), ("vendidointeresado", "vendido"), ("revisando_informaciónproxima_fecha", "proxima_fecha"), ("contacto_iniciadosólo_consulta", "sólo_consulta"), ] def estado_objecion(etiquetas): """Réplica del DAX ESTADO/OBJECION (versión de 28 sustituciones del PBI).""" if etiquetas is None: return "no trabajado" s = str(etiquetas).replace(",", "").replace(" ", "") for buscar, reemplazar in _SUST: s = s.replace(buscar, reemplazar) s = s.strip() return s if s else "no trabajado" # ── Ultima_Etiqueta (réplica EXACTA del DAX, usada para "Total Contactados") ── _ULTIMA_NO_TRABAJADO = { "", "(en blanco)", "aprobación", "atencion humana", "atención humana", "grupo arequipa", "grupo trujillo", "importacion masiva", "inicio", "negociacion", "sin respuesta", "pendiente", } def ultima_etiqueta(etiquetas): if etiquetas is None or str(etiquetas).strip() == "": return "NO TRABAJADO" partes = str(etiquetas).split(", ") limpia = partes[-1].replace("_", " ").strip() if limpia in _ULTIMA_NO_TRABAJADO: return "NO TRABAJADO" return limpia if limpia else "NO TRABAJADO" # ════════════════════════════════════════════════════════════════ # PROCESAMIENTO DE LEADS (calcula Cantidad_Veces y agrupa) # ════════════════════════════════════════════════════════════════ def procesar_leads(filas_chatwoot): """Marca el lead único (1ª vez de cada teléfono) y arma estructura limpia. Ordenamos por fecha para que la 1ª aparición de cada teléfono sea la 'única'.""" # Ordenar por fecha ascendente (el query puede venir ordenado por id) filas_chatwoot = sorted(filas_chatwoot, key=lambda f: (_to_date(f.get("fecha_creada")) or date.min)) # Teléfonos de prueba a excluir (igual que el PBI: 924374783) EXCLUIR = {"924374783"} vistos = set() out = [] for f in filas_chatwoot: tel = str(f.get("telefono") or "").strip() if not tel or tel in EXCLUIR: continue es_unico = tel not in vistos vistos.add(tel) fecha = _to_date(f.get("fecha_creada")) asesor = (f.get("asesor") or "").strip() sede_campania = (f.get("sede") or "").strip() or "SIN SEDE" sede = sede_act(asesor, sede_campania) # réplica DAX Sede_Act (reasigna por asesor) cargo = (f.get("programa") or "").strip() # TEAC / TERC / VRF / etc. (del mapa campañas) out.append({ "telefono": tel, "asesor": asesor, "fecha": fecha, "estado": estado_objecion(f.get("etiquetas")), "ultima": ultima_etiqueta(f.get("etiquetas")), "sede": sede, "cargo": cargo, # TEAC / TERC / VRF / CO2 / etc. "tipo_programa": tipo_programa_cohorte(sede, cargo), # PROGRAMAS/PROVINCIA/SEMINARIOS "codigo": (f.get("codigo") or "").strip() or "-", # código de campaña "es_unico": es_unico, }) return out # ════════════════════════════════════════════════════════════════ # KPIs DEL MÓDULO LEADS # ════════════════════════════════════════════════════════════════ def kpis_leads(leads, cursos, matriculas, ano, mes, dia): # Leads en el periodo (solo únicos = Cantidad_Veces == 1) leads_periodo = [l for l in leads if l["es_unico"] and _en_periodo(l["fecha"], ano, mes, dia)] recibidos = len(leads_periodo) procesados = sum(1 for l in leads_periodo if l["asesor"]) contactados = sum(1 for l in leads_periodo if l["asesor"] and l["ultima"] != "NO TRABAJADO") pct_procesados = (procesados / recibidos) if recibidos > 0 else 0.0 pct_contactados = (contactados / procesados) if procesados > 0 else 0.0 # ── Cursos del periodo (por fecha de inicio) ── cursos_periodo = [c for c in cursos if _en_periodo(_to_date(c.get("fch_inicio")), ano, mes, dia)] cursos_programados = len(cursos_periodo) suspendidos = sum(1 for c in cursos_periodo if str(c.get("cod_estado", "")).strip().upper() == "SUS") hoy = date.today() iniciados = sum(1 for c in cursos_periodo if str(c.get("cod_estado", "")).strip().upper() != "SUS" and (_to_date(c.get("fch_inicio")) or hoy) <= hoy) # ── Total matriculados (= medida Matriculas_Totales del PBI) ── # Cuenta matrículas ALU/PRE cuya FECHA DE MATRÍCULA cae en el mes, # excluyendo 2 vendedoras. El Calendario del PBI va por fch_matricula. VENDEDORES_EXCLUIDOS = {"CALDERON S. LISSA GENA", "CRUZ G. FIORELLA MELISSA"} total_matriculados = sum( 1 for m in matriculas if str(m.get("estado_matricula", "")).strip().upper() in ("ALU", "PRE") and str(m.get("dsc_vendedor", "")).strip() not in VENDEDORES_EXCLUIDOS and _en_periodo(_to_date(m.get("fch_matricula")), ano, mes, dia) ) # Ocupabilidad (PBI): SUM(Inscritos) / SUM(Meta). # Excluye cursos suspendidos (SUS) y num_indice excluidos manualmente. # (Ver FUTUROS_CAMBIOS.md: mover esta lista a GitHub/Supabase) CURSOS_EXCLUIDOS = {"1154", "1121"} cursos_ocup = [c for c in cursos_periodo if str(c.get("cod_estado", "")).strip().upper() != "SUS" and str(c.get("num_indice", "")).strip() not in CURSOS_EXCLUIDOS] sum_meta = sum(meta_curso(c.get("dsc_programa")) for c in cursos_ocup) sum_inscritos_cursos = sum(int(c.get("Inscritos_Totales", 0) or 0) for c in cursos_ocup) # ── Matrículas en los cursos del mes ── # Matrículas ALU/PRE que: (1) se hicieron en el mes (fch_matricula) Y # (2) son de un curso que INICIA en el mes (dsc_promocion = dsc_det_programa # de los cursos del periodo). Réplica de Cant_inscritos_Mes del PBI. promos_periodo = {str(c.get("dsc_det_programa", "")).strip() for c in cursos_periodo if str(c.get("dsc_det_programa", "")).strip()} mats_mes = [m for m in matriculas if str(m.get("dsc_promocion", "")).strip() in promos_periodo and str(m.get("estado_matricula", "")).strip().upper() in ("ALU", "PRE") and _en_periodo(_to_date(m.get("fch_matricula")), ano, mes, dia)] matriculas_mes = len({str(m.get("num_matricula")) for m in mats_mes}) ocupabilidad = (sum_inscritos_cursos / sum_meta) if sum_meta > 0 else 0.0 return { "leads_recibidos": recibidos, "leads_procesados": procesados, "leads_contactados": contactados, "pct_procesados": round(pct_procesados * 100, 2), "pct_contactados": round(pct_contactados * 100, 2), "total_matriculados": total_matriculados, "matriculas_mes": matriculas_mes, "ocupabilidad": round(ocupabilidad * 100, 2), "cursos_programados": cursos_programados, "cursos_reprogramados": 0, # requiere SharePoint (fase 2) "cursos_suspendidos": suspendidos, "cursos_iniciados": iniciados, } # ── Tabla Estado/Objeción (agrupa por Ultima_Etiqueta + desglose por asesor) ── def tabla_estado_objecion(leads, ano, mes, dia): leads_periodo = [l for l in leads if l["es_unico"] and l["asesor"] and _en_periodo(l["fecha"], ano, mes, dia)] agg = {} # estado -> total por_asesor = {} # estado -> {asesor: cantidad} telefonos = {} # (estado, asesor) -> [telefonos] for l in leads_periodo: est = l["ultima"] or "NO TRABAJADO" ase = l["asesor"] or "SIN ASESOR" agg[est] = agg.get(est, 0) + 1 por_asesor.setdefault(est, {}) por_asesor[est][ase] = por_asesor[est].get(ase, 0) + 1 telefonos.setdefault((est, ase), []).append(l["telefono"]) filas = [] for estado in sorted(agg.keys(), key=lambda x: x.lower()): asesores = sorted(por_asesor[estado].items(), key=lambda x: -x[1]) filas.append({ "estado": estado, "cantidad": agg[estado], "asesores": [ {"asesor": a, "cantidad": c, "telefonos": sorted(telefonos.get((estado, a), []))} for a, c in asesores ], }) total = sum(agg.values()) return {"filas": filas, "total": total} # ── Matrículas por día (gráfico de línea) ── # Cuenta TODAS las matrículas del día (misma base que Total Matriculados): # estado ALU/PRE, excluyendo las 2 vendedoras, por día de fch_matricula. def matriculas_por_dia(matriculas, cursos, ano, mes, dia): VENDEDORES_EXCLUIDOS = {"CALDERON S. LISSA GENA", "CRUZ G. FIORELLA MELISSA"} conteo = {} # dia -> total por_tipo = {} # dia -> {tipo_programa: cantidad} for m in matriculas: if str(m.get("estado_matricula", "")).strip().upper() not in ("ALU", "PRE"): continue if str(m.get("dsc_vendedor", "")).strip() in VENDEDORES_EXCLUIDOS: continue d = _to_date(m.get("fch_matricula")) if not _en_periodo(d, ano, mes, dia): continue conteo[d.day] = conteo.get(d.day, 0) + 1 tp = tipo_programa_curso(m.get("dsc_programa")) # TEAC/TERC/PROVINCIA.../SEMINARIOS/OTROS por_tipo.setdefault(d.day, {}) por_tipo[d.day][tp] = por_tipo[d.day].get(tp, 0) + 1 def _detalle(day): items = sorted(por_tipo.get(day, {}).items(), key=lambda x: -x[1]) return [{"tipo": t, "cantidad": c} for t, c in items] import calendar if mes not in ("TODOS", None) and ano not in ("TODOS", None): try: ndias = calendar.monthrange(int(ano), int(mes))[1] return [{"dia": d, "cantidad": conteo.get(d, 0), "detalle": _detalle(d)} for d in range(1, ndias + 1)] except (ValueError, TypeError): pass return [{"dia": d, "cantidad": conteo[d], "detalle": _detalle(d)} for d in sorted(conteo.keys())] # ── Gráfico "Leads por Programa": serie diaria Totales vs Procesados ── def leads_por_dia(leads, ano, mes, dia): lp = [l for l in leads if l["es_unico"] and _en_periodo(l["fecha"], ano, mes, dia)] tot = {} # dia -> totales proc = {} # dia -> procesados (con asesor) for l in lp: d = l["fecha"].day tot[d] = tot.get(d, 0) + 1 if l["asesor"]: proc[d] = proc.get(d, 0) + 1 import calendar if mes not in ("TODOS", None) and ano not in ("TODOS", None): try: ndias = calendar.monthrange(int(ano), int(mes))[1] dias = range(1, ndias + 1) except (ValueError, TypeError): dias = sorted(set(list(tot.keys()) + list(proc.keys()))) else: dias = sorted(set(list(tot.keys()) + list(proc.keys()))) return [{"dia": d, "totales": tot.get(d, 0), "procesados": proc.get(d, 0)} for d in dias] # ── Tabla por Sede_Act -> Tipo_Programa con 4 medidas (cohorte de leads) ── def tabla_pauta(leads, matriculas, ano, mes, dia, importe_por_pauta_periodo=None, resultados_por_pauta_periodo=None, campanias_sede_cargo=None): importe_por_pauta_periodo = importe_por_pauta_periodo or {} resultados_por_pauta_periodo = resultados_por_pauta_periodo or {} campanias_sede_cargo = campanias_sede_cargo or {} # 1) Leads únicos del periodo con su teléfono, fecha, sede y tipo lp = [l for l in leads if l["es_unico"] and _en_periodo(l["fecha"], ano, mes, dia)] # mapa teléfono -> fecha del lead (para "matrícula posterior al lead") fecha_lead = {} for l in lp: fecha_lead.setdefault(l["telefono"], l["fecha"]) # 2) Matrículas que cruzan por teléfono y son POSTERIORES a la fecha del lead # (réplica de Matriculas_Cohorte_Lead e Inversion_x_Pauta_Cohorte) mat_por_tel = {} # telefono -> {"mats": set(num_matricula), "inv": float} for m in matriculas: tel = str(m.get("dsc_telefono_1") or "").strip() if tel not in fecha_lead: continue fm = _to_date(m.get("fch_matricula")) fl = fecha_lead[tel] if not fm or not fl or not (fm > fl): continue # INV_NETA_FINAL: si DOL, ×3.34 try: inv = float(m.get("INV_NETA", 0) or 0) except: inv = 0.0 if str(m.get("cod_moneda", "")).strip().upper() == "DOL": inv *= 3.34 e = mat_por_tel.setdefault(tel, {"mats": set(), "inv": 0.0}) e["mats"].add(str(m.get("num_matricula"))) e["inv"] += inv # 3) Agrupar por Sede -> Tipo_Programa -> Código def _nuevo(): return {"recibidos": 0, "procesados": 0, "mats": set(), "inv": 0.0} grupos = {} # sede -> tipo -> codigo -> métricas for l in lp: sede = l["sede"]; tp = grupo_tipo_cohorte(l["tipo_programa"]); cod = l.get("codigo", "-"); tel = l["telefono"] g = grupos.setdefault(sede, {}).setdefault(tp, {}).setdefault(cod, _nuevo()) g["recibidos"] += 1 if l["asesor"]: g["procesados"] += 1 if tel in mat_por_tel: g["mats"] |= mat_por_tel[tel]["mats"] g["inv"] += mat_por_tel[tel]["inv"] def _fila(d): return {"recibidos": d["recibidos"], "procesados": d["procesados"], "matriculas": len(d["mats"]), "inversion": round(d["inv"], 0)} # IMPORTE y RESULTADOS por PAUTA -> se asignan a su (sede, tipo) de CAMPANIAS, # una sola vez por pauta (no por sede del lead), para NO duplicar. # imp_st[(sede,tipo)][cod] = importe ; res_st[(sede,tipo)][cod] = resultados imp_st = {} res_st = {} for cod, imp_val in importe_por_pauta_periodo.items(): info = campanias_sede_cargo.get(str(cod).strip()) if not info: continue se = str(info.get("sede") or "").upper() tp = grupo_tipo_cohorte(tipo_programa_cohorte(se, info.get("cargo"))) imp_st.setdefault((se, tp), {})[str(cod).strip()] = float(imp_val or 0.0) for cod, res_val in resultados_por_pauta_periodo.items(): info = campanias_sede_cargo.get(str(cod).strip()) if not info: continue se = str(info.get("sede") or "").upper() tp = grupo_tipo_cohorte(tipo_programa_cohorte(se, info.get("cargo"))) res_st.setdefault((se, tp), {})[str(cod).strip()] = int(res_val or 0) filas = [] for sede in sorted(grupos.keys()): subfilas = [] s_rec = s_proc = 0; s_mats = set(); s_inv = 0.0; s_imp = 0.0; s_res = 0 for tp in sorted(grupos[sede].keys()): cods = grupos[sede][tp] imp_map = imp_st.get((sede.upper(), tp), {}) # importe por pauta de esta sede/tipo res_map = res_st.get((sede.upper(), tp), {}) # métricas del tipo (sumando sus códigos) t_rec = t_proc = 0; t_mats = set(); t_inv = 0.0; t_imp = 0.0; t_res = 0 codigos = [] for cod in sorted(cods.keys()): d = cods[cod] imp_cod = float(imp_map.get(str(cod).strip(), 0.0)) res_cod = int(res_map.get(str(cod).strip(), 0)) codigos.append({"codigo": cod, **_fila(d), "importe": round(imp_cod, 2), "resultados": res_cod}) t_rec += d["recibidos"]; t_proc += d["procesados"] t_mats |= d["mats"]; t_inv += d["inv"]; t_imp += imp_cod; t_res += res_cod subfilas.append({ "tipo": tp, "recibidos": t_rec, "procesados": t_proc, "matriculas": len(t_mats), "inversion": round(t_inv, 0), "importe": round(t_imp, 2), "resultados": t_res, "codigos": codigos, }) s_rec += t_rec; s_proc += t_proc; s_mats |= t_mats; s_inv += t_inv; s_imp += t_imp; s_res += t_res filas.append({ "sede": sede, "recibidos": s_rec, "procesados": s_proc, "matriculas": len(s_mats), "inversion": round(s_inv, 0), "importe": round(s_imp, 2), "resultados": s_res, "subfilas": subfilas, }) # Totales t_rec = sum(f["recibidos"] for f in filas) t_proc = sum(f["procesados"] for f in filas) t_inv = round(sum(f["inversion"] for f in filas), 0) # Total de importe/resultados: suma de TODAS las pautas que estan en campanias # (una vez cada una), aunque no tengan fila con leads. Asi el total no se pierde. t_imp = round(sum(float(v or 0.0) for cod, v in importe_por_pauta_periodo.items() if str(cod).strip() in campanias_sede_cargo), 2) t_res = sum(int(v or 0) for cod, v in resultados_por_pauta_periodo.items() if str(cod).strip() in campanias_sede_cargo) # matrículas total: unión global all_mats = set() for tel, e in mat_por_tel.items(): all_mats |= e["mats"] return {"filas": filas, "total": {"recibidos": t_rec, "procesados": t_proc, "matriculas": len(all_mats), "inversion": t_inv, "importe": t_imp, "resultados": t_res}} # ── Personalizado: replica la columna Fact_SQL_Base_Cursos[Personalizado] (M) ── # Base = dsc_det_programa, limpiando saltos de línea, luego cadena de reemplazos. _PERS_REEMPLAZOS = [ ("TECNICO ESPECIALISTA EN AIRE ACONDICIONADO", "TEAC"), ("TECNICO ESPECIALISTA EN REFRIGERACION COMERCIAL", "TERC"), ("TECNICO ESPECIALISTA EN REFRIGERACION DOMÉSTICA Y COMERCIAL", "TERC"), ("CARRERA TECNICA DE AIRE ACONDICIONADO Y REFRIGERACION ", "CARRERA"), ("VIRTUAL SEMINARIO SUPERVISION DE OBRAS EN AIRE ACONDICIONADO", "SUPERVISION DE OBRAS"), ("SEMINARIO APLICACIÓN DE VARIADORES DE FRECUENCIA EN SISTEMAS HVAC", "VARIADORES DE FRECUENCIA"), ("VIRTUAL - SISTEMAS DE REFRIGERACION CON CO2 - FASE SUBCRITICA Y TRANSCRITICA", "REFRIGERACION CON CO2"), ("VIRTUAL SEMINARIO DISEÑO DE CAMARAS DE REFRIGERACION CON SISTEMAS CON FREON", "DISEÑO DE CAMARAS"), ("VIRTUAL SEM. DETERMINACIÓN DE CAPACIDAD DE EQUIPOS DE AIRE ACONDICIONADO- CARGAS TÉRMICAS -(A/A)", "CARGAS TERMICAS"), ("VIRTUAL SEMINARIO DIBUJO TECNICO Y DISEÑO ASISTIDO POR COMPUTADORA (CAD) PARA HVAC -(A/A)", "DIBUJO TECNICO (CAD)"), ("VIRTUAL SEMINARIO: SISTEMAS DE REFRIGERACIÓN INDUSTRIAL POR AMONIACO (NH3)", "AMONIACO (NH3)"), ("SEMINARIO VIRTUAL METRADO , COSTEO Y PRESUPUESTOS DE HVAC (AA)", "METRADO, COSTEO Y PRESUPUESTOS"), ("CARRERA TECNICA LIMA GESTION DE EMPRESAS", "GESTION DE EMPRESAS"), ("VIRTUAL SEMINARIO DISEÑO DE CHILLERS PARA PROCESOS INDUSTRIALES DE REFRIGERACIÓN", "DISEÑO DE CHILLERS"), ("VIRTUAL SEMINARIO PRACTICO VENTILACION DE SOTANO Y PREZURIZACION DE ESCALERA -(A/A)", "VENTILACION DE SOTANO"), ("SEMINARIO USO DE SOFTWARE EN CÁLCULOS DE AIRE ACONDICIONADO", "SOFTWARE EN CALCULOS DE A/A"), ("DISEÑO DE SISTEMAS DE AIRE ACONDICIONADO", "DISEÑO DE SISTEMAS DE A/A"), ("SEMINARIO INSTALACIÓN DE SISTEMAS DE A/A DE VOLUMEN VARIABLE VRV/VRF - SAB", "VOLUMEN VARIABLE VRF"), # Seminarios VRF por sede y CO2 → nombre corto (la frecuencia se conserva al final) ("PIURA - SEMINARIO INSTALACIÓN DE SISTEMAS DE A/A DE VOLUMEN VARIABLE - VRF", "PIURA - VRF"), ("AREQUIPA - SEMINARIO INSTALACIÓN DE SISTEMAS DE A/A DE VOLUMEN VARIABLE - VRF", "AREQUIPA - VRF"), ("TRUJILLO - SEMINARIO INSTALACIÓN DE SISTEMAS DE A/A DE VOLUMEN VARIABLE - VRF", "TRUJILLO - VRF"), ("VIRTUAL - DISEÑO Y OPERACIÓN DE SISTEMAS DE REFRIGERACIÓN CON CO2 EN FASE SUBCRITICA Y TRANSCRITICA", "CO2"), # Reemplazos sobre Nombre_Programa_Detalle (bloque ductos y otros) ("VIRTUAL - DIMENSIONAMIENTO DE", ""), ("METÁLICOS EN AIRE ACONDICIONADO Y VENTILACIÓN", ""), ("MASTERCLASS: PROGRAMACIÓN DE CONTROLADORES PARA REFRIGERACIÓN", "PROGRAMACION DE CONTROLADORES"), ("DIPLOMADO INTERNACIONAL DE AIRE ACONDICIONADO", "DIPLOMADO DE A/A"), # Reemplazos finales (sedes/carreras/masterclass) ("AREQUIPA - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - TAR", "AREQUIPA - CARRERA - TARDE"), ("PIURA - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - TAR", "PIURA - CARRERA - TARDE"), ("TRUJILLO - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - TAR", "TRUJILLO - CARRERA - TARDE"), ("AREQUIPA - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - MAN", "AREQUIPA - CARRERA - MAÑANA"), ("PIURA - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - MAN", "PIURA - CARRERA - MAÑANA"), ("TRUJILLO - CARRERA TÉCNICA DE AIRE ACONDICIONADO Y REFRIGERACIÓN - MAN", "TRUJILLO - CARRERA - MAÑANA"), ("GESTIÓN DE VENTA DE EQUIPOS DE AIRE ACONDICIONADO Y EQUIPOS DE REFRIGERACIÓN - NOC", "GESTIÓN DE VENTA - NOC"), ("MASTERCLASS SELECCIÓN E INSTALACIÓN DE TARJETAS ELECTRÓNICAS UNIVERSALES EN EQUIPOS DE A/A .INVERTER - MAN", "MASTERCLASS TARJETAS UNIVERSALES - MAN"), ("MASTERCLASS: SISTEMAS DE A/A CON VRF - SAB", "MASTERCLASS: A/A CON VRF - SAB"), ("GESTIÓN DE VENTA DE EQUIPOS DE AIRE ACONDICIONADO Y EQUIPOS DE REFRIGERACIÓN - TAR", "GESTIÓN DE VENTA - TAR"), ] def personalizado_curso(dsc_programa, cod_frecuencia=""): """Replica Fact_SQL_Base_Cursos[Personalizado] del PBI: base = dsc_programa & ' - ' & cod_frecuencia (NO dsc_det_programa).""" s = f"{str(dsc_programa or '')} - {str(cod_frecuencia or '')}" # Text.Clean + reemplazar saltos de línea por espacio, luego Trim s = s.replace("\r", " ").replace("\n", " ") s = "".join(ch for ch in s if ch >= " " or ch == " ") # Text.Clean (quita control) s = s.strip() for buscar, reemplazo in _PERS_REEMPLAZOS: s = s.replace(buscar, reemplazo) # colapsar espacios múltiples que pudieran quedar de reemplazos vacíos s = " ".join(s.split()) return s # ── Matriz por Curso (num_indice): Personalizado + métricas de leads por pauta ── # NOTA: las métricas de leads cruzan por el CÓDIGO DE PAUTA del curso, que viene de # SharePoint (no disponible aún). Por eso salen en 0 hasta conectar esa fuente. # Ver FUTUROS_CAMBIOS.md punto 5. Lo que sí se calcula: Personalizado y Fecha Inicio. def matriz_cursos(cursos, leads, ano, mes, dia, pauta_map=None, campanias_map=None, cartera_all=None, cartera_asig=None, mat_por_indice=None, cartera_tels=None, cartera_origen=None, importe_por_pauta=None, importe_por_pauta_periodo=None, cartera_canal=None, conjunto_por_pauta=None, pautas_por_indice=None): pauta_map = pauta_map or {} pautas_por_indice = pautas_por_indice or {} # {num_indice: [pautas]} (muchos-a-muchos) campanias_map = campanias_map or {} cartera_all = cartera_all or {} cartera_asig = cartera_asig or {} mat_por_indice = mat_por_indice or {} cartera_tels = cartera_tels or set() cartera_origen = cartera_origen or {} importe_por_pauta = importe_por_pauta or {} importe_por_pauta_periodo = importe_por_pauta_periodo or {} conjunto_por_pauta = conjunto_por_pauta or {} # {pauta: [conjuntos de anuncios]} # cartera_canal = {"PAUTA": {"all","asig","tels"}, "OTROS": {...}} para el desglose por canal cartera_canal = cartera_canal or {} CURSOS_EXCLUIDOS = {"1121", "1154"} def _excluir(c): ni = str(c.get("num_indice", "")) if ni in CURSOS_EXCLUIDOS: return True info = pauta_map.get(ni) if not info: return False # sin info en Supabase → no se excluye # Excluir SUSPENDIDO o contar = NO return info.get("contar") == "NO" or info.get("estado") == "SUSPENDIDO" cursos_periodo = [c for c in cursos if _en_periodo(_to_date(c.get("fch_inicio")), ano, mes, dia) and not _excluir(c)] # Leads únicos agrupados por codigo (= Pauta). Históricos y del mes (filtro fecha). leads_unicos = [l for l in leads if l.get("es_unico")] hist_por_cod = {} # codigo → set teléfonos hist_ases_cod = {} # codigo → set teléfonos con asesor mes_por_cod = {} # codigo → set teléfonos (en periodo filtrado) mes_ases_cod = {} # codigo → set teléfonos con asesor (en periodo) for l in leads_unicos: cod = l.get("codigo", "-") tel = l["telefono"] hist_por_cod.setdefault(cod, set()).add(tel) if l.get("asesor"): hist_ases_cod.setdefault(cod, set()).add(tel) if _en_periodo(l["fecha"], ano, mes, dia): mes_por_cod.setdefault(cod, set()).add(tel) if l.get("asesor"): mes_ases_cod.setdefault(cod, set()).add(tel) # Construir filas con sus medidas pre = [] for c in sorted(cursos_periodo, key=lambda x: (_to_date(x.get("fch_inicio")) or date.min)): nombre = personalizado_curso(c.get("dsc_programa"), c.get("cod_frecuencia")) fch = _to_date(c.get("fch_inicio")) ni = str(c.get("num_indice", "")).strip() # LISTA de pautas del num_indice (muchos-a-muchos). Si no hay en la tabla nueva, # usa la de basebi (compatibilidad). cods = pautas_por_indice.get(ni) if not cods: _info = pauta_map.get(ni) cods = [_info.get("pauta")] if _info and _info.get("pauta") else [] cods = [str(x).strip() for x in cods if x] # Leads = union de telefonos de TODAS las pautas (sin duplicar) def _union(mapa): s = set() for cd in cods: s |= mapa.get(cd, set()) return len(s) nuevos = _union(hist_por_cod) nuevos_a = _union(hist_ases_cod) nuevos_m = _union(mes_por_cod) nuevos_ma = _union(mes_ases_cod) pre.append({ "num_indice": ni, "personalizado": nombre, "fch": fch, "pauta": cods[0] if cods else None, # pauta principal (para mostrar) "pautas": cods, # todas las pautas (para sumar) "fecha_inicio": fch.strftime("%d/%m/%Y") if fch else "", "leads_nuevos": nuevos, "leads_nuevos_asesor": nuevos_a, "leads_nuevos_mes": nuevos_m, "leads_nuevos_mes_asesor": nuevos_ma, }) # Índice HISTÓRICO completo (TODOS los cursos, sin filtro de fecha) por # personalizado → lista de (fecha, leads_nuevos). Para el acumulado. hist_cursos = {} for c in cursos: if _excluir(c): continue nom = personalizado_curso(c.get("dsc_programa"), c.get("cod_frecuencia")) f = _to_date(c.get("fch_inicio")) info = pauta_map.get(str(c.get("num_indice", ""))) cd = info.get("pauta") if info else None nv = len(hist_por_cod.get(cd, ())) if cd else 0 hist_cursos.setdefault(nom, []).append((f, nv)) # Leads_Acumulados_Historico: suma de leads_nuevos de TODOS los cursos del # MISMO personalizado con fecha <= la del curso actual (histórico completo). filas = [] for r in pre: acum = sum(nv for (f, nv) in hist_cursos.get(r["personalizado"], []) if f and r["fch"] and f <= r["fch"]) # CARTERA TOTAL: teléfonos únicos de cartera_junta con la misma SEDE+PROGRAMA # del curso (según su pauta → campanias) cuya fecha_creada <= fch_inicio del curso. # cartera_total = todos; cartera_total_asig = solo con asesor asignado. cartera_total = 0 cartera_total_asig = 0 cod = r.get("pauta") cods = r.get("pautas") or ([cod] if cod else []) if cods and r["fch"]: # Cartera = union de telefonos de la sede+programa de TODAS las pautas del num_indice tels_all = {}; tels_asig = {} for cd in cods: sp = campanias_map.get(str(cd).upper()) if not sp: continue for tel, fx in cartera_all.get(sp, {}).items(): if fx and (tel not in tels_all or fx < tels_all[tel]): tels_all[tel] = fx for tel, fx in cartera_asig.get(sp, {}).items(): if fx and (tel not in tels_asig or fx < tels_asig[tel]): tels_asig[tel] = fx cartera_total = sum(1 for fx in tels_all.values() if fx <= r["fch"]) cartera_total_asig = sum(1 for fx in tels_asig.values() if fx <= r["fch"]) # MATRÍCULAS NO IDENTIFICADAS: matrículas de este curso (num_indice) cuyo teléfono # (tel_1, o tel_2 si el 1 está vacío) NO aparece en cartera_junta. Sin filtro de fecha. _ni = r["num_indice"] _ni = _ni[:-2] if _ni.endswith(".0") else _ni _mats = mat_por_indice.get(_ni, []) # lista de (telefono, fecha_matricula) _tels_no_iden = [ph for (ph, fm) in _mats if ph not in cartera_tels] matriculas_no_iden = len(_tels_no_iden) # MATRICULAS LEADS NUEVOS: matrícula hecha dentro de 45 días DESPUÉS del origen # del lead (registro es_origen=SI de ese teléfono en la cartera). mat_leads_nuevos = 0 mat_leads_antiguos = 0 _tels_nuevos = [] _tels_antiguos = [] for (ph, fm) in _mats: if not ph or not fm: continue forig = cartera_origen.get(ph) if forig is None: continue dias = (fm - forig).days if 0 <= dias <= 45: mat_leads_nuevos += 1 _tels_nuevos.append(ph) else: # >45 días, o matrícula anterior al origen del lead (dias<0) → Antiguo mat_leads_antiguos += 1 _tels_antiguos.append(ph) # IMPORTE PAUTA: gasto de Meta Ads sumando TODAS las pautas del num_indice. importe_pauta = round(sum(float(importe_por_pauta.get(str(cd), 0.0)) for cd in cods), 2) # IMPORTE PAUTA EN EL MES: igual, pero solo del periodo filtrado. importe_pauta_mes = round(sum(float(importe_por_pauta_periodo.get(str(cd), 0.0)) for cd in cods), 2) # ── SUBFILAS por CANAL (PAUTA / OTROS) ── mismo criterio que la fila padre # (SEDE+PROGRAMA vía campanias, misma fecha) + filtro de canal desde cartera. def _subfila(clave): cc = cartera_canal.get(clave, {}) c_all = cc.get("all", {}); c_asig = cc.get("asig", {}); c_tels = cc.get("tels", set()) ct = ct_asig = 0 l_rec = l_proc = 0 l_rec_mes = l_proc_mes = 0 if cod and r["fch"]: sp = campanias_map.get(str(cod).upper()) if sp: fa = c_all.get(sp, {}) fg = c_asig.get(sp, {}) # Cartera Total/Asig del canal: fecha_creada <= inicio del curso (igual que padre) ct = sum(1 for fx in fa.values() if fx and fx <= r["fch"]) ct_asig = sum(1 for fx in fg.values() if fx and fx <= r["fch"]) # L. Recibidos/Procesados del canal = misma cartera del canal (mismo corte) l_rec = ct l_proc = ct_asig # "del Mes" = fecha_creada en el periodo filtrado l_rec_mes = sum(1 for fx in fa.values() if _en_periodo(fx, ano, mes, dia)) l_proc_mes = sum(1 for fx in fg.values() if _en_periodo(fx, ano, mes, dia)) # Matrículas del curso, clasificadas por canal: # - NO identificadas (tel NO en cartera) -> todas a OTROS. # - identificadas -> al canal de su telefono (c_tels), regla 45 dias. mt_noiden = mt_nuevos = mt_antiguos = 0 for (ph, fm) in _mats: if not ph: continue en_cartera = ph in cartera_tels if not en_cartera: # no identificada: solo cuenta en la subfila OTROS if clave == "OTROS": mt_noiden += 1 continue # identificada: solo cuenta si su tel pertenece a ESTE canal if ph not in c_tels: continue forig = cartera_origen.get(ph) if fm and forig: d = (fm - forig).days if 0 <= d <= 45: mt_nuevos += 1 else: mt_antiguos += 1 return {"canal": clave, "cartera_total": ct, "cartera_total_asig": ct_asig, "leads_nuevos": l_rec, "leads_nuevos_asesor": l_proc, "leads_nuevos_mes": l_rec_mes, "leads_nuevos_mes_asesor": l_proc_mes, "matriculas_no_iden": mt_noiden, "mat_leads_nuevos": mt_nuevos, "mat_leads_antiguos": mt_antiguos} subfilas = [_subfila("PAUTA"), _subfila("WEB"), _subfila("OTROS")] if cartera_canal else [] filas.append({ "num_indice": r["num_indice"], "personalizado": r["personalizado"], "fecha_inicio": r["fecha_inicio"], "pauta": r.get("pauta"), "importe_pauta": importe_pauta, "importe_pauta_mes": importe_pauta_mes, "cartera_total": cartera_total, "cartera_total_asig": cartera_total_asig, "matriculas_no_iden": matriculas_no_iden, "matriculas_no_iden_tels": _tels_no_iden, "mat_leads_nuevos": mat_leads_nuevos, "mat_leads_antiguos": mat_leads_antiguos, "mat_leads_nuevos_tels": _tels_nuevos, "mat_leads_antiguos_tels": _tels_antiguos, "leads_acumulados": acum, "leads_nuevos": r["leads_nuevos"], "leads_nuevos_asesor": r["leads_nuevos_asesor"], "leads_nuevos_mes": r["leads_nuevos_mes"], "leads_nuevos_mes_asesor": r["leads_nuevos_mes_asesor"], "subfilas_canal": subfilas, "conjuntos": [cj for cd in cods for cj in conjunto_por_pauta.get(str(cd).strip(), [])], "contar": (pauta_map.get(str(r["num_indice"]).strip()) or {}).get("contar") or "SI", }) return {"filas": filas}