José Manuel Gómez
deed1db80e
- analyze_creatives.py: nuevo script independiente que analiza visualmente todos los anuncios activos, detecta fatiga creativa (CTR 3d vs 7d) y compara creatividades dentro del mismo adset usando Claude Sonnet con visión - agent.py: analyze_creative_deep() con métricas de rendimiento + detección de fatiga, compare_adset_creatives() para comparativa multi-imagen, fallback de descarga de imágenes por lista de URLs, prompts en español - meta_ads_client.py: get_ads_with_creatives() incluye adset_id, image_url separado de thumbnail_url, y video_thumbnail_url via AdVideo.picture para vídeos - baserow_client.py: get_all_creative_analyses() y get_creative_history_by_ad() - dashboard.py: nueva pestaña Creatividades con tabla seleccionable, panel lateral con thumbnail + análisis + recomendaciones + gráfico de evolución del score - slack_notifier.py: scorecard compacto (una línea por anuncio con acción breve), fix del límite de 50 bloques via flush proactivo antes de cada adset Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
604 lines
25 KiB
Python
604 lines
25 KiB
Python
"""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="expanded",
|
|
)
|
|
|
|
_STRATEGY_LABELS = {
|
|
"LOWEST_COST_WITHOUT_CAP": "Menor coste",
|
|
"LOWEST_COST_WITH_BID_CAP": "Cap. puja",
|
|
"COST_CAP": "Cap. coste",
|
|
"MINIMUM_ROAS": "ROAS mín.",
|
|
}
|
|
|
|
|
|
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 "—"
|
|
|
|
|
|
@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
|
|
|
|
|
|
# ── Sidebar ───────────────────────────────────────────────────────────────────
|
|
|
|
st.sidebar.title("Filtros")
|
|
today = date.today()
|
|
yesterday = today - timedelta(days=1)
|
|
first_of_month = today.replace(day=1)
|
|
|
|
c1, c2 = st.sidebar.columns(2)
|
|
date_from = c1.date_input("Desde", value=first_of_month, max_value=yesterday)
|
|
date_to = c2.date_input("Hasta", value=yesterday, min_value=date_from, max_value=yesterday)
|
|
|
|
if st.sidebar.button("🔄 Actualizar", use_container_width=True):
|
|
st.cache_data.clear()
|
|
|
|
date_from_str = date_from.strftime("%Y-%m-%d")
|
|
date_to_str = date_to.strftime("%Y-%m-%d")
|
|
|
|
if date_from > date_to:
|
|
st.error("La fecha inicio debe ser anterior a la fecha fin.")
|
|
st.stop()
|
|
|
|
# ── Data ──────────────────────────────────────────────────────────────────────
|
|
|
|
daily_rows, campaign_metrics, vertical_cpls = _load_data(date_from_str, date_to_str)
|
|
|
|
# Aggregate daily totals with per-vertical margins
|
|
_daily: dict = {}
|
|
for row in daily_rows:
|
|
v = _extract_vertical(row["campaign_name"])
|
|
target = vertical_cpls.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())
|
|
]
|
|
|
|
# Aggregate verticals
|
|
verticals: dict = {}
|
|
for cid, m in campaign_metrics.items():
|
|
v = _extract_vertical(m["name"])
|
|
target = vertical_cpls.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:
|
|
verticals[v] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": target}
|
|
verticals[v]["spend"] += m["spend"]
|
|
verticals[v]["leads"] += m["leads"]
|
|
verticals[v]["margin"] += margin
|
|
|
|
# Vertical filter (populated after load)
|
|
v_options = ["Todos"] + sorted(verticals.keys())
|
|
selected_vertical = st.sidebar.selectbox("Vertical", v_options)
|
|
if selected_vertical != "Todos":
|
|
campaign_metrics = {
|
|
cid: m for cid, m in campaign_metrics.items()
|
|
if _extract_vertical(m["name"]) == selected_vertical
|
|
}
|
|
|
|
# ── Header ────────────────────────────────────────────────────────────────────
|
|
|
|
st.title(f"Meta Optimizer — {config.META_CAMPAIGN_PREFIX}")
|
|
st.caption(f"Período: **{date_from.strftime('%d/%m/%Y')}** → **{date_to.strftime('%d/%m/%Y')}**")
|
|
|
|
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()
|
|
|
|
# ── Tabs ──────────────────────────────────────────────────────────────────────
|
|
|
|
tab1, tab2, tab3, tab4, tab5 = st.tabs(["📅 Por día", "📊 Campañas", "🏷️ Verticales", "🗂️ Histórico", "🎨 Creatividades"])
|
|
|
|
|
|
# ── Tab 1: Por día ────────────────────────────────────────────────────────────
|
|
with tab1:
|
|
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],
|
|
)
|
|
if selected_day:
|
|
day_camp: dict = {}
|
|
for row in daily_rows:
|
|
if row["date"] != selected_day:
|
|
continue
|
|
key = row["campaign_name"]
|
|
if key not in day_camp:
|
|
v = _extract_vertical(key)
|
|
target = vertical_cpls.get(v, config.META_TARGET_CPL)
|
|
day_camp[key] = {
|
|
"name": key, "vertical": v,
|
|
"spend": 0.0, "leads": 0, "target_cpl": target,
|
|
}
|
|
day_camp[key]["spend"] += row["spend"]
|
|
day_camp[key]["leads"] += row["leads"]
|
|
|
|
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)
|
|
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 rows:
|
|
st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
|
|
else:
|
|
st.info("Sin campañas activas ese día.")
|
|
|
|
|
|
# ── Tab 2: Campañas ───────────────────────────────────────────────────────────
|
|
with tab2:
|
|
if not campaign_metrics:
|
|
st.info("Sin campañas para el período seleccionado.")
|
|
else:
|
|
camp_rows = []
|
|
for cid, m in sorted(campaign_metrics.items(), key=lambda x: -x[1]["spend"]):
|
|
v = _extract_vertical(m["name"])
|
|
target = vertical_cpls.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()))
|
|
|
|
if selected_camp:
|
|
selected_cid = camp_id_map[selected_camp]
|
|
adsets, ads, bid_cfg = _load_detail(selected_cid, date_from_str, date_to_str)
|
|
|
|
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:
|
|
if not verticals:
|
|
st.info("Sin datos de verticales.")
|
|
else:
|
|
vert_rows = []
|
|
for v, data in sorted(verticals.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 ──────────────────────────────────────────────────────────
|
|
|
|
_ACTION_COLORS = {
|
|
"INCREASE_BUDGET": "🟢",
|
|
"REDUCE_BUDGET": "🟠",
|
|
"PAUSE": "🔴",
|
|
"REVIEW_CREATIVES": "🟣",
|
|
"MAINTAIN": "⚪",
|
|
}
|
|
|
|
|
|
@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"])
|
|
|
|
|
|
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:
|
|
fmt_date = lambda s: s[8:10] + "/" + s[5:7] + "/" + s[:4]
|
|
selected_date = st.selectbox(
|
|
"Fecha del análisis",
|
|
dates,
|
|
format_func=fmt_date,
|
|
)
|
|
if st.button("🔄 Recargar", key="reload_hist"):
|
|
st.cache_data.clear()
|
|
|
|
snapshots = _load_snapshots(selected_date)
|
|
if not snapshots:
|
|
st.info("Sin datos para esa fecha.")
|
|
else:
|
|
# ── Resumen del día ───────────────────────────────────────────────
|
|
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()
|
|
|
|
# ── Tabla de campañas clicable ────────────────────────────────────
|
|
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 — expanders con evaluación visible ─────────────────
|
|
adsets = snap["adsets"]
|
|
if adsets:
|
|
st.markdown("**Conjuntos de anuncios**")
|
|
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']}")
|
|
|
|
# ── Anuncios — expanders con evaluación visible ───────────────
|
|
ads = snap["ads"]
|
|
if ads:
|
|
st.markdown("**Anuncios**")
|
|
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:
|
|
@st.cache_data(ttl=300, show_spinner="Cargando análisis de creatividades...")
|
|
def _load_creatives():
|
|
return BaserowClient().get_all_creative_analyses()
|
|
|
|
creatives_raw = _load_creatives()
|
|
|
|
if not creatives_raw:
|
|
st.info("No hay análisis de creatividades. Ejecuta `python analyze_creatives.py` para generar datos.")
|
|
st.stop()
|
|
|
|
# Build dataframe
|
|
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))
|
|
|
|
# ── Filters ───────────────────────────────────────────────────────────────
|
|
f1, f2, f3 = st.columns([2, 2, 2])
|
|
|
|
dates_available = sorted(df_all["created_at"].dropna().unique(), reverse=True)
|
|
sel_date = f1.selectbox("Fecha análisis", dates_available)
|
|
|
|
camp_ids = sorted(df_all["campaign_id"].dropna().unique().tolist()) if "campaign_id" in df_all.columns else []
|
|
sel_camp = f2.selectbox("Campaña", ["Todas"] + camp_ids)
|
|
|
|
score_min = f3.slider("Score mínimo", 0.0, 10.0, 0.0, step=0.5)
|
|
|
|
# Apply filters
|
|
df = df_all.copy()
|
|
if sel_date:
|
|
df = df[df["created_at"] == sel_date]
|
|
if sel_camp != "Todas" and "campaign_id" in df.columns:
|
|
df = df[df["campaign_id"] == 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)
|
|
|
|
# ── Fatigue alerts ────────────────────────────────────────────────────────
|
|
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_id": "Campaña ID",
|
|
"ad_name": "Anuncio",
|
|
"score": "Score",
|
|
"created_at": "Fecha",
|
|
"analysis": "Análisis",
|
|
"recommendations": "Recomendaciones",
|
|
}
|
|
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)
|
|
|
|
# Score evolution across runs
|
|
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.")
|