From deed1db80edf8526467d5bc7ef3810f76a57f97c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Manuel=20G=C3=B3mez?= Date: Fri, 19 Jun 2026 12:03:10 +0200 Subject: [PATCH] Add deep creative analysis: standalone script, dashboard tab, compact Slack scorecard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- agent.py | 178 ++++++++++++++++++ analyze_creatives.py | 213 ++++++++++++++++++++++ baserow_client.py | 24 +++ dashboard.py | 130 +++++++++++++- meta_ads_client.py | 36 ++-- slack_notifier.py | 416 ++++++++++++++++++++++++------------------- 6 files changed, 802 insertions(+), 195 deletions(-) create mode 100644 analyze_creatives.py diff --git a/agent.py b/agent.py index 33060dd..8ebda00 100644 --- a/agent.py +++ b/agent.py @@ -192,3 +192,181 @@ def analyze_creative(image_url: str, ad_name: str) -> dict: return {"score": 0, "analysis": "Error parsing creative analysis.", "recommendations": ""} except Exception as e: return {"score": 0, "analysis": f"Creative analysis failed: {e}", "recommendations": ""} + + +CREATIVE_DEEP_SYSTEM = """ +IDIOMA: Responde SIEMPRE en español. Todos los campos del JSON deben estar en español. + +Eres un experto en análisis de creatividades de Meta Ads con conocimiento de neuromarketing y diseño persuasivo. +Recibirás una imagen publicitaria junto con sus métricas de rendimiento reales. + +Evalúa considerando: +1. CALIDAD VISUAL: claridad del mensaje, jerarquía visual, CTA, copy, atractivo y relevancia +2. CORRELACIÓN CON RENDIMIENTO: ¿el CTR y CPL reales son consistentes con la calidad visual? +3. SEÑAL DE FATIGA: si CTR 3d < CTR 7d × 0.75 indica saturación de audiencia +4. RECOMENDACIONES: mejoras concretas y priorizadas para mejorar CTR y conversiones + +Devuelve SOLO JSON válido sin markdown (todos los textos en español): +{ + "score": 7.5, + "analysis": "análisis conciso en español: qué funciona, qué no, correlación con rendimiento real", + "recommendations": "mejoras concretas en español en orden de impacto esperado", + "fatigue": false, + "fatigue_reason": null +} + +Score 1-10: 1-3 crítico (pausar), 4-5 bajo, 6-7 aceptable, 8-9 bueno, 10 excelente. +Si el anuncio tiene buen rendimiento real (CPL bajo, CTR alto) pero diseño mediocre, sube el score. +Si el diseño parece bueno pero el rendimiento es pobre, baja el score y explica la desconexión. +""" + +CREATIVE_COMPARE_SYSTEM = """ +IDIOMA: Responde SIEMPRE en español. Todos los campos del JSON deben estar en español. + +Eres un experto en análisis comparativo de creatividades de Meta Ads. +Recibirás varios anuncios del mismo adset con sus imágenes y métricas de rendimiento. +Evalúa cuál funciona mejor considerando tanto calidad visual como rendimiento real. + +Devuelve SOLO JSON válido sin markdown (todos los textos en español): +{ + "winner": "nombre exacto del anuncio ganador", + "ranking": [ + {"name": "nombre completo del anuncio", "rank": 1, "reason": "razón en español"} + ], + "insights": "observación comparativa clave en español: ¿qué diferencia visualmente al ganador del resto?" +} +""" + + +def _download_image(image_url) -> tuple | None: + """Returns (base64_data, media_type) or None. Accepts str or list of URLs (tries in order).""" + urls = [image_url] if isinstance(image_url, str) else image_url + for url in urls: + if not url: + continue + try: + resp = requests.get(url, timeout=15) + resp.raise_for_status() + content_type = resp.headers.get("content-type", "image/jpeg") + if not content_type.startswith("image/"): + continue + data = base64.standard_b64encode(resp.content).decode("utf-8") + return data, content_type.split(";")[0] + except Exception: + continue + return None + + +def _parse_json_response(raw: str) -> dict: + import re + clean = re.sub(r"```json\s*", "", raw.strip()) + clean = re.sub(r"```\s*", "", clean).strip() + try: + return json.loads(clean) + except json.JSONDecodeError: + start, end = clean.find("{"), clean.rfind("}") + if start != -1 and end > start: + try: + return json.loads(clean[start:end + 1]) + except json.JSONDecodeError: + pass + return {} + + +def analyze_creative_deep(image_url: str, ad_name: str, metrics: dict) -> dict: + """Deep creative analysis combining visual quality with performance data and fatigue detection.""" + _default = {"score": 0, "analysis": "", "recommendations": "", "fatigue": False, "fatigue_reason": None} + + downloaded = _download_image(image_url) + if not downloaded: + return {**_default, "analysis": "Error descargando imagen."} + image_data, media_type = downloaded + + ctr_7d = metrics.get("ctr_7d", 0) + ctr_3d = metrics.get("ctr_3d", 0) + fatigue_hint = "" + if ctr_7d > 0 and ctr_3d > 0 and ctr_3d < ctr_7d * 0.75: + fatigue_hint = f"\n⚠️ SEÑAL DE FATIGA DETECTADA: CTR cayó de {ctr_7d:.2f}% (7d) a {ctr_3d:.2f}% (3d) — posible saturación" + + context = ( + f'Anuncio: "{ad_name}"\n\n' + f"Métricas reales:\n" + f"- 7 días: gasto {metrics.get('spend_7d', 0):.0f}€, " + f"{metrics.get('leads_7d', 0)} leads, " + f"CPL {metrics.get('cpl_7d', 0):.2f}€, " + f"CTR {ctr_7d:.2f}%\n" + f"- 3 días: gasto {metrics.get('spend_3d', 0):.0f}€, " + f"{metrics.get('leads_3d', 0)} leads, " + f"CPL {metrics.get('cpl_3d', 0):.2f}€, " + f"CTR {ctr_3d:.2f}%\n" + f"- Objetivo CPL máximo: {metrics.get('max_cpl', 0):.2f}€" + f"{fatigue_hint}\n\n" + f"Analiza esta creatividad:" + ) + + try: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=700, + system=CREATIVE_DEEP_SYSTEM, + messages=[{ + "role": "user", + "content": [ + {"type": "image", "source": {"type": "base64", "media_type": media_type, "data": image_data}}, + {"type": "text", "text": context}, + ], + }], + ) + result = _parse_json_response(response.content[0].text) + if not result: + return {**_default, "analysis": "Error parseando respuesta."} + result.setdefault("fatigue", False) + result.setdefault("fatigue_reason", None) + return result + except Exception as e: + return {**_default, "analysis": f"Error en análisis: {e}"} + + +def compare_adset_creatives(ads: list) -> dict: + """Compare up to 4 ads within the same adset. ads: list of analyzed ad dicts.""" + _default = {"winner": "", "ranking": [], "insights": "Sin datos suficientes para comparar."} + + top_ads = sorted(ads, key=lambda x: -x.get("spend_7d", 0))[:4] + content_blocks = [] + + for i, ad in enumerate(top_ads, 1): + downloaded = _download_image(ad["image_url"]) + if downloaded: + image_data, media_type = downloaded + content_blocks.append({ + "type": "image", + "source": {"type": "base64", "media_type": media_type, "data": image_data}, + }) + fatigue_note = f" ⚠️ Fatiga: {ad.get('fatigue_reason','')}" if ad.get("fatigue") else "" + content_blocks.append({ + "type": "text", + "text": ( + f"Anuncio {i}: \"{ad['ad_name']}\"\n" + f"Score: {ad.get('score', 0):.1f}/10 | " + f"CTR 7d: {ad.get('ctr_7d', 0):.2f}% | " + f"CPL 7d: {ad.get('cpl_7d', 0):.2f}€ | " + f"Leads 7d: {ad.get('leads_7d', 0)} | " + f"Gasto 7d: {ad.get('spend_7d', 0):.0f}€" + f"{fatigue_note}" + ), + }) + + if not content_blocks: + return _default + + try: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=600, + system=CREATIVE_COMPARE_SYSTEM, + messages=[{"role": "user", "content": content_blocks}], + ) + result = _parse_json_response(response.content[0].text) + return result if result else _default + except Exception as e: + return {**_default, "insights": f"Error en comparativa: {e}"} diff --git a/analyze_creatives.py b/analyze_creatives.py new file mode 100644 index 0000000..38bd7ea --- /dev/null +++ b/analyze_creatives.py @@ -0,0 +1,213 @@ +""" +Análisis profundo de creatividades de Meta Ads. +Analiza visualmente cada anuncio activo, correlaciona con métricas de rendimiento, +detecta fatiga creativa y compara anuncios dentro del mismo adset. + +Uso: + python analyze_creatives.py # todas las campañas + python analyze_creatives.py --campaign VIVIFUL_5 # filtrar por nombre + python analyze_creatives.py --no-slack # sin envío a Slack +""" + +import argparse +import sys +import time +from datetime import datetime + +sys.stdout.reconfigure(encoding="utf-8", errors="replace") + +import config +from meta_ads_client import MetaAdsClient +from baserow_client import BaserowClient +from agent import analyze_creative_deep, compare_adset_creatives +from slack_notifier import send_creative_analysis_report + + +def _get_max_cpl(campaign_name: str, verticals_data: dict) -> float: + name_lower = campaign_name.lower() + for vert_name, vert_cfg in verticals_data.items(): + if vert_name.lower() in name_lower: + return float(vert_cfg.get("target_cpl") or 0) + return float(config.META_TARGET_CPL or 6.0) + + +def main(): + parser = argparse.ArgumentParser(description="Deep creative analysis for Meta Ads") + parser.add_argument("--campaign", help="Filter campaigns by name substring (case-insensitive)") + parser.add_argument("--no-slack", action="store_true", help="Skip Slack report") + args = parser.parse_args() + + meta = MetaAdsClient() + db = BaserowClient() + + print(f"\n{'='*60}") + print(f" ANÁLISIS DE CREATIVIDADES — {datetime.now().strftime('%d/%m/%Y %H:%M')}") + print(f"{'='*60}\n") + + # Active campaigns (last 7 days) + campaigns = meta.get_period_campaign_metrics(7) + if args.campaign: + campaigns = {k: v for k, v in campaigns.items() + if args.campaign.upper() in v["name"].upper()} + + if not campaigns: + print("No hay campañas activas en los últimos 7 días.") + return + + print(f"→ {len(campaigns)} campañas a analizar\n") + + verticals_data = {v["Nombre"]: v for v in db.get_all_verticals()} + + all_results: dict = {} + total_analyzed = 0 + total_errors = 0 + + for cid, camp_metrics in campaigns.items(): + campaign_name = camp_metrics["name"] + max_cpl = _get_max_cpl(campaign_name, verticals_data) + + print(f" ● {campaign_name} (objetivo CPL: {max_cpl:.2f}€)") + + ads_with_creatives = meta.get_ads_with_creatives(cid) + if not ads_with_creatives: + print(" — sin anuncios activos con creatividades, omitiendo\n") + continue + + # Metrics for both windows + ads_7d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, 7)} + ads_3d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, 3)} + + # Adset name lookup from 7d metrics + adset_names = {a["id"]: a["name"] for a in meta.get_period_adset_metrics(cid, 7)} + + # Group ads by adset + adset_groups: dict = {} + for ad in ads_with_creatives: + if not ad.get("thumbnail_url"): + continue + ad_id = ad["ad_id"] + adset_id = ad.get("adset_id", "unknown") + adset_name = adset_names.get(adset_id, adset_id) + + m7 = ads_7d.get(ad_id, {}) + m3 = ads_3d.get(ad_id, {}) + metrics = { + "spend_7d": m7.get("spend", 0), + "leads_7d": m7.get("leads", 0), + "cpl_7d": m7.get("cpl", 0), + "ctr_7d": m7.get("ctr", 0), + "spend_3d": m3.get("spend", 0), + "leads_3d": m3.get("leads", 0), + "cpl_3d": m3.get("cpl", 0), + "ctr_3d": m3.get("ctr", 0), + "max_cpl": max_cpl, + } + + if adset_id not in adset_groups: + adset_groups[adset_id] = {"name": adset_name, "ads": []} + adset_groups[adset_id]["ads"].append({ + "ad_id": ad_id, + "ad_name": ad["ad_name"], + "campaign_id": cid, + "adset_id": adset_id, + "adset_name": adset_name, + # Fallback chain: signed thumbnail → permanent video picture → static image_url + "image_url": [ad["thumbnail_url"], ad["video_thumbnail_url"], ad["image_url"]], + **metrics, + }) + + # Analyze each ad individually + analyzed_adsets: dict = {} + for adset_id, adset_data in adset_groups.items(): + adset_name = adset_data["name"] + analyzed_ads = [] + + for ad in adset_data["ads"]: + short_name = ad["ad_name"][:50] + print(f" [{adset_name[:30]}] {short_name}...", end=" ", flush=True) + + result = analyze_creative_deep( + image_url=ad["image_url"], + ad_name=ad["ad_name"], + metrics={k: ad[k] for k in ( + "spend_7d", "leads_7d", "cpl_7d", "ctr_7d", + "spend_3d", "leads_3d", "cpl_3d", "ctr_3d", "max_cpl" + )}, + ) + + score = result.get("score", 0) + fatigue_flag = " ⚠️FATIGA" if result.get("fatigue") else "" + print(f"score={score:.1f}{fatigue_flag}") + + ad_result = {**ad, **result} + analyzed_ads.append(ad_result) + + # Save to Baserow + analysis_text = result.get("analysis", "") + if result.get("fatigue") and result.get("fatigue_reason"): + analysis_text += f"\n\n⚠️ FATIGA CREATIVA: {result['fatigue_reason']}" + + try: + urls = ad["image_url"] + saved_url = urls[0] if isinstance(urls, list) else urls + db.save_creative_analysis({ + "ad_id": ad["ad_id"], + "ad_name": ad["ad_name"], + "campaign_id": cid, + "image_url": saved_url, + "analysis": analysis_text, + "score": score, + "recommendations": result.get("recommendations", ""), + }) + total_analyzed += 1 + except Exception as e: + print(f" [WARN] Baserow: {e}") + total_errors += 1 + + time.sleep(0.3) + + # Comparative analysis for adsets with 2+ ads + comparison = None + if len(analyzed_ads) >= 2: + print(f" [comparativa] {adset_name[:40]}...", end=" ", flush=True) + comparison = compare_adset_creatives(analyzed_ads) + winner = comparison.get("winner", "") + print(f"ganador: {winner[:40]}") + + analyzed_adsets[adset_id] = { + "name": adset_name, + "ads": analyzed_ads, + "comparison": comparison, + } + + all_results[cid] = { + "name": campaign_name, + "max_cpl": max_cpl, + "adsets": analyzed_adsets, + } + print() + + # Summary + total_ads = sum(len(as_d["ads"]) for c in all_results.values() for as_d in c["adsets"].values()) + total_fatigue = sum( + 1 for c in all_results.values() + for as_d in c["adsets"].values() + for ad in as_d["ads"] if ad.get("fatigue") + ) + print(f"{'='*60}") + print(f" Finalizado: {len(all_results)} campañas, {total_ads} anuncios analizados") + if total_fatigue: + print(f" ⚠️ {total_fatigue} anuncios con fatiga creativa detectada") + if total_errors: + print(f" ⚠️ {total_errors} errores al guardar en Baserow") + print(f"{'='*60}\n") + + # Slack report + if not args.no_slack and all_results: + print("→ Enviando informe a Slack...") + send_creative_analysis_report(all_results) + print(" ✓ Informe enviado.") + + +if __name__ == "__main__": + main() diff --git a/baserow_client.py b/baserow_client.py index 6ac0488..530308d 100644 --- a/baserow_client.py +++ b/baserow_client.py @@ -116,6 +116,30 @@ class BaserowClient: # ── creative_analyses ───────────────────────────────────────────────────── + def get_all_creative_analyses(self, filters: dict = None) -> list: + """Returns creative analyses from Baserow, up to 200 rows.""" + params = {"user_field_names": "true", "page_size": 200, "order_by": "-created_at"} + if filters: + params.update(filters) + try: + resp = requests.get( + self._url(config.BASEROW_TABLE_CREATIVES), + headers=self._headers, params=params, timeout=15, + ) + if not resp.ok: + return [] + return resp.json().get("results", []) + except requests.RequestException: + return [] + + def get_creative_history_by_ad(self, ad_id: str) -> list: + """Returns all analyses for an ad_id sorted by date ascending (for score evolution).""" + rows = self._get_rows( + config.BASEROW_TABLE_CREATIVES, + {"filter__ad_id__equal": ad_id}, + ) + return sorted(rows, key=lambda r: r.get("created_at", "")) + def save_creative_analysis(self, analysis: dict) -> dict: return self._create_row(config.BASEROW_TABLE_CREATIVES, { "ad_id": analysis["ad_id"], diff --git a/dashboard.py b/dashboard.py index b155cf7..4f7fc05 100644 --- a/dashboard.py +++ b/dashboard.py @@ -170,7 +170,7 @@ st.divider() # ── Tabs ────────────────────────────────────────────────────────────────────── -tab1, tab2, tab3, tab4 = st.tabs(["📅 Por día", "📊 Campañas", "🏷️ Verticales", "🗂️ Histórico"]) +tab1, tab2, tab3, tab4, tab5 = st.tabs(["📅 Por día", "📊 Campañas", "🏷️ Verticales", "🗂️ Histórico", "🎨 Creatividades"]) # ── Tab 1: Por día ──────────────────────────────────────────────────────────── @@ -473,3 +473,131 @@ with tab4: 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.") diff --git a/meta_ads_client.py b/meta_ads_client.py index 7dfe912..1a922c6 100644 --- a/meta_ads_client.py +++ b/meta_ads_client.py @@ -9,6 +9,7 @@ from facebook_business.adobjects.campaign import Campaign from facebook_business.adobjects.adset import AdSet from facebook_business.adobjects.ad import Ad from facebook_business.adobjects.adcreative import AdCreative +from facebook_business.adobjects.advideo import AdVideo import config from datetime import datetime, timedelta @@ -209,30 +210,45 @@ class MetaAdsClient: """ campaign = Campaign(campaign_id) ads = campaign.get_ads( - fields=[Ad.Field.id, Ad.Field.name, Ad.Field.status, Ad.Field.creative], + fields=[Ad.Field.id, Ad.Field.name, Ad.Field.status, Ad.Field.creative, "adset_id"], params={"effective_status": ["ACTIVE"]}, ) result = [] for ad in ads: - creative_ref = ad.get("creative", {}) - creative_id = creative_ref.get("id") if creative_ref else None - thumbnail = "" + creative_ref = ad.get("creative", {}) + creative_id = creative_ref.get("id") if creative_ref else None + thumbnail_url = "" + image_url = "" + video_thumbnail_url = "" if creative_id: try: creative = AdCreative(creative_id).api_get( - fields=["thumbnail_url", "image_url"] + fields=["thumbnail_url", "image_url", "video_id"] ) - thumbnail = creative.get("thumbnail_url") or creative.get("image_url", "") + thumbnail_url = creative.get("thumbnail_url", "") + image_url = creative.get("image_url", "") + video_id = creative.get("video_id", "") + + # For video creatives: fetch a permanent thumbnail via AdVideo.picture + if video_id and not image_url: + try: + vdata = AdVideo(video_id).api_get(fields=["picture"]) + video_thumbnail_url = vdata.get("picture", "") + except Exception: + pass except Exception: pass result.append({ - "ad_id": ad["id"], - "ad_name": ad["name"], - "campaign_id": campaign_id, - "thumbnail_url": thumbnail, + "ad_id": ad["id"], + "ad_name": ad["name"], + "campaign_id": campaign_id, + "adset_id": ad.get("adset_id", ""), + "thumbnail_url": thumbnail_url, + "image_url": image_url, + "video_thumbnail_url": video_thumbnail_url, }) return result diff --git a/slack_notifier.py b/slack_notifier.py index 3be0067..728e302 100644 --- a/slack_notifier.py +++ b/slack_notifier.py @@ -157,14 +157,12 @@ def _adset_ad_table(items: list, label: str, show_bid: bool = False, show_7d: bo return "\n".join(lines) -def _vertical_status_emoji(v_cpl: float, v_obj: float, has_action: bool) -> str: +def _vert_status(v_cpl: float, v_obj: float, has_issues: bool) -> str: if v_cpl == 0 or v_obj == 0: return "⚪" - if v_cpl <= v_obj: - return "⚠️" if has_action else "✅" - if v_cpl <= v_obj * 1.3: - return "⚠️" - return "❌" + if has_issues: + return "🚨" if v_cpl > v_obj * 1.3 else "⚠️" + return "✅" if v_cpl <= v_obj else "⚠️" def send_daily_report( @@ -186,33 +184,34 @@ def send_daily_report( prefix = config.META_CAMPAIGN_PREFIX mode_label = "DRY RUN" if mode == "DRY_RUN" else "PRODUCCIÓN" - action_map = {a["campaign_name"]: a for a in actions} - details_map = campaign_details or {} - name_to_cid = {d["name"]: cid for cid, d in details_map.items()} - - # ── Classify campaigns: issues vs OK ───────────────────────────────────── - camps_issues: dict = {} # name → (cid, detail, action_or_None) - camps_ok: dict = {} # name → (cid, detail) + action_map = {a["campaign_name"]: a for a in actions} + details_map = campaign_details or {} + # Group ALL campaigns by vertical + by_vertical: dict = {} for cid, detail in details_map.items(): - name = detail["name"] - act = action_map.get(name) - has_action = bool(act and act["action_type"] != "MAINTAIN") - has_ad_pause = any( - ad.get("accion") == "PAUSE" and ad.get("row_id") - for ad in detail.get("ads", []) + act = action_map.get(detail["name"]) + by_vertical.setdefault(detail["vertical"], []).append((cid, detail, act)) + + def _has_issues(camp_list): + return any( + (act and act["action_type"] != "MAINTAIN") or + any(ad.get("accion") == "PAUSE" and ad.get("row_id") + for ad in detail.get("ads", [])) + for _, detail, act in camp_list ) - if has_action or has_ad_pause: - camps_issues[name] = (cid, detail, act) - else: - camps_ok[name] = (cid, detail) - # Group issues by vertical - by_vertical_issues: dict = {} - for name, (cid, detail, act) in camps_issues.items(): - by_vertical_issues.setdefault(detail["vertical"], []).append((cid, detail, act)) + # Sort verticals: issues first (by margin asc), then OK (by margin desc) + def _vert_sort_key(item): + v, cl = item + v_data = (verticals or {}).get(v, {}) + margin = v_data.get("margin", 0) + has_iss = _has_issues(cl) + return (0 if has_iss else 1, margin if has_iss else -margin) - # ── Message 1: Executive dashboard ─────────────────────────────────────── + sorted_verticals = sorted(by_vertical.items(), key=_vert_sort_key) + + # ── Message 1: Dashboard ───────────────────────────────────────────────── blocks: list = [ { "type": "header", @@ -228,12 +227,12 @@ def send_daily_report( if monthly_verticals else [] ) cw = 7 - header = f"{'Día':<5} {'Gasto':>6} {'Leads':>5} {'CPL':>7}" + hdr = f"{'Día':<5} {'Gasto':>6} {'Leads':>5} {'CPL':>7}" for v in v_order: - header += f" {v[:6]:>{cw}}" - header += " Est" - sep = "─" * len(header) - lines = [header, sep] + hdr += f" {v[:6]:>{cw}}" + hdr += " Est" + sep = "─" * len(hdr) + lines = [hdr, sep] total_spend = total_leads = total_margin = 0.0 total_v = {v: 0.0 for v in v_order} for d in daily_totals: @@ -276,111 +275,31 @@ def send_daily_report( # Vertical scorecard if verticals: - lines = [f"{'Vertical':<16} {'Gasto':>6} {'Leads':>5} {'CPL':>7} {'Obj':>7} {'Margen':>9}"] - lines.append("─" * 58) - for v, data in sorted(verticals.items(), key=lambda x: -x[1]["margin"]): - v_leads = data["leads"] - v_spend = data["spend"] + lines = [f"{'':>2} {'Vertical':<14} {'Gasto':>6} {'Leads':>5} {'CPL':>7} {'Obj':>7} {'Margen':>9}"] + lines.append("─" * 60) + for v, cl in sorted_verticals: + data = (verticals or {}).get(v, {}) + v_leads = data.get("leads", 0) + v_spend = data.get("spend", 0) v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 - v_m = data["margin"] + v_m = data.get("margin", 0) v_obj = data.get("target_cpl", 0) m_sign = f"+{v_m:.0f}€" if v_m >= 0 else f"{v_m:.0f}€" obj_str = f"{v_obj:.2f}€" if v_obj else " —" - has_act = v in by_vertical_issues - st = _vertical_status_emoji(v_cpl, v_obj, has_act) + st = _vert_status(v_cpl, v_obj, _has_issues(cl)) lines.append( f"{st} {v:<14} {v_spend:>5.0f}€ {v_leads:>5} {v_cpl:>6.2f}€ {obj_str:>7} {m_sign:>9}" ) blocks.append({ "type": "section", "text": {"type": "mrkdwn", - "text": "*Verticales · ayer*\n```" + "\n".join(lines) + "```"}, - }) - - # Actions section - if actions: - blocks.append({"type": "divider"}) - blocks.append({ - "type": "section", - "text": {"type": "mrkdwn", "text": "*🚨 Acciones recomendadas*"}, - }) - for act in actions: - atype = act["action_type"] - emoji, alabel = _ACTION_DISPLAY.get(atype, ("⚪", atype)) - cid = name_to_cid.get(act["campaign_name"]) - bid_cfg = details_map.get(cid, {}).get("bid_config", {}) if cid else {} - budget = bid_cfg.get("daily_budget_eur") - text = f"{emoji} *{act['campaign_name'][:60]}* → *{alabel}*" - if act.get("justification"): - text += f"\n_{act['justification'][:200]}_" - if act.get("alert"): - text += f"\n:warning: {act['alert'][:150]}" - effect = _effect_text(act, budget) - if effect: - text += f"\n{effect}" - blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": text}}) - if atype in _ACTIONABLE: - blocks.append({ - "type": "actions", - "elements": [ - { - "type": "button", - "text": {"type": "plain_text", "text": "✅ Aprobar"}, - "style": "primary", - "value": f"approve:{act['row_id']}", - "action_id": f"approve_{act['row_id']}", - }, - { - "type": "button", - "text": {"type": "plain_text", "text": "❌ Rechazar"}, - "style": "danger", - "value": f"reject:{act['row_id']}", - "action_id": f"reject_{act['row_id']}", - }, - ], - }) - - # Ad pause count notice - total_ad_pauses = sum( - 1 for _, detail, _ in camps_issues.values() - for ad in detail.get("ads", []) - if ad.get("accion") == "PAUSE" and ad.get("row_id") - ) - if total_ad_pauses > 0: - noun = "anuncio" if total_ad_pauses == 1 else "anuncios" - blocks.append({ - "type": "context", - "elements": [{"type": "mrkdwn", - "text": f"⛔ {total_ad_pauses} {noun} recomendados para pausar — ver detalle por vertical"}], - }) - - # "Todo en orden" summary — grouped by vertical - if camps_ok: - blocks.append({"type": "divider"}) - ok_by_v: dict = {} - for name, (_, detail) in camps_ok.items(): - ok_by_v.setdefault(detail["vertical"], []).append((name, detail)) - ok_lines = ["*Todo en orden*"] - for vert in sorted(ok_by_v.keys()): - v_data = (verticals or {}).get(vert, {}) - v_spend = v_data.get("spend", 0) - v_leads = v_data.get("leads", 0) - v_cpl = round(v_spend / v_leads, 2) if v_leads > 0 else 0.0 - v_obj = v_data.get("target_cpl", 0) - ok_lines.append(f"\n✅ *{vert.upper()}* · CPL {v_cpl:.2f}€ obj {v_obj:.2f}€") - for name, detail in sorted(ok_by_v[vert], key=lambda x: -x[1].get("spend_1d", 0)): - ok_lines.append( - f" • {name} · " - f"{detail.get('spend_1d', 0):.0f}€ / {detail.get('leads_1d', 0)} leads ayer" - ) - blocks.append({ - "type": "section", - "text": {"type": "mrkdwn", "text": "\n".join(ok_lines)}, + "text": "*Resumen · ayer*\n```" + "\n".join(lines) + "```"}, }) blocks.append({ "type": "context", - "elements": [{"type": "mrkdwn", "text": f"{campaigns_analyzed} campañas analizadas"}], + "elements": [{"type": "mrkdwn", + "text": f"{campaigns_analyzed} campañas analizadas — detalle por vertical a continuación"}], }) result = _post( @@ -391,8 +310,8 @@ def send_daily_report( ) ts = result.get("ts") - # ── Messages 2-N: one per vertical with issues ──────────────────────────── - for v, camp_list in sorted(by_vertical_issues.items()): + # ── One message per vertical ────────────────────────────────────────────── + for v, camp_list in sorted_verticals: v_data = (verticals or {}).get(v, {}) v_spend = v_data.get("spend", 0) v_leads = v_data.get("leads", 0) @@ -401,11 +320,13 @@ def send_daily_report( v_margin = v_data.get("margin", 0) m_str = f"+{v_margin:.0f}€" if v_margin >= 0 else f"{v_margin:.0f}€" obj_str = f" · obj {v_obj:.2f}€" if v_obj else "" + has_iss = _has_issues(camp_list) + st = _vert_status(v_cpl, v_obj, has_iss) v_blocks: list = [ { "type": "header", - "text": {"type": "plain_text", "text": f"📊 {v.upper()}"}, + "text": {"type": "plain_text", "text": f"{st} {v.upper()}"}, }, { "type": "section", @@ -417,6 +338,7 @@ def send_daily_report( ), }, }, + {"type": "divider"}, ] for i, (cid, detail, act) in enumerate( @@ -424,79 +346,205 @@ def send_daily_report( ): if i > 0: v_blocks.append({"type": "divider"}) - name = detail["name"] - spend_1d = detail.get("spend_1d", 0.0) - leads_1d = detail.get("leads_1d", 0) - margin = detail["margin"] - m_str2 = f"+{margin:.2f}€" if margin >= 0 else f"{margin:.2f}€" - adsets = detail.get("adsets", []) - ads = detail.get("ads", []) - bid_cfg = detail.get("bid_config", {}) - budget = bid_cfg.get("daily_budget_eur") - strategy = bid_cfg.get("bid_strategy", "") + + name = detail["name"] + spend_1d = detail.get("spend_1d", 0.0) + leads_1d = detail.get("leads_1d", 0) + margin = detail["margin"] + m_str2 = f"+{margin:.2f}€" if margin >= 0 else f"{margin:.2f}€" + adsets = detail.get("adsets", []) + ads = detail.get("ads", []) + bid_cfg = detail.get("bid_config", {}) + budget = bid_cfg.get("daily_budget_eur") + strategy = bid_cfg.get("bid_strategy", "") strat_label = _STRATEGY_LABELS.get(strategy, strategy or "—") + atype = act["action_type"] if act else "MAINTAIN" + cemoji, alabel = _ACTION_DISPLAY.get(atype, ("⚪", atype)) - atype = act["action_type"] if act else "MAINTAIN" - emoji, alabel = _ACTION_DISPLAY.get(atype, ("⚪", atype)) - - camp_text = ( - f"{emoji} *{name}*\n" - f"Ayer: {spend_1d:.0f}€ / {leads_1d} leads · Margen: {m_str2} · " - f"`{strat_label}`" + (f" · {budget:.0f}€/día" if budget else "") + - f"\n*{alabel}*" - ) - if act and act.get("justification"): - camp_text += f" — _{act['justification'][:160]}_" - if act and act.get("alert"): - camp_text += f"\n:warning: {act['alert'][:130]}" - - v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": camp_text}}) - - # Approve/Reject buttons for campaign action - if act and atype in _ACTIONABLE: - effect = _effect_text(act, budget) - if effect: - v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": effect}}) + if atype == "MAINTAIN" and not any( + ad.get("accion") == "PAUSE" and ad.get("row_id") for ad in ads + ): + # Compact header for clean campaigns v_blocks.append({ - "type": "actions", - "elements": [ - { - "type": "button", - "text": {"type": "plain_text", "text": "✅ Aprobar"}, - "style": "primary", - "value": f"approve:{act['row_id']}", - "action_id": f"approve_{act['row_id']}", - }, - { - "type": "button", - "text": {"type": "plain_text", "text": "❌ Rechazar"}, - "style": "danger", - "value": f"reject:{act['row_id']}", - "action_id": f"reject_{act['row_id']}", - }, - ], + "type": "section", + "text": { + "type": "mrkdwn", + "text": ( + f"{cemoji} *{name}*\n" + f"Ayer: {spend_1d:.0f}€ / {leads_1d} leads · " + f"Margen: {m_str2} · `{strat_label}`" + + (f" · {budget:.0f}€/día" if budget else "") + ), + }, }) + # Still show adset breakdown for context + if adsets: + tbl = _adset_ad_table(adsets[:3], "Conjuntos (3 días)", show_bid=True) + if tbl: + for chunk in [tbl[j:j+2900] for j in range(0, len(tbl), 2900)]: + v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": chunk}}) + else: + # Full block for campaigns with action or ad pauses + camp_text = ( + f"{cemoji} *{name}*\n" + f"Ayer: {spend_1d:.0f}€ / {leads_1d} leads · Margen: {m_str2} · " + f"`{strat_label}`" + (f" · {budget:.0f}€/día" if budget else "") + + f"\n*{alabel}*" + ) + if act and act.get("justification"): + camp_text += f" — _{act['justification'][:160]}_" + if act and act.get("alert"): + camp_text += f"\n:warning: {act['alert'][:130]}" + v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": camp_text}}) - # Adset table — top 3, only for non-MAINTAIN campaigns - if atype != "MAINTAIN" and adsets: - adset_table = _adset_ad_table(adsets[:3], "Conjuntos de anuncios (3 días)", show_bid=True) - if adset_table: - for chunk in [adset_table[i:i+2900] for i in range(0, len(adset_table), 2900)]: - v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": chunk}}) + # Approve/Reject buttons + if act and atype in _ACTIONABLE: + effect = _effect_text(act, budget) + if effect: + v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": effect}}) + v_blocks.append({ + "type": "actions", + "elements": [ + { + "type": "button", + "text": {"type": "plain_text", "text": "✅ Aprobar"}, + "style": "primary", + "value": f"approve:{act['row_id']}", + "action_id": f"approve_{act['row_id']}", + }, + { + "type": "button", + "text": {"type": "plain_text", "text": "❌ Rechazar"}, + "style": "danger", + "value": f"reject:{act['row_id']}", + "action_id": f"reject_{act['row_id']}", + }, + ], + }) - # Ad pause buttons - v_blocks.extend(_ad_action_blocks(ads)) + # Adset table (top 3) — only for non-MAINTAIN + if atype != "MAINTAIN" and adsets: + tbl = _adset_ad_table(adsets[:3], "Conjuntos (3 días)", show_bid=True) + if tbl: + for chunk in [tbl[j:j+2900] for j in range(0, len(tbl), 2900)]: + v_blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": chunk}}) + + # Ad pause buttons + v_blocks.extend(_ad_action_blocks(ads)) _post( "chat.postMessage", channel=config.SLACK_CHANNEL_ID, blocks=v_blocks, - text=f"Detalle vertical: {v}", + text=f"{v.upper()} · {v_spend:.0f}€ · {v_leads} leads", ) return ts +def _score_emoji(score: float) -> str: + if score >= 8: return "🟢" + if score >= 6: return "🟡" + if score >= 4: return "🟠" + return "🔴" + + +def _brief_action(score: float, fatigue: bool) -> str: + if score == 0: return "sin imagen" + if fatigue: return "renovar urgente" + if score >= 8: return "mantener" + if score >= 6: return "optimizar" + if score >= 4: return "renovar" + return "reemplazar" + + +def send_creative_analysis_report(all_results: dict) -> None: + """Envía scorecard compacto de creatividades a Slack. Un mensaje por campaña.""" + now = datetime.now() + date_label = now.strftime("%d/%m/%Y %H:%M") + + total_ads = sum(len(as_d["ads"]) for c in all_results.values() for as_d in c["adsets"].values()) + total_fatigue = sum( + 1 for c in all_results.values() + for as_d in c["adsets"].values() + for ad in as_d["ads"] if ad.get("fatigue") + ) + scored = [ + ad.get("score", 0) for c in all_results.values() + for as_d in c["adsets"].values() + for ad in as_d["ads"] if ad.get("score", 0) > 0 + ] + avg_score = round(sum(scored) / len(scored), 1) if scored else 0.0 + + summary = f"*{len(all_results)} campañas* · *{total_ads} anuncios* · score medio *{avg_score}/10*" + if total_fatigue: + summary += f"\n⚠️ *{total_fatigue} con fatiga creativa detectada*" + + _post( + "chat.postMessage", + channel=config.SLACK_CHANNEL_ID, + blocks=[ + {"type": "header", "text": {"type": "plain_text", "text": f"Creatividades — {date_label}"}}, + {"type": "section", "text": {"type": "mrkdwn", "text": summary}}, + ], + text=f"Creatividades — {date_label}", + ) + + # ── One message per campaign ────────────────────────────────────────────── + for cid, camp_data in all_results.items(): + camp_name = camp_data["name"] + adsets = camp_data.get("adsets", {}) + if not adsets: + continue + + def _flush(buf: list) -> list: + if len(buf) > 1: + try: + _post("chat.postMessage", channel=config.SLACK_CHANNEL_ID, + blocks=buf, text=camp_name) + except RuntimeError as e: + print(f" [WARN] Slack: {e}") + return [{"type": "header", "text": {"type": "plain_text", "text": f"{camp_name} (cont.)"}}] + + blocks: list = [{"type": "header", "text": {"type": "plain_text", "text": camp_name}}] + + for as_data in adsets.values(): + adset_name = as_data["name"] + ads = as_data["ads"] + ads_sorted = sorted(ads, key=lambda x: -x.get("score", 0)) + + # Compact monospace table — one line per ad + lines = [ + f"*{adset_name}* _({len(ads)} anuncios)_", + "```", + f"{'Nombre':<33} {'Sc':>4} {'CTR':>5} {'CPL':>6} Acción", + "─" * 63, + ] + for ad in ads_sorted: + name = _table_name(ad["ad_name"], 33) + score = ad.get("score", 0) + ctr = ad.get("ctr_7d", 0) + cpl = ad.get("cpl_7d", 0) + fat = "⚠" if ad.get("fatigue") else " " + action = _brief_action(score, ad.get("fatigue", False)) + cpl_s = f"{cpl:.2f}€" if cpl > 0 else " —" + sc_s = f"{score:.1f}" if score > 0 else " —" + lines.append(f"{name:<33}{fat} {sc_s:>4} {ctr:>4.1f}% {cpl_s:>6} {action}") + lines.append("```") + + winner = as_data.get("comparison", {}) or {} + if winner.get("winner"): + lines.append(f"🏆 _{winner['winner'][:70]}_") + + ab = [{"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}}] + + if len(blocks) + len(ab) > 48: + blocks = _flush(blocks) + blocks.extend(ab) + + _flush(blocks) + + def send_execution_summary(log: dict): """Resumen plano de ejecución (fallback).""" mode_label = "DRY RUN" if log.get("mode") == "DRY_RUN" else "PRODUCCIÓN"