"""Dashboard interactivo para leads-optimizer — Streamlit.""" import streamlit as st import pandas as pd import json import subprocess import sys import os import glob from datetime import datetime, date import calendar sys.path.insert(0, os.path.dirname(__file__)) from airtable_client import AirtableClient from analyzer import analyze st.set_page_config( page_title="Leads Optimizer", layout="wide", initial_sidebar_state="expanded", ) # ── Helpers ─────────────────────────────────────────────────────────────────── URGENCIA_ICON = { "PAUSAR": "⛔", "SPRINT": "🚀", "ACELERAR": "📈", "FRENAR": "📉", "EN_RITMO": "✅", } URGENCIA_ORDER = {"PAUSAR": 0, "SPRINT": 1, "ACELERAR": 2, "FRENAR": 3, "EN_RITMO": 4} CRITICIDAD_ORDER = {"Crítico": 0, "Peligro": 1, "Mantener": 2} def _eur(v: float) -> str: return f"{v:.2f}€" def _pct(v: float) -> str: sign = "+" if v >= 0 else "" return f"{sign}{v * 100:.1f}%" def _ritmo_bar(ritmo: float) -> str: filled = int(abs(ritmo) * 10) bar = "█" * min(filled, 10) return ("🟢 +" if ritmo > 0 else "🔴 ") + bar if ritmo != 0 else "⚪ —" # ── Data loading ────────────────────────────────────────────────────────────── @st.cache_data(ttl=300, show_spinner="Cargando datos de Airtable...") def _load_gcm(mes: int, anio: int) -> list[dict]: """Lee GACampaignMes para el mes/año dado.""" at = AirtableClient() mes_num = str(mes) anio_str = str(anio) formula = f"AND({{Mes}}='{mes_num}',{{Año}}='{anio_str}')" records = at.gacampaignmes.all( formula=formula, fields=[ "CampaignID", "Campaign Name (from CampaignID)", "Mes", "Año", "Status", "PPL", "CPAMax", "CapTotalMes", "CosteMes", "ConvMes", "ConvLeadsLakeMesFinal", "MetricasDiarias", "Consejo", "Criticidad", "Log", ], ) campaigns_records = at.campaigns.all(fields=["CampaignID"]) at_id_to_gid = {r["id"]: str(r["fields"].get("CampaignID", "")).strip() for r in campaigns_records} result = [] for r in records: f = r["fields"] at_cids = f.get("CampaignID", []) gid = at_id_to_gid.get(at_cids[0], "") if at_cids else "" name_list = f.get("Campaign Name (from CampaignID)", []) name = name_list[0] if name_list else "Sin nombre" metricas_raw = f.get("MetricasDiarias") or "{}" try: metricas = json.loads(metricas_raw) except (json.JSONDecodeError, TypeError): metricas = {} result.append({ "airtable_id": r["id"], "google_campaign_id": gid, "nombre": name, "status": f.get("Status", "Pausada"), "ppl": float(f.get("PPL") or 0), "cpa_max": float(f.get("CPAMax") or 0), "cap": int(f.get("CapTotalMes") or 0), "coste_mes": float(f.get("CosteMes") or 0), "conv_mes": float(f.get("ConvMes") or 0), "leads_lake": int(f.get("ConvLeadsLakeMesFinal") or 0), "metricas": metricas, "consejo": f.get("Consejo", ""), "criticidad": f.get("Criticidad", "Mantener"), "log": f.get("Log", ""), }) return result @st.cache_data(ttl=300, show_spinner="Cargando cappings de cursos...") def _load_cursomes(mes: int, anio: int) -> list[dict]: at = AirtableClient() meses_es = { 1: "Enero", 2: "Febrero", 3: "Marzo", 4: "Abril", 5: "Mayo", 6: "Junio", 7: "Julio", 8: "Agosto", 9: "Septiembre", 10: "Octubre", 11: "Noviembre", 12: "Diciembre", } mes_nombre = meses_es[mes] formula = f"AND({{Mes}}='{mes_nombre}',{{Año}}='{anio}')" records = at.cursomes.all( formula=formula, fields=["CursoID", "Mes", "Año", "Caping Admitido"], ) cursos_records = at.cursos.all(fields=["CursoID", "Nombre"]) rec_to_cursoid = {r["id"]: str(int(r["fields"].get("CursoID", 0))) for r in cursos_records if r["fields"].get("CursoID")} rec_to_nombre = {r["id"]: r["fields"].get("Nombre", "") for r in cursos_records} result = [] for r in records: f = r["fields"] curso_rec_ids = f.get("CursoID", []) cursoid = rec_to_cursoid.get(curso_rec_ids[0], "") if curso_rec_ids else "" nombre = rec_to_nombre.get(curso_rec_ids[0], cursoid) if curso_rec_ids else cursoid result.append({ "airtable_id": r["id"], "cursoid": cursoid, "nombre": nombre, "mes": f.get("Mes", ""), "anio": f.get("Año", ""), "cap": int(f.get("Caping Admitido") or 0), }) return sorted(result, key=lambda x: x["nombre"]) def _compute_analysis(row: dict, dia_actual: int, dias_mes: int) -> dict: """Calcula urgencia/ritmo con los datos de Airtable (sin Google Ads API).""" leads = row["leads_lake"] or int(row["conv_mes"]) cap = row["cap"] ppl = row["ppl"] cpa_max = row["cpa_max"] gasto = row["coste_mes"] ratio_leads = leads / cap if cap > 0 else 0 ratio_mes = dia_actual / dias_mes ritmo = ratio_leads - ratio_mes cpa_actual = gasto / leads if leads > 0 else 0 revenue = leads * ppl margen = (revenue - gasto) / revenue if revenue > 0 else 0 leads_restantes = cap - leads dias_restantes = dias_mes - dia_actual if ratio_leads >= 1.0: urgencia = "PAUSAR" elif cap > 0 and ratio_leads < ratio_mes - 0.15 and dias_restantes <= 5: urgencia = "SPRINT" elif ritmo < -0.15: urgencia = "ACELERAR" elif ritmo > 0.15: urgencia = "FRENAR" else: urgencia = "EN_RITMO" return { "leads": leads, "cap": cap, "ratio_leads": ratio_leads, "ratio_mes": ratio_mes, "ritmo": ritmo, "urgencia": urgencia, "cpa_actual": cpa_actual, "rentable": cpa_actual <= cpa_max if cpa_actual > 0 else True, "margen": margen, "revenue": revenue, "leads_restantes": leads_restantes, "dias_restantes": dias_restantes, } # ── Sidebar ─────────────────────────────────────────────────────────────────── st.sidebar.title("Leads Optimizer") today = date.today() col_m, col_a = st.sidebar.columns(2) mes_sel = col_m.number_input("Mes", min_value=1, max_value=12, value=today.month) anio_sel = col_a.number_input("Año", min_value=2024, max_value=2030, value=today.year, step=1) if st.sidebar.button("🔄 Actualizar datos", use_container_width=True): st.cache_data.clear() st.sidebar.divider() # Filtros adicionales (se rellenan tras cargar datos) urgencia_filter = st.sidebar.multiselect( "Urgencia", ["PAUSAR", "SPRINT", "ACELERAR", "FRENAR", "EN_RITMO"], default=[], placeholder="Todas", ) solo_activas = st.sidebar.checkbox("Solo campañas activas", value=True) # ── Carga de datos ──────────────────────────────────────────────────────────── gcm_rows = _load_gcm(int(mes_sel), int(anio_sel)) dia_actual = today.day if (today.month == int(mes_sel) and today.year == int(anio_sel)) else calendar.monthrange(int(anio_sel), int(mes_sel))[1] dias_mes = calendar.monthrange(int(anio_sel), int(mes_sel))[1] # Enriquecer con análisis calculado for row in gcm_rows: row["analysis"] = _compute_analysis(row, dia_actual, dias_mes) # Aplicar filtros filtered = gcm_rows if solo_activas: filtered = [r for r in filtered if r["status"] == "Activa"] if urgencia_filter: filtered = [r for r in filtered if r["analysis"]["urgencia"] in urgencia_filter] # Ordenar por urgencia filtered.sort(key=lambda r: URGENCIA_ORDER.get(r["analysis"]["urgencia"], 9)) # ── Título ──────────────────────────────────────────────────────────────────── st.title(f"Leads Optimizer · {calendar.month_name[int(mes_sel)]} {int(anio_sel)}") st.caption(f"Día {dia_actual} de {dias_mes} · {len(filtered)} campañas") # ── KPIs globales ───────────────────────────────────────────────────────────── total_gasto = sum(r["coste_mes"] for r in filtered) total_leads = sum(r["analysis"]["leads"] for r in filtered) total_revenue = sum(r["analysis"]["revenue"] for r in filtered) total_margen = total_revenue - total_gasto margen_pct = total_margen / total_revenue if total_revenue > 0 else 0 k1, k2, k3, k4, k5 = st.columns(5) k1.metric("Gasto total", _eur(total_gasto)) k2.metric("Leads entregados", f"{total_leads:,}") k3.metric("Revenue estimado", _eur(total_revenue)) k4.metric("Margen total", _eur(total_margen)) k5.metric("Margen %", _pct(margen_pct)) st.divider() # ── Pestañas ────────────────────────────────────────────────────────────────── tab_resumen, tab_campanas, tab_historico, tab_ejecucion = st.tabs([ "📊 Resumen", "📋 Campañas", "📈 Histórico", "⚡ Ejecución", ]) # ════════════════════════════════════════════════════════════════════════════ # # Tab 1 — RESUMEN # # ════════════════════════════════════════════════════════════════════════════ # with tab_resumen: # ── Conteo por urgencia ─────────────────────────────────────────────────── urgencia_counts = {} for r in filtered: u = r["analysis"]["urgencia"] urgencia_counts[u] = urgencia_counts.get(u, 0) + 1 u_cols = st.columns(5) for i, (urg, icon) in enumerate(URGENCIA_ICON.items()): cnt = urgencia_counts.get(urg, 0) u_cols[i].metric(f"{icon} {urg}", cnt) st.divider() # ── Alertas activas ─────────────────────────────────────────────────────── alerts = [] for r in filtered: if r["log"]: alerts.append((r["nombre"], r["log"])) a = r["analysis"] if a["margen"] < -0.5 and a["revenue"] > 0: alerts.append((r["nombre"], f"Margen muy negativo: {_pct(a['margen'])}")) if alerts: with st.expander(f"⚠️ {len(alerts)} alertas activas", expanded=True): for nombre, msg in alerts: st.warning(f"**{nombre}** — {msg}") # ── Tabla de campañas ───────────────────────────────────────────────────── st.subheader("Estado de campañas") if not filtered: st.info("Sin campañas con los filtros actuales.") else: table_rows = [] for r in filtered: a = r["analysis"] icon = URGENCIA_ICON.get(a["urgencia"], "") table_rows.append({ "Urgencia": f"{icon} {a['urgencia']}", "Campaña": r["nombre"], "Leads": a["leads"], "Cap": a["cap"], "Ritmo": _pct(a["ritmo"]), "Gasto": _eur(r["coste_mes"]), "CPA act.": _eur(a["cpa_actual"]) if a["cpa_actual"] > 0 else "—", "CPA máx.": _eur(r["cpa_max"]), "Margen": _pct(a["margen"]) if a["revenue"] > 0 else "—", "Criticidad": r["criticidad"] or "—", "Consejo": (r["consejo"] or "")[:80] + ("…" if len(r["consejo"] or "") > 80 else ""), }) df = pd.DataFrame(table_rows) event = st.dataframe( df, use_container_width=True, hide_index=True, on_select="rerun", selection_mode="single-row", ) sel = event.selection.rows if sel: row = filtered[sel[0]] a = row["analysis"] st.divider() st.subheader(row["nombre"]) c1, c2, c3, c4 = st.columns(4) c1.metric("Leads / Cap", f"{a['leads']} / {a['cap']}") c2.metric("Ritmo", _pct(a["ritmo"])) c3.metric("CPA actual", _eur(a["cpa_actual"]) if a["cpa_actual"] > 0 else "—") c4.metric("Margen", _pct(a["margen"]) if a["revenue"] > 0 else "—") if row["consejo"]: st.info(f"💡 {row['consejo']}") if row["log"]: st.warning(f"⚠️ {row['log']}") link = f"https://ads.google.com/aw/campaigns?campaignId={row['google_campaign_id']}" st.markdown(f"[Abrir en Google Ads ↗]({link})") # ════════════════════════════════════════════════════════════════════════════ # # Tab 2 — CAMPAÑAS (editor de parámetros) # # ════════════════════════════════════════════════════════════════════════════ # with tab_campanas: st.subheader("Parámetros de campañas — GACampaignMes") st.caption("Edita PPL, CPA Máx y Cap directamente. Pulsa Guardar para escribir en Airtable.") if not gcm_rows: st.info("Sin datos para el mes seleccionado.") else: editor_data = pd.DataFrame([ { "Campaña": r["nombre"], "Status": r["status"], "PPL (€)": r["ppl"], "CPA Máx (€)": r["cpa_max"], "Cap mes": r["cap"], "Gasto": _eur(r["coste_mes"]), "Leads lake": r["leads_lake"], "_id": r["airtable_id"], } for r in gcm_rows ]) edited = st.data_editor( editor_data.drop(columns=["_id"]), use_container_width=True, hide_index=True, disabled=["Campaña", "Status", "Gasto", "Leads lake"], column_config={ "PPL (€)": st.column_config.NumberColumn(min_value=0.0, format="%.2f"), "CPA Máx (€)": st.column_config.NumberColumn(min_value=0.0, format="%.2f"), "Cap mes": st.column_config.NumberColumn(min_value=0, step=1), }, ) if st.button("💾 Guardar cambios en Airtable", type="primary"): at = AirtableClient() batch = [] for i, orig in enumerate(gcm_rows): new_ppl = float(edited.at[i, "PPL (€)"]) new_cpa = float(edited.at[i, "CPA Máx (€)"]) new_cap = int(edited.at[i, "Cap mes"]) changes = {} if new_ppl != orig["ppl"]: changes["PPL"] = new_ppl if new_cpa != orig["cpa_max"]: changes["CPAMax"] = new_cpa if new_cap != orig["cap"]: changes["CapTotalMes"] = new_cap if changes: batch.append({"id": orig["airtable_id"], "fields": changes}) if batch: for i in range(0, len(batch), 10): at.gacampaignmes.batch_update(batch[i:i+10]) st.success(f"✅ {len(batch)} registros actualizados.") st.cache_data.clear() else: st.info("Sin cambios detectados.") st.divider() st.subheader("Cappings por curso — CursoMes") st.caption("Ajusta el capping mensual admitido por curso.") cursomes_rows = _load_cursomes(int(mes_sel), int(anio_sel)) if not cursomes_rows: st.info("Sin registros de CursoMes para el período seleccionado.") else: cm_data = pd.DataFrame([ { "Curso ID": r["cursoid"], "Nombre": r["nombre"], "Cap admitido": r["cap"], "_id": r["airtable_id"], } for r in cursomes_rows ]) cm_edited = st.data_editor( cm_data.drop(columns=["_id"]), use_container_width=True, hide_index=True, disabled=["Curso ID", "Nombre"], column_config={ "Cap admitido": st.column_config.NumberColumn(min_value=0, step=1), }, ) if st.button("💾 Guardar cappings en Airtable", type="primary", key="save_cursomes"): at = AirtableClient() batch = [] for i, orig in enumerate(cursomes_rows): new_cap = int(cm_edited.at[i, "Cap admitido"]) if new_cap != orig["cap"]: batch.append({"id": orig["airtable_id"], "fields": {"Caping Admitido": new_cap}}) if batch: for j in range(0, len(batch), 10): at.cursomes.batch_update(batch[j:j+10]) st.success(f"✅ {len(batch)} cappings actualizados.") st.cache_data.clear() else: st.info("Sin cambios detectados.") # ════════════════════════════════════════════════════════════════════════════ # # Tab 3 — HISTÓRICO # # ════════════════════════════════════════════════════════════════════════════ # with tab_historico: st.subheader("Métricas diarias por campaña") camp_names = [r["nombre"] for r in filtered if r["metricas"]] if not camp_names: st.info("Sin métricas diarias disponibles para las campañas filtradas.") else: sel_camp = st.selectbox("Campaña", camp_names, key="hist_camp") camp_row = next((r for r in filtered if r["nombre"] == sel_camp), None) if camp_row and camp_row["metricas"]: metricas = camp_row["metricas"] # MetricasDiarias puede ser dict {fecha: {coste, ingresos, margen, leads, ...}} # o list [{fecha, coste, ...}] if isinstance(metricas, dict): daily = [{"fecha": k, **v} for k, v in sorted(metricas.items())] else: daily = sorted(metricas, key=lambda x: x.get("fecha", "")) if daily: df_daily = pd.DataFrame(daily) st.caption(f"Columnas raw: {list(df_daily.columns)}") # Normalizar nombres de columnas (pueden variar) col_map = {} for c in df_daily.columns: cl = c.lower() if c in col_map.values(): continue if ("fecha" in cl or "date" in cl) and "Fecha" not in col_map.values(): col_map[c] = "Fecha" elif ("coste" in cl or "cost" in cl) and "Gasto" not in col_map.values(): col_map[c] = "Gasto" elif ("ingres" in cl or "revenue" in cl) and "Revenue" not in col_map.values(): col_map[c] = "Revenue" elif ("margen" in cl or "margin" in cl) and "Margen" not in col_map.values(): col_map[c] = "Margen" elif "lead" in cl and "Leads" not in col_map.values(): col_map[c] = "Leads" df_daily = df_daily.rename(columns=col_map) # KPIs del período d1, d2, d3, d4 = st.columns(4) if "Gasto" in df_daily: d1.metric("Gasto total", _eur(df_daily["Gasto"].sum())) if "Revenue" in df_daily: d2.metric("Revenue total", _eur(df_daily["Revenue"].sum())) if "Margen" in df_daily: gasto_sum = df_daily["Gasto"].sum() if "Gasto" in df_daily else 1 rev_sum = df_daily["Revenue"].sum() if "Revenue" in df_daily else 0 margen_total = rev_sum - gasto_sum d3.metric("Margen €", _eur(margen_total)) lead_col = next((c for c in df_daily.columns if "lead" in c.lower()), None) if lead_col: d4.metric("Leads totales", int(pd.to_numeric(df_daily[lead_col], errors="coerce").fillna(0).sum())) # Gráfico de ritmo: ratio_leads vs ratio_mes día a día if "Leads" in df_daily and camp_row["cap"] > 0: st.markdown("**Ritmo: leads acumulados vs objetivo esperado**") df_ritmo = df_daily[["Fecha", "Leads"]].copy() if "Fecha" in df_daily else df_daily[["Leads"]].copy() df_ritmo["Leads acum."] = df_daily["Leads"].cumsum() df_ritmo["Objetivo"] = [ round((i + 1) / dias_mes * camp_row["cap"], 1) for i in range(len(df_ritmo)) ] if "Fecha" in df_ritmo.columns: df_ritmo = df_ritmo.set_index("Fecha") st.line_chart(df_ritmo[["Leads acum.", "Objetivo"]]) # Gráfico gasto vs revenue numeric_cols = [c for c in ["Gasto", "Revenue"] if c in df_daily.columns] if numeric_cols: st.markdown("**Gasto vs Revenue diario**") chart_data = df_daily[numeric_cols].copy() if "Fecha" in df_daily.columns: chart_data.index = df_daily["Fecha"] st.bar_chart(chart_data) # Tabla detalle st.markdown("**Detalle diario**") display_cols = [c for c in ["Fecha", "Leads", "Gasto", "Revenue", "Margen"] if c in df_daily.columns] st.dataframe(df_daily[display_cols], use_container_width=True, hide_index=True) else: st.info("Sin datos diarios para esta campaña.") else: st.info("Sin métricas diarias para esta campaña.") st.divider() st.subheader("Portfolio agregado (GAMes)") @st.cache_data(ttl=300, show_spinner="Cargando portfolio...") def _load_games(mes: int, anio: int) -> dict: at = AirtableClient() formula = f"AND({{Mes}}='{mes}',{{Año}}='{anio}')" records = at.games.all(formula=formula) if not records: return {} f = records[0]["fields"] try: metricas = json.loads(f.get("MetricasDiarias") or "{}") except Exception: metricas = {} return { "metricas": metricas, "coste_mes": float(f.get("CosteMes") or 0), "conv_mes": float(f.get("ConvMes") or 0), "ppl_medio": float(f.get("PPLMedio") or 0), "cpa_medio": float(f.get("CPAMedio") or 0), } games = _load_games(int(mes_sel), int(anio_sel)) if games: g1, g2, g3, g4 = st.columns(4) g1.metric("Gasto mes (portfolio)", _eur(games["coste_mes"])) g2.metric("Leads mes (portfolio)", f"{games['conv_mes']:.0f}") g3.metric("PPL medio", _eur(games["ppl_medio"])) g4.metric("CPA medio", _eur(games["cpa_medio"])) if games["metricas"]: if isinstance(games["metricas"], dict): gdf = pd.DataFrame([{"Fecha": k, **v} for k, v in sorted(games["metricas"].items())]) else: gdf = pd.DataFrame(sorted(games["metricas"], key=lambda x: x.get("fecha", ""))) col_map = {} for c in gdf.columns: cl = c.lower() if "coste" in cl or "cost" in cl: col_map[c] = "Gasto" elif "ingres" in cl or "revenue" in cl: col_map[c] = "Revenue" elif "margen" in cl: col_map[c] = "Margen" elif "lead" in cl or "conv" in cl: col_map[c] = "Leads" elif "fecha" in cl or "date" in cl: col_map[c] = "Fecha" gdf = gdf.rename(columns=col_map) numeric_cols = [c for c in ["Gasto", "Revenue"] if c in gdf.columns] if numeric_cols: st.markdown("**Portfolio: Gasto vs Revenue diario**") chart_data = gdf[numeric_cols].copy() if "Fecha" in gdf.columns: chart_data.index = gdf["Fecha"] st.bar_chart(chart_data) else: st.info("Sin datos de portfolio para el período seleccionado.") # ════════════════════════════════════════════════════════════════════════════ # # Tab 4 — EJECUCIÓN # # ════════════════════════════════════════════════════════════════════════════ # with tab_ejecucion: st.subheader("Lanzar run.py") col_dry, col_prod = st.columns(2) with col_dry: st.markdown("**Dry Run** — Solo análisis, sin cambios en Google Ads") if st.button("▶ Ejecutar Dry Run", use_container_width=True): st.session_state["run_mode"] = "dry" st.session_state["run_started"] = True with col_prod: st.markdown("**Producción** — Aplica cambios reales en Google Ads") confirm = st.checkbox("Confirmo que quiero ejecutar en producción") prod_disabled = not confirm if st.button("🚀 Ejecutar en Producción", use_container_width=True, disabled=prod_disabled): st.session_state["run_mode"] = "prod" st.session_state["run_started"] = True if st.session_state.get("run_started"): st.session_state["run_started"] = False mode = st.session_state.get("run_mode", "dry") env = os.environ.copy() env["DRY_RUN"] = "true" if mode == "dry" else "false" run_py = os.path.join(os.path.dirname(__file__), "run.py") st.info(f"Ejecutando run.py en modo {'Dry Run' if mode == 'dry' else 'PRODUCCIÓN'}…") log_box = st.empty() lines = [] with st.spinner("En ejecución…"): proc = subprocess.Popen( [sys.executable, run_py], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", env=env, cwd=os.path.dirname(__file__), ) for line in proc.stdout: lines.append(line.rstrip()) log_box.code("\n".join(lines[-60:]), language="") proc.wait() if proc.returncode == 0: st.success("✅ Ejecución completada correctamente.") else: st.error(f"❌ Terminó con código {proc.returncode}.") st.cache_data.clear() st.divider() st.subheader("Historial de ejecuciones") logs_dir = os.path.join(os.path.dirname(__file__), "logs") log_files = sorted(glob.glob(os.path.join(logs_dir, "*.log")), reverse=True) if not log_files: st.info("Sin logs guardados. Los logs se generan al ejecutar run.py.") else: log_names = [os.path.basename(f) for f in log_files] sel_log = st.selectbox("Fichero de log", log_names) if sel_log: log_path = os.path.join(logs_dir, sel_log) with open(log_path, encoding="utf-8", errors="replace") as fh: content = fh.read() st.code(content, language="")