"""Interactive Meta Optimizer dashboard β€” Streamlit.""" import streamlit as st from datetime import date, timedelta import pandas as pd import sys import os sys.path.insert(0, os.path.dirname(__file__)) from meta_ads_client import MetaAdsClient from baserow_client import BaserowClient import config def _extract_vertical(name: str) -> str: """VIVIFUL_13_telefonia_leadads β†’ telefonia""" prefix = config.META_CAMPAIGN_PREFIX rest = name[len(prefix):].lstrip("_") parts = rest.split("_") start = 1 if parts and parts[0].isdigit() else 0 return parts[start].lower() if start < len(parts) else "otros" st.set_page_config( page_title=f"Meta Optimizer β€” {config.META_CAMPAIGN_PREFIX}", layout="wide", initial_sidebar_state="collapsed", ) import streamlit.components.v1 as components components.html(""" """, height=0) _STRATEGY_LABELS = { "LOWEST_COST_WITHOUT_CAP": "Menor coste", "LOWEST_COST_WITH_BID_CAP": "Cap. puja", "COST_CAP": "Cap. coste", "MINIMUM_ROAS": "ROAS mΓ­n.", } _ACTION_COLORS = { "INCREASE_BUDGET": "🟒", "REDUCE_BUDGET": "🟠", "PAUSE": "πŸ”΄", "REVIEW_CREATIVES": "🟣", "MAINTAIN": "βšͺ", } _today = date.today() _yesterday = _today - timedelta(days=1) _default_from = _yesterday - timedelta(days=6) def _eur(val: float) -> str: return f"{val:.2f}€" def _margin(val: float) -> str: return f"+{val:.0f}€" if val >= 0 else f"{val:.0f}€" def _status(leads: int, spend: float) -> str: if leads > 0: return "βœ…" if spend > 0: return "❌" return "β€”" def _date_row(key: str, n_extra_cols: int = 0) -> tuple: """Renders [Desde | Hasta | ...extra... | πŸ”„] columns. Returns (date_from, date_to, *extra_cols).""" cols = st.columns([2, 2] + [2] * n_extra_cols + [1]) d_from = cols[0].date_input("Desde", value=_default_from, max_value=_yesterday, key=f"{key}_from") d_to = cols[1].date_input("Hasta", value=_yesterday, min_value=d_from, max_value=_yesterday, key=f"{key}_to") if cols[-1].button("πŸ”„", key=f"{key}_ref", use_container_width=True, help="Limpiar cachΓ©"): st.cache_data.clear() st.rerun() extra = tuple(cols[2:-1]) return (d_from, d_to) + extra # ── Cached data loaders ─────────────────────────────────────────────────────── @st.cache_data(ttl=300, show_spinner="Cargando datos de Meta API...") def _load_data(date_from: str, date_to: str): meta = MetaAdsClient() baserow = BaserowClient() vertical_cpls: dict = {} try: for v in baserow.get_all_verticals(): name = (v.get("Nombre") or "").strip().lower() cpl = float(v.get("target_cpl") or 0) if name and cpl: vertical_cpls[name] = cpl except Exception: pass daily_rows = meta.get_daily_campaign_rows(date_from, date_to) campaign_metrics = meta.get_campaign_metrics(date_from, date_to) return daily_rows, campaign_metrics, vertical_cpls @st.cache_data(ttl=300, show_spinner="Cargando detalle de campaΓ±a...") def _load_detail(campaign_id: str, date_from: str, date_to: str): meta = MetaAdsClient() adsets = meta.get_adset_metrics(campaign_id, date_from, date_to) ads = meta.get_ad_metrics(campaign_id, date_from, date_to) bid = meta.get_campaign_bid_config(campaign_id) bids = meta.get_adset_bid_configs(campaign_id) for adset in adsets: b = bids.get(adset["id"], {}) adset["cost_cap_eur"] = b.get("cost_cap_eur") adset["bid_strategy"] = b.get("bid_strategy", "") return adsets, ads, bid @st.cache_data(ttl=3600, show_spinner=False) def _load_campaign_names() -> dict: """Returns {campaign_id: campaign_name} for the last 30 days. Cached 1h.""" meta = MetaAdsClient() end = _yesterday.strftime("%Y-%m-%d") start = (_yesterday - timedelta(days=29)).strftime("%Y-%m-%d") try: metrics = meta.get_campaign_metrics(start, end) return {cid: m["name"] for cid, m in metrics.items()} except Exception: return {} @st.cache_data(ttl=120, show_spinner="Cargando fechas disponibles...") def _load_snapshot_dates(): return BaserowClient().get_snapshot_dates() @st.cache_data(ttl=120, show_spinner="Cargando anΓ‘lisis del dΓ­a...") def _load_snapshots(run_date: str): import json rows = BaserowClient().get_snapshots_for_date(run_date) result = [] for r in rows: try: adsets = json.loads(r.get("adsets_json") or "[]") except Exception: adsets = [] try: ads = json.loads(r.get("ads_json") or "[]") except Exception: ads = [] result.append({ "campaign_name": r.get("campaign_name", ""), "vertical": r.get("vertical", ""), "spend": float(r.get("spend") or 0), "leads": int(r.get("leads") or 0), "cpl": float(r.get("cpl") or 0), "margin": float(r.get("margin") or 0), "action_type": r.get("action_type", "MAINTAIN"), "justification": r.get("justification", ""), "adsets": adsets, "ads": ads, }) return sorted(result, key=lambda x: -x["spend"]) @st.cache_data(ttl=300, show_spinner="Cargando anΓ‘lisis de creatividades...") def _load_creatives(): return BaserowClient().get_all_creative_analyses() # ── Header ──────────────────────────────────────────────────────────────────── st.title(f"Meta Optimizer β€” {config.META_CAMPAIGN_PREFIX}") # ── Tabs ────────────────────────────────────────────────────────────────────── tab1, tab2, tab3, tab4, tab5 = st.tabs( ["πŸ“… Por dΓ­a", "πŸ“Š CampaΓ±as", "🏷️ Verticales", "πŸ—‚οΈ HistΓ³rico", "🎨 Creatividades"] ) # ── Tab 1: Por dΓ­a ──────────────────────────────────────────────────────────── with tab1: d_from_1, d_to_1 = _date_row("t1") if d_from_1 > d_to_1: st.error("La fecha inicio debe ser anterior a la fecha fin.") else: try: daily_rows, _cm1, vertical_cpls_1 = _load_data( d_from_1.strftime("%Y-%m-%d"), d_to_1.strftime("%Y-%m-%d") ) except Exception as e: st.error(f"Error cargando datos de Meta API: {e}") daily_rows, vertical_cpls_1 = [], {} _daily: dict = {} for row in daily_rows: v = _extract_vertical(row["campaign_name"]) target = vertical_cpls_1.get(v, config.META_TARGET_CPL) margin = ( round((target - row["spend"] / row["leads"]) * row["leads"], 2) if row["leads"] > 0 else round(-row["spend"], 2) ) d = _daily.setdefault(row["date"], {"spend": 0.0, "leads": 0, "margin": 0.0}) d["spend"] += row["spend"] d["leads"] += row["leads"] d["margin"] += margin daily_totals = [ { "date": dt, "spend": round(d["spend"], 2), "leads": int(d["leads"]), "cpl": round(d["spend"] / d["leads"], 2) if d["leads"] > 0 else 0.0, "margin": round(d["margin"], 2), } for dt, d in sorted(_daily.items()) ] total_spend = sum(d["spend"] for d in daily_totals) total_leads = sum(d["leads"] for d in daily_totals) total_cpl = round(total_spend / total_leads, 2) if total_leads > 0 else 0.0 total_margin = sum(d["margin"] for d in daily_totals) k1, k2, k3, k4 = st.columns(4) k1.metric("Gasto total", _eur(total_spend)) k2.metric("Leads totales", f"{total_leads:,}") k3.metric("CPL medio", _eur(total_cpl)) k4.metric("Margen total", _margin(total_margin)) st.divider() if not daily_totals: st.info("Sin datos para el perΓ­odo seleccionado.") else: df_daily = pd.DataFrame([ { "DΓ­a": d["date"][8:10] + "/" + d["date"][5:7], "Gasto": _eur(d["spend"]), "Leads": d["leads"], "CPL": _eur(d["cpl"]), "Margen": _margin(d["margin"]), "Est": _status(d["leads"], d["spend"]), } for d in daily_totals ]) st.dataframe(df_daily, use_container_width=True, hide_index=True) st.subheader("Desglose por campaΓ±a") day_opts = [d["date"] for d in reversed(daily_totals)] selected_day = st.selectbox( "Selecciona un dΓ­a", day_opts, format_func=lambda s: s[8:10] + "/" + s[5:7] + "/" + s[:4], key="t1_day", ) if selected_day: day_camp: dict = {} for row in daily_rows: if row["date"] != selected_day: continue k = row["campaign_name"] if k not in day_camp: v = _extract_vertical(k) target = vertical_cpls_1.get(v, config.META_TARGET_CPL) day_camp[k] = {"name": k, "vertical": v, "spend": 0.0, "leads": 0, "target_cpl": target} day_camp[k]["spend"] += row["spend"] day_camp[k]["leads"] += row["leads"] camp_rows = [] for c in sorted(day_camp.values(), key=lambda x: -x["spend"]): cpl = round(c["spend"] / c["leads"], 2) if c["leads"] > 0 else 0.0 margin = ( round((c["target_cpl"] - cpl) * c["leads"], 2) if c["leads"] > 0 else round(-c["spend"], 2) ) camp_rows.append({ "CampaΓ±a": c["name"], "Vertical": c["vertical"], "Gasto": _eur(c["spend"]), "Leads": c["leads"], "CPL": _eur(cpl) if c["leads"] > 0 else "β€”", "Obj": _eur(c["target_cpl"]), "Margen": _margin(margin), }) if camp_rows: st.dataframe(pd.DataFrame(camp_rows), use_container_width=True, hide_index=True) else: st.info("Sin campaΓ±as activas ese dΓ­a.") # ── Tab 2: CampaΓ±as ─────────────────────────────────────────────────────────── with tab2: d_from_2, d_to_2, col_vert_2 = _date_row("t2", n_extra_cols=1) if d_from_2 > d_to_2: st.error("La fecha inicio debe ser anterior a la fecha fin.") else: try: _dr2, campaign_metrics_2, vertical_cpls_2 = _load_data( d_from_2.strftime("%Y-%m-%d"), d_to_2.strftime("%Y-%m-%d") ) except Exception as e: st.error(f"Error cargando datos de Meta API: {e}") campaign_metrics_2, vertical_cpls_2 = {}, {} v_opts_2 = ["Todos"] + sorted({_extract_vertical(m["name"]) for m in campaign_metrics_2.values()}) sel_vert_2 = col_vert_2.selectbox("Vertical", v_opts_2, key="t2_vert") if sel_vert_2 != "Todos": campaign_metrics_2 = { cid: m for cid, m in campaign_metrics_2.items() if _extract_vertical(m["name"]) == sel_vert_2 } if not campaign_metrics_2: st.info("Sin campaΓ±as para el perΓ­odo seleccionado.") else: camp_rows = [] for cid, m in sorted(campaign_metrics_2.items(), key=lambda x: -x[1]["spend"]): v = _extract_vertical(m["name"]) target = vertical_cpls_2.get(v, config.META_TARGET_CPL) margin = round((target - m["cpl"]) * m["leads"], 2) if m["leads"] > 0 else round(-m["spend"], 2) camp_rows.append({ "CampaΓ±a": m["name"], "Vertical": v, "Gasto": _eur(m["spend"]), "Leads": m["leads"], "CPL": _eur(m["cpl"]) if m["leads"] > 0 else "β€”", "Obj": _eur(target), "Margen": _margin(margin), "CTR": f"{m['ctr']:.1f}%", "_cid": cid, }) df_camps = pd.DataFrame([{k: v for k, v in r.items() if k != "_cid"} for r in camp_rows]) st.dataframe(df_camps, use_container_width=True, hide_index=True) st.subheader("Detalle de campaΓ±a") camp_id_map = {r["CampaΓ±a"]: r["_cid"] for r in camp_rows} selected_camp = st.selectbox("Selecciona una campaΓ±a", list(camp_id_map.keys()), key="t2_camp") if selected_camp: selected_cid = camp_id_map[selected_camp] adsets, ads, bid_cfg = _load_detail( selected_cid, d_from_2.strftime("%Y-%m-%d"), d_to_2.strftime("%Y-%m-%d"), ) strategy = bid_cfg.get("bid_strategy", "") strat_label = _STRATEGY_LABELS.get(strategy, strategy or "β€”") budget = bid_cfg.get("daily_budget_eur") budget_str = f"{budget:.0f}€/dΓ­a" if budget else "β€”" st.caption(f"Estrategia: **{strat_label}** | Presupuesto: **{budget_str}**") if adsets: st.markdown("**Conjuntos de anuncios**") df_adsets = pd.DataFrame([ { "Nombre": a["name"], "Gasto": _eur(a["spend"]), "Leads": a["leads"], "CPL": _eur(a["cpl"]) if a["leads"] > 0 else "β€”", "CTR": f"{a['ctr']:.1f}%", "Cap": _eur(a["cost_cap_eur"]) if a.get("cost_cap_eur") else "Auto", } for a in adsets ]) st.dataframe(df_adsets, use_container_width=True, hide_index=True) else: st.info("Sin conjuntos de anuncios con gasto en este perΓ­odo.") if ads: st.markdown("**Anuncios**") df_ads = pd.DataFrame([ { "Nombre": a["name"], "Gasto": _eur(a["spend"]), "Leads": a["leads"], "CPL": _eur(a["cpl"]) if a["leads"] > 0 else "β€”", "CTR": f"{a['ctr']:.1f}%", "CPM": _eur(a["cpm"]), } for a in ads ]) st.dataframe(df_ads, use_container_width=True, hide_index=True) else: st.info("Sin anuncios con gasto en este perΓ­odo.") # ── Tab 3: Verticales ───────────────────────────────────────────────────────── with tab3: d_from_3, d_to_3 = _date_row("t3") if d_from_3 > d_to_3: st.error("La fecha inicio debe ser anterior a la fecha fin.") else: try: _dr3, campaign_metrics_3, vertical_cpls_3 = _load_data( d_from_3.strftime("%Y-%m-%d"), d_to_3.strftime("%Y-%m-%d") ) except Exception as e: st.error(f"Error cargando datos de Meta API: {e}") campaign_metrics_3, vertical_cpls_3 = {}, {} verticals_3: dict = {} for cid, m in campaign_metrics_3.items(): v = _extract_vertical(m["name"]) target = vertical_cpls_3.get(v, config.META_TARGET_CPL) margin = round((target - m["cpl"]) * m["leads"], 2) if m["leads"] > 0 else round(-m["spend"], 2) if v not in verticals_3: verticals_3[v] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": target} verticals_3[v]["spend"] += m["spend"] verticals_3[v]["leads"] += m["leads"] verticals_3[v]["margin"] += margin if not verticals_3: st.info("Sin datos de verticales.") else: vert_rows = [] for v, data in sorted(verticals_3.items(), key=lambda x: -x[1]["margin"]): v_leads = data["leads"] v_spend = data["spend"] v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 vert_rows.append({ "Vertical": v, "Gasto": _eur(v_spend), "Leads": v_leads, "CPL": _eur(v_cpl), "Obj": _eur(data["target_cpl"]) if data.get("target_cpl") else "β€”", "Margen": _margin(data["margin"]), }) st.dataframe(pd.DataFrame(vert_rows), use_container_width=True, hide_index=True) # ── Tab 4: HistΓ³rico ────────────────────────────────────────────────────────── with tab4: dates = _load_snapshot_dates() if not dates: st.info("Sin anΓ‘lisis guardados aΓΊn. Los snapshots se generan al ejecutar run.py.") else: c1, c2 = st.columns([3, 1]) fmt_date = lambda s: s[8:10] + "/" + s[5:7] + "/" + s[:4] selected_date = c1.selectbox("Fecha del anΓ‘lisis", dates, format_func=fmt_date, key="t4_date") if c2.button("πŸ”„ Recargar", key="t4_ref", use_container_width=True): st.cache_data.clear() st.rerun() snapshots = _load_snapshots(selected_date) if not snapshots: st.info("Sin datos para esa fecha.") else: d_spend = sum(s["spend"] for s in snapshots) d_leads = sum(s["leads"] for s in snapshots) d_cpl = round(d_spend / d_leads, 2) if d_leads > 0 else 0.0 d_margin = sum(s["margin"] for s in snapshots) h1, h2, h3, h4 = st.columns(4) h1.metric("Gasto", _eur(d_spend)) h2.metric("Leads", f"{d_leads:,}") h3.metric("CPL", _eur(d_cpl)) h4.metric("Margen", _margin(d_margin)) st.divider() df_snap = pd.DataFrame([ { "AcciΓ³n": _ACTION_COLORS.get(s["action_type"], "βšͺ") + " " + s["action_type"], "CampaΓ±a": s["campaign_name"], "Vertical": s["vertical"], "Gasto": _eur(s["spend"]), "Leads": s["leads"], "CPL": _eur(s["cpl"]) if s["leads"] > 0 else "β€”", "Margen": _margin(s["margin"]), } for s in snapshots ]) event = st.dataframe( df_snap, use_container_width=True, hide_index=True, on_select="rerun", selection_mode="single-row", ) sel_rows = event.selection.rows if sel_rows: snap = snapshots[sel_rows[0]] st.subheader(snap["campaign_name"]) st.caption( f"Vertical: **{snap['vertical']}** | " f"DecisiΓ³n: **{snap['action_type']}** | " f"Margen: **{_margin(snap['margin'])}**" ) if snap["justification"]: st.info(snap["justification"]) adsets = snap["adsets"] if adsets: st.markdown("**Conjuntos de anuncios** _(ΓΊltimos 3 dΓ­as)_") for a in adsets: label = ( f"{a['name']} β€” " f"{_eur(a['spend'])} Β· {a['leads']} leads Β· " f"CPL {_eur(a['cpl']) if a['leads'] > 0 else 'β€”'} Β· " f"CTR {a.get('ctr', 0):.1f}%" ) with st.expander(label): if a.get("cost_cap_eur"): st.caption(f"Cap: {_eur(a['cost_cap_eur'])}") if a.get("evaluacion"): st.write(f"_{a['evaluacion']}_") if a.get("recomendacion"): st.write(f"β†’ {a['recomendacion']}") ads = snap["ads"] if ads: st.markdown("**Anuncios** _(ΓΊltimos 7 dΓ­as)_") for a in ads: label = ( f"{a['name']} β€” " f"{_eur(a['spend'])} Β· {a['leads']} leads Β· " f"CPL {_eur(a['cpl']) if a['leads'] > 0 else 'β€”'} Β· " f"CTR {a.get('ctr', 0):.1f}% Β· " f"CPM {_eur(a.get('cpm', 0))}" ) with st.expander(label): if a.get("evaluacion"): st.write(f"_{a['evaluacion']}_") if a.get("recomendacion"): st.write(f"β†’ {a['recomendacion']}") # ── Tab 5: Creatividades ────────────────────────────────────────────────────── with tab5: creatives_raw = _load_creatives() if not creatives_raw: st.info("No hay anΓ‘lisis de creatividades. Ejecuta `python analyze_creatives.py` para generar datos.") else: df_all = pd.DataFrame(creatives_raw) df_all["score"] = pd.to_numeric(df_all.get("score", 0), errors="coerce").fillna(0) df_all["created_at"] = df_all.get("created_at", pd.Series(dtype=str)) # Map campaign_id β†’ name using a dedicated cached call (last 30d) camp_id_to_name = _load_campaign_names() df_all["campaign_name"] = df_all["campaign_id"].map( lambda cid: camp_id_to_name.get(str(cid), str(cid)) ) df_all["vertical"] = df_all["campaign_name"].map(_extract_vertical) # ── Filters ─────────────────────────────────────────────────────────── f1, f2, f3, f4 = st.columns([2, 2, 2, 2]) dates_available = sorted(df_all["created_at"].dropna().unique(), reverse=True) sel_date = f1.selectbox("Fecha anΓ‘lisis", dates_available, key="cr_date") verts_available = sorted(df_all["vertical"].dropna().unique().tolist()) sel_vert_cr = f2.selectbox("Vertical", ["Todas"] + verts_available, key="cr_vert") camp_names = sorted(df_all["campaign_name"].dropna().unique().tolist()) sel_camp = f3.selectbox("CampaΓ±a", ["Todas"] + camp_names, key="cr_camp") score_min = f4.slider("Score mΓ­nimo", 0.0, 10.0, 0.0, step=0.5, key="cr_score") # Apply filters df = df_all.copy() if sel_date: df = df[df["created_at"] == sel_date] if sel_vert_cr != "Todas": df = df[df["vertical"] == sel_vert_cr] if sel_camp != "Todas": df = df[df["campaign_name"] == sel_camp] if score_min > 0: df = df[df["score"] >= score_min] # ── KPIs ────────────────────────────────────────────────────────────── scored_df = df[df["score"] > 0] avg_sc = round(scored_df["score"].mean(), 1) if not scored_df.empty else 0.0 fatigue_n = int(df["analysis"].str.contains("FATIGA", na=False).sum()) if "analysis" in df.columns else 0 last_run = df_all["created_at"].max() if not df_all.empty else "β€”" k1, k2, k3, k4 = st.columns(4) k1.metric("Anuncios", len(df)) k2.metric("Score medio", f"{avg_sc}/10") k3.metric("Con fatiga", fatigue_n) k4.metric("Última ejecuciΓ³n", last_run) if fatigue_n: fatigued = df[df["analysis"].str.contains("FATIGA", na=False)] with st.expander(f"⚠️ {fatigue_n} anuncios con fatiga creativa", expanded=True): for _, row in fatigued.iterrows(): st.warning(f"**{row.get('ad_name', 'β€”')}** β€” Score {row.get('score', 0):.1f}/10") st.divider() # ── Table + Detail panel ────────────────────────────────────────────── rename_map = { "campaign_name": "CampaΓ±a", "vertical": "Vertical", "ad_name": "Anuncio", "score": "Score", "created_at": "Fecha", } display_cols = [c for c in rename_map if c in df.columns] df_display = df[display_cols].rename(columns=rename_map).reset_index(drop=True) col_table, col_detail = st.columns([3, 2]) with col_table: st.caption("Haz clic en una fila para ver el detalle β†’") event = st.dataframe( df_display, use_container_width=True, selection_mode="single-row", on_select="rerun", column_config={ "Score": st.column_config.ProgressColumn( "Score", min_value=0, max_value=10, format="%.1f" ), }, hide_index=True, ) selected_rows = event.selection.rows if hasattr(event, "selection") else [] with col_detail: if selected_rows: row = df.iloc[selected_rows[0]] score = float(row.get("score", 0)) sc_emoji = "🟒" if score >= 8 else "🟑" if score >= 6 else "🟠" if score >= 4 else "πŸ”΄" st.markdown(f"### {row.get('ad_name', 'β€”')}") st.markdown(f"{sc_emoji} **Score: {score:.1f} / 10**") img_url = str(row.get("image_url", "")) if img_url.startswith("http"): try: st.image(img_url, use_container_width=True) except Exception: st.caption("_Imagen no disponible_") analysis = str(row.get("analysis", "")) if analysis: st.markdown("**AnΓ‘lisis**") st.write(analysis) rec = str(row.get("recommendations", "")) if rec: st.markdown("**Recomendaciones**") st.info(rec) ad_id = str(row.get("ad_id", "")) if ad_id: history = BaserowClient().get_creative_history_by_ad(ad_id) if len(history) >= 2: st.markdown("**EvoluciΓ³n del score**") hist_df = pd.DataFrame(history)[["created_at", "score"]].dropna() hist_df["score"] = pd.to_numeric(hist_df["score"], errors="coerce") hist_df = hist_df[hist_df["score"] > 0].set_index("created_at") st.line_chart(hist_df) else: st.info("← Selecciona un anuncio en la tabla para ver el detalle.")