José Manuel Gómez
769d86c896
- Broaden Airtable lead counting to attr_utm_source IN ('Lead ads','landingpage')
— the 'landingpage' leads (100% fbclid, 0% gclid) were being missed entirely,
undercounting real leads for '_web' suffixed campaigns and skewing
capping/pacing decisions since yesterday's first production run.
- Add airtable_client.get_meta_leads_bulk() for day/curso-level aggregation.
- Drop per-familia Slack sectioning in favor of a single Formación block,
chunked by campaign batches instead.
- Add daily AT-vs-Meta table, per-curso PPL/CPL contrast table (leadform vs
landing breakdown), and a Claude-generated portfolio strategic diagnosis
(ported from leads-optimizer's portfolio_daily_analysis).
- Persist daily aggregate totals to a new Baserow table (daily_metrics) so
the dashboard and future reports don't depend on Meta's historical API
access remaining available indefinitely.
- Surface adset/ad-level recommendations in the campaign cards instead of
only numeric tables.
657 lines
28 KiB
Python
657 lines
28 KiB
Python
"""Interactive Meta Optimizer Formación 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 airtable_client import AirtableClient, extract_cursoid
|
||
from baserow_client import BaserowClient
|
||
import config
|
||
|
||
|
||
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("""
|
||
<script>
|
||
// Ping cada 30s para mantener el WebSocket activo y evitar el error 401 por inactividad
|
||
setInterval(function() {
|
||
fetch('/_stcore/health').catch(function() {});
|
||
}, 30000);
|
||
</script>
|
||
""", 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": "🔴",
|
||
"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 _familia_of(name: str, familia_lookup: dict) -> str:
|
||
return familia_lookup.get(extract_cursoid(name) or "", "Sin familia")
|
||
|
||
|
||
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 PPL/familia de Airtable...")
|
||
def _load_lookups():
|
||
ppl_lookup, _, familia_lookup = AirtableClient().build_campaign_lookups()
|
||
return ppl_lookup, familia_lookup
|
||
|
||
|
||
@st.cache_data(ttl=300, show_spinner="Cargando datos de Meta API...")
|
||
def _load_data(date_from: str, date_to: str):
|
||
meta = MetaAdsClient()
|
||
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
|
||
|
||
|
||
@st.cache_data(ttl=300, show_spinner="Cargando métricas diarias (Baserow)...")
|
||
def _load_daily_metrics(date_from: str, date_to: str):
|
||
"""Totales diarios persistidos por run.py (Leads Meta vs Leads Airtable) —
|
||
no depende de volver a pedirle el histórico a Meta."""
|
||
rows = BaserowClient().get_daily_metrics(date_from, date_to)
|
||
return [
|
||
{
|
||
"date": r["date"],
|
||
"spend": float(r.get("spend") or 0),
|
||
"leads_meta": int(r.get("leads_meta") or 0),
|
||
"leads_at": int(r.get("leads_at") or 0),
|
||
"ing_meta": float(r.get("ing_meta") or 0),
|
||
"ing_at": float(r.get("ing_at") or 0),
|
||
"margin": float(r.get("margin") or 0),
|
||
"margin_pct": float(r.get("margin_pct") or 0),
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@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", ""),
|
||
"familia": r.get("familia", ""),
|
||
"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}")
|
||
|
||
ppl_lookup, familia_lookup = _load_lookups()
|
||
|
||
# ── Tabs ──────────────────────────────────────────────────────────────────────
|
||
|
||
tab1, tab2, tab3, tab4, tab5 = st.tabs(
|
||
["📅 Por día", "📊 Campañas", "🏷️ Familias", "🗂️ 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_totals = _load_daily_metrics(d_from_1.strftime("%Y-%m-%d"), d_to_1.strftime("%Y-%m-%d"))
|
||
except Exception as e:
|
||
st.error(f"Error cargando daily_metrics de Baserow: {e}")
|
||
daily_totals = []
|
||
try:
|
||
daily_rows, _cm1 = _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 = []
|
||
|
||
total_spend = sum(d["spend"] for d in daily_totals)
|
||
total_leads_m = sum(d["leads_meta"] for d in daily_totals)
|
||
total_leads_at = sum(d["leads_at"] for d in daily_totals)
|
||
total_ing_m = sum(d["ing_meta"] for d in daily_totals)
|
||
total_margin = total_ing_m - total_spend
|
||
total_pct = round(total_margin / total_ing_m * 100, 1) if total_ing_m > 0 else 0.0
|
||
|
||
k1, k2, k3, k4, k5 = st.columns(5)
|
||
k1.metric("Gasto total", _eur(total_spend))
|
||
k2.metric("Leads Meta", f"{total_leads_m:,}")
|
||
k3.metric("Leads Airtable", f"{total_leads_at:,}")
|
||
k4.metric("Margen (Meta)", _margin(total_margin))
|
||
k5.metric("% Margen", f"{total_pct:+.1f}%")
|
||
st.divider()
|
||
|
||
if not daily_totals:
|
||
st.info("Sin datos persistidos para el período seleccionado — ejecuta run.py o amplía el rango.")
|
||
else:
|
||
df_daily = pd.DataFrame([
|
||
{
|
||
"Día": d["date"][8:10] + "/" + d["date"][5:7],
|
||
"L. AT": d["leads_at"],
|
||
"L. Meta": d["leads_meta"],
|
||
"Gasto": _eur(d["spend"]),
|
||
"€ AT": _eur(d["ing_at"]),
|
||
"€ Meta": _eur(d["ing_meta"]),
|
||
"Margen": _margin(d["margin"]),
|
||
"% Margen": f"{d['margin_pct']:+.1f}%",
|
||
"Est": _status(d["leads_meta"], d["spend"]),
|
||
}
|
||
for d in daily_totals
|
||
])
|
||
st.dataframe(df_daily, use_container_width=True, hide_index=True)
|
||
st.caption("L. AT = leads Airtable (leadform + landing) · L. Meta = conversión propia de Meta · "
|
||
"€ AT / € Meta = leads × PPL de cada fuente · el margen oficial usa el tracking de Meta.")
|
||
|
||
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:
|
||
ppl = ppl_lookup.get(extract_cursoid(k) or "", 0)
|
||
day_camp[k] = {"name": k, "familia": _familia_of(k, familia_lookup),
|
||
"spend": 0.0, "leads": 0, "ppl": ppl}
|
||
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["leads"] * c["ppl"] - c["spend"], 2)
|
||
camp_rows.append({
|
||
"Campaña": c["name"],
|
||
"Familia": c["familia"],
|
||
"Gasto": _eur(c["spend"]),
|
||
"Leads": c["leads"],
|
||
"CPL": _eur(cpl) if c["leads"] > 0 else "—",
|
||
"PPL": _eur(c["ppl"]) if c["ppl"] else "—",
|
||
"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_fam_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 = _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 = {}
|
||
|
||
fam_opts_2 = ["Todas"] + sorted({_familia_of(m["name"], familia_lookup) for m in campaign_metrics_2.values()})
|
||
sel_fam_2 = col_fam_2.selectbox("Familia", fam_opts_2, key="t2_fam")
|
||
|
||
if sel_fam_2 != "Todas":
|
||
campaign_metrics_2 = {
|
||
cid: m for cid, m in campaign_metrics_2.items()
|
||
if _familia_of(m["name"], familia_lookup) == sel_fam_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"]):
|
||
ppl = ppl_lookup.get(extract_cursoid(m["name"]) or "", 0)
|
||
margin = round(m["leads"] * ppl - m["spend"], 2)
|
||
camp_rows.append({
|
||
"Campaña": m["name"],
|
||
"Familia": _familia_of(m["name"], familia_lookup),
|
||
"Gasto": _eur(m["spend"]),
|
||
"Leads": m["leads"],
|
||
"CPL": _eur(m["cpl"]) if m["leads"] > 0 else "—",
|
||
"PPL": _eur(ppl) if ppl else "—",
|
||
"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: Familias ───────────────────────────────────────────────────────────
|
||
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 = _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 = {}
|
||
|
||
familias_3: dict = {}
|
||
for cid, m in campaign_metrics_3.items():
|
||
fam = _familia_of(m["name"], familia_lookup)
|
||
ppl = ppl_lookup.get(extract_cursoid(m["name"]) or "", 0)
|
||
margin = round(m["leads"] * ppl - m["spend"], 2)
|
||
if fam not in familias_3:
|
||
familias_3[fam] = {"spend": 0.0, "leads": 0, "margin": 0.0}
|
||
familias_3[fam]["spend"] += m["spend"]
|
||
familias_3[fam]["leads"] += m["leads"]
|
||
familias_3[fam]["margin"] += margin
|
||
|
||
if not familias_3:
|
||
st.info("Sin datos de familias.")
|
||
else:
|
||
fam_rows = []
|
||
for fam, data in sorted(familias_3.items(), key=lambda x: -x[1]["margin"]):
|
||
f_leads = data["leads"]
|
||
f_spend = data["spend"]
|
||
f_cpl = round(f_spend / f_leads, 2) if f_leads > 0 else 0.0
|
||
fam_rows.append({
|
||
"Familia": fam,
|
||
"Gasto": _eur(f_spend),
|
||
"Leads": f_leads,
|
||
"CPL": _eur(f_cpl),
|
||
"Margen": _margin(data["margin"]),
|
||
})
|
||
st.dataframe(pd.DataFrame(fam_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"],
|
||
"Familia": s["familia"],
|
||
"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"Familia: **{snap['familia']}** | "
|
||
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["familia"] = df_all["campaign_name"].map(lambda n: _familia_of(n, familia_lookup))
|
||
|
||
# ── 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")
|
||
|
||
fams_available = sorted(df_all["familia"].dropna().unique().tolist())
|
||
sel_fam_cr = f2.selectbox("Familia", ["Todas"] + fams_available, key="cr_fam")
|
||
|
||
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_fam_cr != "Todas":
|
||
df = df[df["familia"] == sel_fam_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",
|
||
"familia": "Familia",
|
||
"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.")
|