diff --git a/.env.example b/.env.example index 8f1cbd0..5022fa1 100644 --- a/.env.example +++ b/.env.example @@ -1,10 +1,27 @@ -AIRTABLE_TOKEN=your_airtable_token -AIRTABLE_BASE_ID=your_base_id - +# Meta Ads META_APP_ID=your_app_id META_APP_SECRET=your_app_secret META_ACCESS_TOKEN=your_long_lived_access_token META_AD_ACCOUNT_ID=act_XXXXXXXXXX +# Anthropic ANTHROPIC_API_KEY=your_anthropic_key -SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... + +# Baserow (self-hosted) +BASEROW_URL=https://baserow.yourdomain.com +BASEROW_TOKEN=your_baserow_api_token +# Run setup_baserow.py once to get the IDs below: +BASEROW_TABLE_CAMPAIGNS=0 +BASEROW_TABLE_ACTIONS=0 +BASEROW_TABLE_CREATIVES=0 +BASEROW_TABLE_LOGS=0 +BASEROW_TABLE_VERTICALS=0 +BASEROW_TABLE_SNAPSHOTS=0 + +# Slack App +SLACK_BOT_TOKEN=xoxb-your-bot-token +SLACK_SIGNING_SECRET=your_signing_secret +SLACK_CHANNEL_ID=C0XXXXXXXXX + +# Operation (set to false in production to actually execute actions) +DRY_RUN=true diff --git a/.github/workflows/daily.yml b/.github/workflows/daily.yml index 35a0cd9..9216009 100644 --- a/.github/workflows/daily.yml +++ b/.github/workflows/daily.yml @@ -1,8 +1,8 @@ name: Daily Meta Optimizer on: - schedule: - - cron: '0 6 * * *' # 8:00 AM hora española (CEST/UTC+2) + # schedule: + # - cron: '0 5 * * *' # 7:00 AM hora española (CEST/UTC+2) — pausado temporalmente workflow_dispatch: jobs: @@ -23,14 +23,21 @@ jobs: - name: Run optimizer env: - AIRTABLE_TOKEN: ${{ secrets.AIRTABLE_TOKEN }} - AIRTABLE_BASE_ID: ${{ secrets.AIRTABLE_BASE_ID }} META_APP_ID: ${{ secrets.META_APP_ID }} META_APP_SECRET: ${{ secrets.META_APP_SECRET }} META_ACCESS_TOKEN: ${{ secrets.META_ACCESS_TOKEN }} META_AD_ACCOUNT_ID: ${{ secrets.META_AD_ACCOUNT_ID }} ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} - SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }} + BASEROW_URL: ${{ secrets.BASEROW_URL }} + BASEROW_TOKEN: ${{ secrets.BASEROW_TOKEN }} + BASEROW_TABLE_CAMPAIGNS: ${{ secrets.BASEROW_TABLE_CAMPAIGNS }} + BASEROW_TABLE_ACTIONS: ${{ secrets.BASEROW_TABLE_ACTIONS }} + BASEROW_TABLE_CREATIVES: ${{ secrets.BASEROW_TABLE_CREATIVES }} + BASEROW_TABLE_LOGS: ${{ secrets.BASEROW_TABLE_LOGS }} + SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }} + SLACK_SIGNING_SECRET: ${{ secrets.SLACK_SIGNING_SECRET }} + SLACK_CHANNEL_ID: ${{ secrets.SLACK_CHANNEL_ID }} + DRY_RUN: ${{ vars.DRY_RUN }} run: python run.py - name: Upload log diff --git a/agent.py b/agent.py index 64169c1..79a3846 100644 --- a/agent.py +++ b/agent.py @@ -1,33 +1,31 @@ import json +import base64 +import requests import anthropic import config client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY) -SYSTEM_PROMPT = """ -Eres un experto en optimización de campañas de Meta Ads (Facebook/Instagram) para generación de leads. -Cada campaña corresponde a un producto o curso con un CPL objetivo (coste por lead) acordado. - -MODELO DE NEGOCIO: -- El objetivo es maximizar el volumen de leads por debajo del CPL máximo rentable. -- La frecuencia alta puede indicar saturación de audiencia. -- El CTR y CPM son indicadores clave de relevancia creativa y competencia en subasta. +DECIDE_SYSTEM = """ +Eres un experto en optimización de campañas de Meta Ads para generación de leads. +Cada campaña tiene un CPL máximo (coste por lead objetivo) que define el límite aceptable. +USA SIEMPRE € como unidad de moneda. Responde SIEMPRE en español. REGLAS DE DECISIÓN: -1. CPL > CPL_máximo → REDUCIR_PRESUPUESTO o revisar creatividades/audiencias. -2. CPL <= CPL_máximo y volumen bajo → AUMENTAR_PRESUPUESTO si hay margen. -3. Frecuencia > 3.0 → considerar rotar creatividades o ampliar audiencia. -4. CTR < 1% → problema creativo, revisar anuncios. -5. Sin leads tras 3+ días de gasto → revisar configuración de conversión. +1. CPL > max_cpl → REDUCE_BUDGET o revisar creatividades/audiencias. +2. CPL <= max_cpl con bajo volumen → INCREASE_BUDGET si hay margen. +3. Frecuencia > 3.0 → considera rotar creatividades o ampliar audiencia. +4. CTR < 1% → problema de creatividad, revisar anuncios. +5. Sin leads tras varios días de inversión → revisar configuración de conversiones. -Devuelve ÚNICAMENTE un JSON válido con esta estructura exacta, sin texto adicional ni markdown: +Responde SOLO con JSON válido, sin texto adicional ni markdown: { - "accion": "PAUSAR | REDUCIR_PRESUPUESTO | AUMENTAR_PRESUPUESTO | MANTENER | REVISAR_CREATIVIDADES", - "parametro": 1.0, - "justificacion": "explicación breve", - "consejo": "acción concreta y específica", - "alerta": "texto si hay algo crítico, null si no", - "confianza": 0.0 + "action": "PAUSE | REDUCE_BUDGET | INCREASE_BUDGET | MAINTAIN | REVIEW_CREATIVES", + "parameter": 1.0, + "justification": "explicación breve en español usando €", + "advice": "acción concreta y específica a realizar", + "alert": "texto crítico si lo hay, null si no", + "confidence": 0.0 } """ @@ -36,12 +34,12 @@ def decide(analysis: dict) -> dict: response = client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=400, - system=SYSTEM_PROMPT, + system=DECIDE_SYSTEM, messages=[{ "role": "user", "content": ( - f"Analiza esta campaña de Meta Ads y devuelve la decisión en JSON:\n\n" - f"{json.dumps(analysis, ensure_ascii=False, indent=2)}" + "Analyze this Meta Ads campaign and return the decision as JSON:\n\n" + + json.dumps(analysis, ensure_ascii=False, indent=2) ), }], ) @@ -50,10 +48,124 @@ def decide(analysis: dict) -> dict: try: return json.loads(clean) except json.JSONDecodeError: + import re as _re + m = _re.search(r"\{.*\}", clean, _re.DOTALL) + if m: + try: + return json.loads(m.group()) + except json.JSONDecodeError: + pass return { - "accion": "MANTENER", - "parametro": 1.0, - "justificacion": "Error parseando respuesta del agente.", - "alerta": f"JSON inválido: {raw[:200]}", - "confianza": 0.0, + "action": "MAINTAIN", + "parameter": 1.0, + "justification": "Error parsing agent response.", + "advice": "", + "alert": f"Invalid JSON: {raw[:200]}", + "confidence": 0.0, } + + +UNIT_SYSTEM = """ +Eres un analista experto en Meta Ads. Analiza las métricas del conjunto de anuncios o anuncio indicado. +USA SIEMPRE € como unidad de moneda. Responde SIEMPRE en español. +Si el conjunto tiene cost_cap_eur (cap de coste), compara el CPL actual con ese cap e indica si está +por encima, dentro o por debajo del límite, y cuánto margen queda (o cuánto se supera). +Responde SOLO con JSON válido (sin markdown): +{ + "evaluacion": "resumen del rendimiento en 2 frases usando €", + "recomendacion": "una acción concreta y específica para mejorar" +} +""" + + +def analyze_unit(metrics: dict, level: str = "adset") -> dict: + """Análisis rápido de un conjunto de anuncios o anuncio individual (solo análisis, sin acción).""" + nivel = "conjunto de anuncios" if level == "adset" else "anuncio" + response = client.messages.create( + model="claude-haiku-4-5-20251001", + max_tokens=200, + system=UNIT_SYSTEM, + messages=[{ + "role": "user", + "content": f"Analiza este {nivel} de Meta Ads:\n" + json.dumps(metrics, ensure_ascii=False), + }], + ) + raw = response.content[0].text.strip() + import re + clean = re.sub(r"```json\s*", "", raw) + clean = re.sub(r"```\s*", "", clean).strip() + clean = clean.replace("“", '"').replace("”", '"') # normalize smart quotes + # Strategy 1: direct parse + try: + return json.loads(clean) + except json.JSONDecodeError: + pass + # Strategy 2: extract first JSON object by brace boundaries + start, end = clean.find("{"), clean.rfind("}") + if start != -1 and end > start: + try: + return json.loads(clean[start:end + 1]) + except json.JSONDecodeError: + pass + # Strategy 3: extract fields individually with regex + ev_m = re.search(r'"evaluacion"\s*:\s*"((?:[^"\\]|\\.)*)"', clean) + rec_m = re.search(r'"recomendacion"\s*:\s*"((?:[^"\\]|\\.)*)"', clean) + if ev_m or rec_m: + return { + "evaluacion": ev_m.group(1) if ev_m else "", + "recomendacion": rec_m.group(1) if rec_m else "", + } + return {"evaluacion": clean[:150], "recomendacion": ""} + + +CREATIVE_SYSTEM = """ +You are an expert in Meta Ads creative analysis. +Analyze the provided ad image and return ONLY valid JSON without markdown: +{ + "score": 7.5, + "analysis": "concise analysis of the visual: messaging, design, call-to-action", + "recommendations": "concrete improvements to optimize CTR and conversions" +} +Score from 1 (very poor) to 10 (excellent). +""" + + +def analyze_creative(image_url: str, ad_name: str) -> dict: + try: + resp = requests.get(image_url, timeout=15) + resp.raise_for_status() + image_data = base64.standard_b64encode(resp.content).decode("utf-8") + media_type = resp.headers.get("content-type", "image/jpeg").split(";")[0] + except Exception as e: + return {"score": 0, "analysis": f"Failed to download image: {e}", "recommendations": ""} + + try: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=600, + system=CREATIVE_SYSTEM, + messages=[{ + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": media_type, + "data": image_data, + }, + }, + { + "type": "text", + "text": f'Ad name: "{ad_name}". Analyze this creative.', + }, + ], + }], + ) + raw = response.content[0].text.strip() + clean = raw.replace("```json", "").replace("```", "").strip() + return json.loads(clean) + except json.JSONDecodeError: + return {"score": 0, "analysis": "Error parsing creative analysis.", "recommendations": ""} + except Exception as e: + return {"score": 0, "analysis": f"Creative analysis failed: {e}", "recommendations": ""} diff --git a/approval_server.py b/approval_server.py new file mode 100644 index 0000000..1f3c693 --- /dev/null +++ b/approval_server.py @@ -0,0 +1,84 @@ +""" +FastAPI server that receives Slack interactive button callbacks (Approve / Reject). + +Setup: +1. Create a Slack App, enable Interactivity, set Request URL to: + https://your-domain.com/slack/actions +2. Set SLACK_SIGNING_SECRET in your .env +3. Run: uvicorn approval_server:app --host 0.0.0.0 --port 3000 + (for local dev: use ngrok to expose port 3000) +""" +import hashlib +import hmac +import json +import time +import urllib.parse + +from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import JSONResponse + +import config +from baserow_client import BaserowClient +import slack_notifier + +app = FastAPI() +baserow = BaserowClient() + + +def _verify_slack_signature(body: bytes, timestamp: str, signature: str) -> bool: + if abs(time.time() - int(timestamp)) > 300: + return False + basestring = f"v0:{timestamp}:{body.decode()}" + computed = "v0=" + hmac.new( + config.SLACK_SIGNING_SECRET.encode(), + basestring.encode(), + hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(computed, signature) + + +@app.post("/slack/actions") +async def slack_actions(request: Request): + body = await request.body() + timestamp = request.headers.get("X-Slack-Request-Timestamp", "0") + signature = request.headers.get("X-Slack-Signature", "") + + if not _verify_slack_signature(body, timestamp, signature): + raise HTTPException(status_code=401, detail="Invalid Slack signature") + + form = urllib.parse.parse_qs(body.decode()) + payload = json.loads(form.get("payload", ["{}"])[0]) + + actions = payload.get("actions", []) + if not actions: + return JSONResponse({"ok": True}) + + action = actions[0] + value = action.get("value", "") # "approve:42" or "reject:42" + channel = payload.get("channel", {}).get("id", config.SLACK_CHANNEL_ID) + message_ts = payload.get("message", {}).get("ts") + + try: + verb, row_id_str = value.split(":", 1) + row_id = int(row_id_str) + except ValueError: + raise HTTPException(status_code=400, detail=f"Unexpected action value: {value}") + + if verb == "approve": + baserow.update_action_status(row_id, "approved") + status_text = "aprobada" + elif verb == "reject": + baserow.update_action_status(row_id, "rejected") + status_text = "rechazada" + else: + raise HTTPException(status_code=400, detail=f"Unknown verb: {verb}") + + if message_ts: + user = payload.get("user", {}).get("name", "unknown") + slack_notifier.update_message( + channel, + message_ts, + f"Accion {status_text} por {user}.", + ) + + return JSONResponse({"ok": True}) diff --git a/backfill.py b/backfill.py new file mode 100644 index 0000000..c0990ba --- /dev/null +++ b/backfill.py @@ -0,0 +1,195 @@ +""" +Backfill: genera snapshots históricos con análisis Claude para un rango de fechas. + +Uso: + python backfill.py # mes en curso → ayer + python backfill.py --from 2026-06-01 --to 2026-06-04 + python backfill.py --skip-existing # no reprocesa días ya guardados +""" +import sys +import io +import argparse +sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) + +from datetime import datetime, timedelta +import config +from meta_ads_client import MetaAdsClient +from agent import decide, analyze_unit +from baserow_client import BaserowClient + +_ACTION_MAP = { + "PAUSE": "PAUSE", "REDUCE_BUDGET": "REDUCE_BUDGET", + "INCREASE_BUDGET": "INCREASE_BUDGET", "MAINTAIN": "MAINTAIN", + "REVIEW_CREATIVES": "REVIEW_CREATIVES", + "PAUSAR": "PAUSE", "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", + "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", "MANTENER": "MAINTAIN", + "REVISAR_CREATIVIDADES": "REVIEW_CREATIVES", +} + +def _extract_vertical(name: str) -> str: + 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" + + +def run_backfill(date_from: str, date_to: str, skip_existing: bool = False): + meta = MetaAdsClient() + baserow = BaserowClient() + + # Vertical CPL targets + 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 as e: + print(f"Warning: could not fetch verticals: {e}") + + # Build date list + d = datetime.strptime(date_from, "%Y-%m-%d") + d_end = datetime.strptime(date_to, "%Y-%m-%d") + dates = [] + while d <= d_end: + dates.append(d.strftime("%Y-%m-%d")) + d += timedelta(days=1) + + print(f"\n{'='*60}") + print(f" BACKFILL {date_from} → {date_to} ({len(dates)} días)") + print(f"{'='*60}\n") + + total_saved = 0 + total_skip = 0 + + for run_date in dates: + print(f"\n── {run_date} ───────────────────────────────────────────────") + + # Pre-load existing snapshots for this date if skip_existing + existing_names: set = set() + if skip_existing: + try: + for r in baserow.get_snapshots_for_date(run_date): + existing_names.add(r.get("campaign_name", "")) + except Exception: + pass + + campaign_metrics = meta.get_campaign_metrics(run_date, run_date) + if not campaign_metrics: + print(" Sin campañas con gasto.") + continue + + print(f" {len(campaign_metrics)} campañas activas.") + + # Adset bid configs (current — bid strategy doesn't change day to day) + adset_bids_cache: dict = {} + + for cid, metrics in campaign_metrics.items(): + camp_name = metrics["name"] + + if skip_existing and camp_name in existing_names: + print(f" SKIP {camp_name[:55]}") + total_skip += 1 + continue + + vertical = _extract_vertical(camp_name) + max_cpl = vertical_cpls.get(vertical, config.META_TARGET_CPL) or config.META_TARGET_CPL + margin = round((max_cpl - metrics["cpl"]) * metrics["leads"], 2) if metrics["leads"] > 0 else round(-metrics["spend"], 2) + + print(f" {camp_name[:55]}") + print(f" Spend {metrics['spend']}€ Leads {metrics['leads']} CPL {metrics['cpl']}€ MaxCPL {max_cpl}€ Margen {margin:+.2f}€") + + # ── Claude: decisión ──────────────────────────────────────────── + analysis = { + "campaign_id": cid, "name": camp_name, "status": "ACTIVE", + "spend": metrics["spend"], "leads": metrics["leads"], + "cpl": metrics["cpl"], "max_cpl": max_cpl, + "ctr": metrics["ctr"], "cpm": metrics["cpm"], + "impressions": metrics["impressions"], "clicks": metrics["clicks"], + } + try: + decision = decide(analysis) + action_type = _ACTION_MAP.get( + decision.get("action") or decision.get("accion", "MAINTAIN"), + "MAINTAIN", + ) + except Exception as e: + print(f" ERROR decide: {e}") + decision = {"action": "MAINTAIN", "justification": "", "parameter": 1.0} + action_type = "MAINTAIN" + + print(f" Decision: {action_type} — {(decision.get('justification') or '')[:70]}") + + # ── Claude: adsets ────────────────────────────────────────────── + adsets_detail = [] + try: + for as_m in meta.get_adset_metrics(cid, run_date, run_date)[:5]: + result = analyze_unit(as_m, "adset") + adsets_detail.append({**as_m, **result}) + print(f" [Adset] {as_m['name'][:45]} — {result.get('evaluacion','')[:50]}") + except Exception as e: + print(f" ERROR adsets: {e}") + + # Add bid configs (cached per campaign) + if cid not in adset_bids_cache: + try: + adset_bids_cache[cid] = meta.get_adset_bid_configs(cid) + except Exception: + adset_bids_cache[cid] = {} + for adset in adsets_detail: + b = adset_bids_cache[cid].get(adset["id"], {}) + adset["cost_cap_eur"] = b.get("cost_cap_eur") + adset["bid_strategy"] = b.get("bid_strategy", "") + + # ── Claude: anuncios ──────────────────────────────────────────── + ads_detail = [] + try: + for ad_m in meta.get_ad_metrics(cid, run_date, run_date)[:5]: + result = analyze_unit(ad_m, "ad") + ads_detail.append({**ad_m, **result}) + print(f" [Ad] {ad_m['name'][:45]} — {result.get('evaluacion','')[:50]}") + except Exception as e: + print(f" ERROR ads: {e}") + + # ── Guardar snapshot ──────────────────────────────────────────── + try: + baserow.save_daily_snapshot({ + "run_date": run_date, + "campaign_id": cid, + "campaign_name": camp_name, + "vertical": vertical, + "spend": metrics["spend"], + "leads": metrics["leads"], + "cpl": metrics["cpl"], + "margin": margin, + "action_type": action_type, + "justification": decision.get("justification") or "", + "adsets": adsets_detail, + "ads": ads_detail, + }) + print(f" ✓ Snapshot guardado") + total_saved += 1 + except Exception as e: + print(f" ERROR snapshot: {e}") + + print(f"\n{'='*60}") + print(f" Backfill completo. Guardados: {total_saved} Saltados: {total_skip}") + print(f"{'='*60}\n") + + +if __name__ == "__main__": + now = datetime.now() + default_from = f"{now.year}-{now.month:02d}-01" + default_to = (now - timedelta(days=1)).strftime("%Y-%m-%d") + + parser = argparse.ArgumentParser(description="Backfill Meta Optimizer snapshots") + parser.add_argument("--from", dest="date_from", default=default_from, + help=f"Fecha inicio YYYY-MM-DD (default: {default_from})") + parser.add_argument("--to", dest="date_to", default=default_to, + help=f"Fecha fin YYYY-MM-DD (default: {default_to})") + parser.add_argument("--skip-existing", action="store_true", + help="No reprocesa campañas que ya tienen snapshot ese día") + args = parser.parse_args() + run_backfill(args.date_from, args.date_to, args.skip_existing) diff --git a/baserow_client.py b/baserow_client.py new file mode 100644 index 0000000..6ac0488 --- /dev/null +++ b/baserow_client.py @@ -0,0 +1,208 @@ +"""Baserow REST API client for meta_optimizer tables.""" +import requests +from datetime import datetime +import config + + +class BaserowClient: + def __init__(self): + self._base = config.BASEROW_URL.rstrip("/") + self._headers = { + "Authorization": f"Token {config.BASEROW_TOKEN}", + "Content-Type": "application/json", + } + + def _url(self, table_id: int, row_id: int = None) -> str: + path = f"/api/database/rows/table/{table_id}/" + if row_id: + path += f"{row_id}/" + return self._base + path + + def _get_rows(self, table_id: int, filters: dict = None) -> list: + params = {"user_field_names": "true"} + if filters: + params.update(filters) + try: + resp = requests.get(self._url(table_id), headers=self._headers, params=params, timeout=15) + if not resp.ok: + return [] + return resp.json().get("results", []) + except requests.RequestException: + return [] + + def _create_row(self, table_id: int, data: dict) -> dict: + resp = requests.post( + self._url(table_id), + headers=self._headers, + params={"user_field_names": "true"}, + json=data, + timeout=15, + ) + resp.raise_for_status() + return resp.json() + + def _delete_row(self, table_id: int, row_id: int): + resp = requests.delete( + self._url(table_id, row_id), + headers=self._headers, + params={"user_field_names": "true"}, + timeout=15, + ) + resp.raise_for_status() + + def _update_row(self, table_id: int, row_id: int, data: dict) -> dict: + resp = requests.patch( + self._url(table_id, row_id), + headers=self._headers, + params={"user_field_names": "true"}, + json=data, + timeout=15, + ) + resp.raise_for_status() + return resp.json() + + # ── verticals ───────────────────────────────────────────────────────────── + + def get_all_verticals(self) -> list: + return self._get_rows(config.BASEROW_TABLE_VERTICALS) + + def get_vertical_config(self, vertical_name: str) -> dict | None: + rows = self._get_rows( + config.BASEROW_TABLE_VERTICALS, + {"filter__Nombre__equal": vertical_name}, + ) + return rows[0] if rows else None + + # ── campaigns ───────────────────────────────────────────────────────────── + + def get_campaign_config(self, campaign_id: str) -> dict | None: + rows = self._get_rows( + config.BASEROW_TABLE_CAMPAIGNS, + {"filter__campaign_id__equal": campaign_id}, + ) + return rows[0] if rows else None + + # ── proposed_actions ────────────────────────────────────────────────────── + + def save_action(self, action: dict) -> dict: + return self._create_row(config.BASEROW_TABLE_ACTIONS, { + "campaign_id": action["campaign_id"], + "campaign_name": action["campaign_name"], + "action_type": action["action_type"], + "parameter": action.get("parameter", 1.0), + "justification": action.get("justification", ""), + "advice": action.get("advice", ""), + "alert": action.get("alert") or "", + "confidence": action.get("confidence", 0.0), + "status": "pending", + "proposed_at": datetime.now().strftime("%Y-%m-%d"), + }) + + def get_approved_actions(self) -> list: + return self._get_rows( + config.BASEROW_TABLE_ACTIONS, + {"filter__status__single_select_equal": "approved"}, + ) + + def update_action_status( + self, row_id: int, status: str, slack_message_ts: str = None + ) -> dict: + data: dict = {"status": status} + if status == "executed": + data["executed_at"] = datetime.now().strftime("%Y-%m-%d") + if slack_message_ts is not None: + data["slack_message_ts"] = slack_message_ts + return self._update_row(config.BASEROW_TABLE_ACTIONS, row_id, data) + + # ── creative_analyses ───────────────────────────────────────────────────── + + def save_creative_analysis(self, analysis: dict) -> dict: + return self._create_row(config.BASEROW_TABLE_CREATIVES, { + "ad_id": analysis["ad_id"], + "ad_name": analysis["ad_name"], + "campaign_id": analysis["campaign_id"], + "image_url": analysis["image_url"], + "analysis": analysis.get("analysis", ""), + "score": analysis.get("score", 0), + "recommendations": analysis.get("recommendations", ""), + "created_at": datetime.now().strftime("%Y-%m-%d"), + }) + + # ── daily_snapshots ─────────────────────────────────────────────────────── + + def save_daily_snapshot(self, snapshot: dict) -> dict: + import json + + # Remove existing snapshots for same day + campaign before saving new one + existing = self._get_rows( + config.BASEROW_TABLE_SNAPSHOTS, + { + "filter__run_date__equal": snapshot["run_date"], + "filter__campaign_name__equal": snapshot["campaign_name"], + }, + ) + for row in existing: + try: + self._delete_row(config.BASEROW_TABLE_SNAPSHOTS, row["id"]) + except Exception: + pass + + def _safe_json(items: list) -> str: + cleaned = [] + for item in items: + cleaned.append({ + k: (str(v)[:500] if isinstance(v, str) else v) + for k, v in item.items() + if isinstance(v, (str, int, float, bool, type(None))) + }) + s = json.dumps(cleaned, ensure_ascii=False) + return s[:60000] # Baserow long_text practical limit + + resp = requests.post( + self._url(config.BASEROW_TABLE_SNAPSHOTS), + headers=self._headers, + params={"user_field_names": "true"}, + json={ + "run_date": snapshot["run_date"], + "campaign_id": snapshot["campaign_id"], + "campaign_name": snapshot["campaign_name"], + "vertical": snapshot["vertical"], + "spend": float(snapshot["spend"]), + "leads": int(snapshot["leads"]), + "cpl": float(snapshot["cpl"]), + "margin": float(snapshot["margin"]), + "action_type": snapshot.get("action_type", "MAINTAIN"), + "justification": (snapshot.get("justification") or "")[:2000], + "adsets_json": _safe_json(snapshot.get("adsets", [])), + "ads_json": _safe_json(snapshot.get("ads", [])), + }, + timeout=15, + ) + if not resp.ok: + raise Exception(f"{resp.status_code} {resp.text[:300]}") + return resp.json() + + def get_snapshots_for_date(self, run_date: str) -> list: + return self._get_rows( + config.BASEROW_TABLE_SNAPSHOTS, + {"filter__run_date__equal": run_date}, + ) + + def get_snapshot_dates(self) -> list: + """Return sorted list of distinct run_date values that have snapshots.""" + rows = self._get_rows(config.BASEROW_TABLE_SNAPSHOTS) + return sorted({r["run_date"] for r in rows if r.get("run_date")}, reverse=True) + + # ── execution_logs ──────────────────────────────────────────────────────── + + def save_execution_log(self, log: dict) -> dict: + return self._create_row(config.BASEROW_TABLE_LOGS, { + "executed_at": datetime.now().strftime("%Y-%m-%d"), + "mode": log.get("mode", "DRY_RUN"), + "campaigns_analyzed": log.get("campaigns_analyzed", 0), + "actions_proposed": log.get("actions_proposed", 0), + "actions_executed": log.get("actions_executed", 0), + "errors": log.get("errors", ""), + "summary": log.get("summary", ""), + "duration_seconds": log.get("duration_seconds", 0.0), + }) diff --git a/config.py b/config.py index 64c72a6..d997d6f 100644 --- a/config.py +++ b/config.py @@ -3,21 +3,33 @@ from dotenv import load_dotenv load_dotenv() -# Airtable -AIRTABLE_TOKEN = os.environ["AIRTABLE_TOKEN"] -AIRTABLE_BASE_ID = os.environ["AIRTABLE_BASE_ID"] - # Meta Ads -META_APP_ID = os.environ["META_APP_ID"] -META_APP_SECRET = os.environ["META_APP_SECRET"] -META_ACCESS_TOKEN = os.environ["META_ACCESS_TOKEN"] -META_AD_ACCOUNT_ID = os.environ["META_AD_ACCOUNT_ID"] # formato: act_XXXXXXXX +META_APP_ID = os.environ["META_APP_ID"] +META_APP_SECRET = os.environ["META_APP_SECRET"] +META_ACCESS_TOKEN = os.environ["META_ACCESS_TOKEN"] +META_AD_ACCOUNT_ID = os.environ["META_AD_ACCOUNT_ID"] # format: act_XXXXXXXX # Anthropic ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"] -# Slack -SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "") +# Baserow +BASEROW_URL = os.environ["BASEROW_URL"] # e.g. https://baserow.yourdomain.com +BASEROW_TOKEN = os.environ["BASEROW_TOKEN"] +BASEROW_TABLE_CAMPAIGNS = int(os.environ["BASEROW_TABLE_CAMPAIGNS"]) +BASEROW_TABLE_ACTIONS = int(os.environ["BASEROW_TABLE_ACTIONS"]) +BASEROW_TABLE_CREATIVES = int(os.environ["BASEROW_TABLE_CREATIVES"]) +BASEROW_TABLE_LOGS = int(os.environ["BASEROW_TABLE_LOGS"]) +BASEROW_TABLE_VERTICALS = int(os.environ["BASEROW_TABLE_VERTICALS"]) +BASEROW_TABLE_SNAPSHOTS = int(os.environ["BASEROW_TABLE_SNAPSHOTS"]) -# Operación -DRY_RUN = True # True = solo sugiere, no aplica cambios en Meta Ads +# Slack (Bot Token, not webhook) +SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"] # xoxb-... +SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"] # for verifying callbacks +SLACK_CHANNEL_ID = os.environ["SLACK_CHANNEL_ID"] # e.g. C0XXXXXXX + +# Campaign filter +META_CAMPAIGN_PREFIX = os.environ.get("META_CAMPAIGN_PREFIX", "VIVIFUL") +META_TARGET_CPL = float(os.environ.get("META_TARGET_CPL", "0")) # 0 = use per-campaign Baserow value + +# Operation +DRY_RUN = os.environ.get("DRY_RUN", "true").lower() != "false" diff --git a/create_snapshots_table.py b/create_snapshots_table.py new file mode 100644 index 0000000..df13551 --- /dev/null +++ b/create_snapshots_table.py @@ -0,0 +1,85 @@ +"""One-time: add daily_snapshots table to existing meta_optimizer database in Baserow.""" +import os +import sys +import requests +from dotenv import load_dotenv + +load_dotenv() + +BASE_URL = os.environ.get("BASEROW_URL", "").rstrip("/") +EMAIL = os.environ.get("BASEROW_EMAIL", "") +PASSWORD = os.environ.get("BASEROW_PASSWORD", "") + +if not BASE_URL or not EMAIL or not PASSWORD: + print("Error: BASEROW_URL, BASEROW_EMAIL and BASEROW_PASSWORD must be set in .env") + sys.exit(1) + +auth = requests.post(f"{BASE_URL}/api/user/token-auth/", + json={"email": EMAIL, "password": PASSWORD}, timeout=10) +if not auth.ok: + print(f"Auth error: {auth.text}") + sys.exit(1) +JWT = auth.json()["access_token"] +HEADERS = {"Authorization": f"JWT {JWT}", "Content-Type": "application/json"} + + +def api(method, path, **kwargs): + resp = requests.request(method, f"{BASE_URL}/api{path}", headers=HEADERS, **kwargs) + if not resp.ok: + print(f" API error {resp.status_code} {method} {path}: {resp.text[:300]}") + resp.raise_for_status() + return resp.json() + + +# Find meta_optimizer database +db_id = None +for ws in api("GET", "/workspaces/"): + for app in api("GET", f"/applications/workspace/{ws['id']}/"): + if app.get("name") == "meta_optimizer": + db_id = app["id"] + break + if db_id: + break + +if not db_id: + db_id_env = os.environ.get("BASEROW_DB_ID") + if db_id_env: + db_id = int(db_id_env) + else: + print("Error: could not find meta_optimizer database. Set BASEROW_DB_ID in .env.") + sys.exit(1) + +print(f"Database: meta_optimizer (id={db_id})") + +# Create table +t = api("POST", f"/database/tables/database/{db_id}/", json={"name": "daily_snapshots"}) +table_id = t["id"] +print(f"Table: daily_snapshots (id={table_id})") + +# Rename primary field +primary_id = api("GET", f"/database/fields/table/{table_id}/")[0]["id"] +api("PATCH", f"/database/fields/{primary_id}/", json={"name": "run_date", "type": "text"}) +print(" ~ primary field: run_date") + +for f in [ + {"name": "campaign_id", "type": "text"}, + {"name": "campaign_name", "type": "text"}, + {"name": "vertical", "type": "text"}, + {"name": "spend", "type": "number", "number_decimal_places": 2}, + {"name": "leads", "type": "number"}, + {"name": "cpl", "type": "number", "number_decimal_places": 2}, + {"name": "margin", "type": "number", "number_decimal_places": 2, "number_negative": True}, + {"name": "action_type", "type": "text"}, + {"name": "justification", "type": "long_text"}, + {"name": "adsets_json", "type": "long_text"}, + {"name": "ads_json", "type": "long_text"}, +]: + api("POST", f"/database/fields/table/{table_id}/", json=f) + print(f" + {f['name']}") + +print(f""" +{'='*50} + Añade esto a tu .env: + BASEROW_TABLE_SNAPSHOTS={table_id} +{'='*50} +""") diff --git a/dashboard.py b/dashboard.py new file mode 100644 index 0000000..b155cf7 --- /dev/null +++ b/dashboard.py @@ -0,0 +1,475 @@ +"""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 = st.tabs(["📅 Por día", "📊 Campañas", "🏷️ Verticales", "🗂️ Histórico"]) + + +# ── 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']}") diff --git a/meta_ads_client.py b/meta_ads_client.py index 7651072..19c67f6 100644 --- a/meta_ads_client.py +++ b/meta_ads_client.py @@ -1,13 +1,16 @@ """ -Cliente para Meta Marketing API. -Documentación: https://developers.facebook.com/docs/marketing-api +Client for Meta Marketing API. +Docs: https://developers.facebook.com/docs/marketing-api SDK: facebook-business """ from facebook_business.api import FacebookAdsApi from facebook_business.adobjects.adaccount import AdAccount 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 import config -from datetime import datetime +from datetime import datetime, timedelta class MetaAdsClient: @@ -19,60 +22,221 @@ class MetaAdsClient: ) self.account = AdAccount(config.META_AD_ACCOUNT_ID) - def get_monthly_metrics_all(self) -> dict: - """ - Métricas del mes en curso para todas las campañas activas. - Retorna dict {campaign_id: {spend, impressions, clicks, ctr, cpm, leads, cpl, status, name}}. - """ - now = datetime.now() - date_start = f"{now.year}-{now.month:02d}-01" - date_end = now.strftime("%Y-%m-%d") - - campaigns = self.account.get_campaigns(fields=[ - Campaign.Field.id, - Campaign.Field.name, - Campaign.Field.status, - Campaign.Field.effective_status, - ], params={"effective_status": ["ACTIVE", "PAUSED"]}) + def _parse_insights_row(self, row: dict) -> dict: + spend = float(row.get("spend", 0)) + impressions = int(row.get("impressions", 0)) + clicks = int(row.get("clicks", 0)) + ctr = float(row.get("ctr", 0)) + cpm = float(row.get("cpm", 0)) + leads = sum(float(a["value"]) for a in row.get("actions", []) + if a["action_type"] in ("lead", "onsite_conversion.lead_grouped")) + cpl = round(spend / leads, 2) if leads > 0 else 0.0 + return { + "campaign_id": row.get("campaign_id", ""), + "name": row.get("campaign_name", ""), + "status": "ACTIVE", + "spend": round(spend, 2), + "impressions": impressions, + "clicks": clicks, + "ctr": round(ctr, 4), + "cpm": round(cpm, 2), + "leads": int(leads), + "cpl": cpl, + } + def get_campaign_metrics(self, date_from: str, date_to: str) -> dict: + """ + Campaign-level metrics aggregated over a date range. + Returns {campaign_id: metrics}, spend > 0 only. + """ + prefix = config.META_CAMPAIGN_PREFIX.upper() + insights = self.account.get_insights( + fields=["campaign_id", "campaign_name", "spend", "impressions", + "clicks", "ctr", "cpm", "actions"], + params={ + "level": "campaign", + "time_range": {"since": date_from, "until": date_to}, + } + ) result = {} - for c in campaigns: - cid = c["id"] - name = c["name"] - status = c.get("effective_status", "UNKNOWN") + for row in insights: + if not row.get("campaign_name", "").upper().startswith(prefix): + continue + m = self._parse_insights_row(row) + if m["spend"] == 0: + continue + result[m["campaign_id"]] = m + return result - insights = c.get_insights(fields=[ - "spend", "impressions", "clicks", "ctr", "cpm", - "actions", # conversiones por tipo (lead, purchase, etc.) - "cost_per_action_type", - ], params={ - "time_range": {"since": date_start, "until": date_end}, - "level": "campaign", + def get_daily_campaign_rows(self, date_from: str, date_to: str) -> list: + """ + Per-campaign per-day rows for a date range. + Returns [{date, campaign_name, spend, leads}] sorted by date. + """ + if date_from > date_to: + return [] + prefix = config.META_CAMPAIGN_PREFIX.upper() + insights = self.account.get_insights( + fields=["date_start", "campaign_name", "spend", "actions"], + params={ + "level": "campaign", + "time_range": {"since": date_from, "until": date_to}, + "time_increment": 1, + } + ) + result = [] + for row in insights: + if not row.get("campaign_name", "").upper().startswith(prefix): + continue + spend = float(row.get("spend", 0)) + leads = sum(float(a["value"]) for a in row.get("actions", []) + if a["action_type"] in ("lead", "onsite_conversion.lead_grouped")) + result.append({ + "date": row["date_start"], + "campaign_name": row.get("campaign_name", ""), + "spend": round(spend, 2), + "leads": int(leads), }) + return sorted(result, key=lambda x: x["date"]) - spend = impressions = clicks = ctr = cpm = leads = 0.0 - if insights: - row = insights[0] - spend = float(row.get("spend", 0)) - impressions = int(row.get("impressions", 0)) - clicks = int(row.get("clicks", 0)) - ctr = float(row.get("ctr", 0)) - cpm = float(row.get("cpm", 0)) - for action in row.get("actions", []): - if action["action_type"] in ("lead", "onsite_conversion.lead_grouped"): - leads += float(action["value"]) - + def _get_sub_insights_range(self, campaign_id: str, level: str, + date_from: str, date_to: str) -> list: + """Ad set or ad level insights for a date range, spend > 0, sorted by spend desc.""" + id_field = f"{level}_id" + name_field = f"{level}_name" + try: + insights = Campaign(campaign_id).get_insights( + fields=[id_field, name_field, "spend", "impressions", + "clicks", "ctr", "cpm", "actions"], + params={ + "level": level, + "time_range": {"since": date_from, "until": date_to}, + } + ) + except Exception: + return [] + result = [] + for row in insights: + spend = float(row.get("spend", 0)) + if spend == 0: + continue + leads = sum(float(a["value"]) for a in row.get("actions", []) + if a["action_type"] in ("lead", "onsite_conversion.lead_grouped")) cpl = round(spend / leads, 2) if leads > 0 else 0.0 - result[cid] = { - "campaign_id": cid, - "name": name, - "status": status, + result.append({ + "id": row.get(id_field, ""), + "name": row.get(name_field, ""), "spend": round(spend, 2), - "impressions": impressions, - "clicks": clicks, - "ctr": round(ctr, 4), - "cpm": round(cpm, 2), + "impressions": int(row.get("impressions", 0)), + "clicks": int(row.get("clicks", 0)), + "ctr": round(float(row.get("ctr", 0)), 4), + "cpm": round(float(row.get("cpm", 0)), 2), "leads": int(leads), "cpl": cpl, - } + }) + return sorted(result, key=lambda x: -x["spend"]) + + def get_yesterday_metrics(self) -> dict: + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + return self.get_campaign_metrics(yesterday, yesterday) + + def get_monthly_daily_totals(self) -> list: + """Per-campaign daily rows for the current month (used by run.py).""" + now = datetime.now() + date_start = f"{now.year}-{now.month:02d}-01" + yesterday = (now - timedelta(days=1)).strftime("%Y-%m-%d") + return self.get_daily_campaign_rows(date_start, yesterday) + + def get_adset_metrics(self, campaign_id: str, date_from: str, date_to: str) -> list: + return self._get_sub_insights_range(campaign_id, "adset", date_from, date_to) + + def get_ad_metrics(self, campaign_id: str, date_from: str, date_to: str) -> list: + return self._get_sub_insights_range(campaign_id, "ad", date_from, date_to) + + def get_yesterday_adset_metrics(self, campaign_id: str) -> list: + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + return self.get_adset_metrics(campaign_id, yesterday, yesterday) + + def get_yesterday_ad_metrics(self, campaign_id: str) -> list: + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + return self.get_ad_metrics(campaign_id, yesterday, yesterday) + + def get_ads_with_creatives(self, campaign_id: str) -> list: + """ + Returns active ads for a campaign with their thumbnail URLs for creative analysis. + """ + campaign = Campaign(campaign_id) + ads = campaign.get_ads( + fields=[Ad.Field.id, Ad.Field.name, Ad.Field.status, Ad.Field.creative], + 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 = "" + + if creative_id: + try: + creative = AdCreative(creative_id).api_get( + fields=["thumbnail_url", "image_url"] + ) + thumbnail = creative.get("thumbnail_url") or creative.get("image_url", "") + except Exception: + pass + + result.append({ + "ad_id": ad["id"], + "ad_name": ad["name"], + "campaign_id": campaign_id, + "thumbnail_url": thumbnail, + }) + return result + + def get_campaign_bid_config(self, campaign_id: str) -> dict: + """Fetch bid strategy and daily/lifetime budget at campaign level.""" + try: + data = Campaign(campaign_id).api_get( + fields=["bid_strategy", "daily_budget", "lifetime_budget"] + ) + daily = float(data.get("daily_budget", 0) or 0) + lifetime = float(data.get("lifetime_budget", 0) or 0) + return { + "bid_strategy": data.get("bid_strategy", ""), + "daily_budget_eur": round(daily / 100, 2) if daily else None, + "lifetime_budget_eur": round(lifetime / 100, 2) if lifetime else None, + } + except Exception: + return {} + + def get_adset_bid_configs(self, campaign_id: str) -> dict: + """Returns {adset_id: {bid_strategy, cost_cap_eur, daily_budget_eur}}.""" + try: + adsets = Campaign(campaign_id).get_ad_sets( + fields=[AdSet.Field.id, AdSet.Field.bid_strategy, + AdSet.Field.bid_amount, AdSet.Field.daily_budget] + ) + result = {} + for as_obj in adsets: + bid_amount = float(as_obj.get("bid_amount", 0) or 0) + daily = float(as_obj.get("daily_budget", 0) or 0) + result[as_obj["id"]] = { + "bid_strategy": as_obj.get("bid_strategy", ""), + "cost_cap_eur": round(bid_amount / 100, 2) if bid_amount else None, + "daily_budget_eur": round(daily / 100, 2) if daily else None, + } + return result + except Exception: + return {} + + def set_campaign_budget(self, campaign_id: str, daily_budget_cents: int): + """Set campaign daily budget (amount in cents).""" + campaign = Campaign(campaign_id) + campaign.api_update(params={"daily_budget": daily_budget_cents}) + + def pause_campaign(self, campaign_id: str): + """Pause a campaign.""" + campaign = Campaign(campaign_id) + campaign.api_update(params={"status": Campaign.Status.paused}) diff --git a/requirements.txt b/requirements.txt index 4ef8532..6493e11 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,8 @@ anthropic==0.95.0 -pyairtable==3.3.0 facebook-business>=19.0.0 python-dotenv==1.2.2 requests>=2.32.0 +fastapi>=0.111.0 +uvicorn>=0.29.0 +streamlit>=1.35.0 +pandas>=2.0.0 diff --git a/run.py b/run.py index ff065f6..deb4c1a 100644 --- a/run.py +++ b/run.py @@ -1,23 +1,46 @@ -""" -Meta Optimizer — punto de entrada principal. -Analiza campañas de Meta Ads y publica resumen en Slack. -""" +"""Meta Optimizer — main entry point.""" import sys import io import os -import json -sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) +import time -from meta_ads_client import MetaAdsClient -from agent import decide -import config from datetime import datetime +import config +from meta_ads_client import MetaAdsClient +from agent import decide, analyze_creative, analyze_unit +from baserow_client import BaserowClient +import slack_notifier + + +_ACTION_MAP = { + "PAUSE": "PAUSE", + "REDUCE_BUDGET": "REDUCE_BUDGET", + "INCREASE_BUDGET": "INCREASE_BUDGET", + "MAINTAIN": "MAINTAIN", + "REVIEW_CREATIVES": "REVIEW_CREATIVES", + # legacy Spanish names + "PAUSAR": "PAUSE", + "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", + "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", + "MANTENER": "MAINTAIN", + "REVISAR_CREATIVIDADES":"REVIEW_CREATIVES", +} + + +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" + class Tee: - def __init__(self, filepath): + def __init__(self, filepath: str): os.makedirs(os.path.dirname(filepath), exist_ok=True) - self._file = open(filepath, "w", encoding="utf-8") + self._file = open(filepath, "w", encoding="utf-8") self._stdout = sys.stdout def write(self, data): @@ -33,50 +56,314 @@ class Tee: self._file.close() +def _execute_action(meta: MetaAdsClient, action: dict): + """Apply an approved action via Meta API.""" + action_type = action.get("action_type", "") + cid = action.get("campaign_id", "") + parameter = float(action.get("parameter") or 1.0) + + if action_type == "PAUSE": + meta.pause_campaign(cid) + + elif action_type in ("REDUCE_BUDGET", "INCREASE_BUDGET"): + try: + bid_cfg = meta.get_campaign_bid_config(cid) + current_budget = bid_cfg.get("daily_budget_eur") + if current_budget: + meta.set_campaign_budget(cid, int(current_budget * parameter * 100)) + except Exception: + pass + + def run(): - now = datetime.now() + start_ts = time.time() + now = datetime.now() + print(f"\n{'='*55}") print(f" META OPTIMIZER — {now.strftime('%d/%m/%Y %H:%M')}") - print(f" Modo: {'DRY RUN (sin cambios)' if config.DRY_RUN else 'PRODUCCIÓN'}") + print(f" Prefix: {config.META_CAMPAIGN_PREFIX} | Target CPL: {config.META_TARGET_CPL}€") + print(f" Mode: {'DRY RUN (no changes)' if config.DRY_RUN else 'PRODUCTION'}") print(f"{'='*55}\n") - meta = MetaAdsClient() + meta = MetaAdsClient() + baserow = BaserowClient() - print("→ Obteniendo métricas del mes desde Meta Ads...") - metrics_all = meta.get_monthly_metrics_all() - print(f" ✓ {len(metrics_all)} campañas encontradas.\n") + # ── Fetch all vertical CPL targets upfront ──────────────────────────────── + 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 - results = [] - for cid, metrics in metrics_all.items(): - analysis = { - "campaign_id": cid, - "name": metrics["name"], - "status": metrics["status"], - "spend": metrics["spend"], - "leads": metrics["leads"], - "cpl": metrics["cpl"], - "cpl_maximo": 0, # TODO: cargar desde Airtable o config por campaña - "ctr": metrics["ctr"], - "cpm": metrics["cpm"], - "impressions": metrics["impressions"], - "clicks": metrics["clicks"], + # ── Execute previously approved actions ─────────────────────────────────── + actions_executed = 0 + if not config.DRY_RUN: + approved = baserow.get_approved_actions() + print(f"→ Executing {len(approved)} approved actions...\n") + for action in approved: + try: + _execute_action(meta, action) + baserow.update_action_status(action["id"], "executed") + actions_executed += 1 + print(f" ✓ {action.get('campaign_name')} — {action.get('action_type')}") + except Exception as e: + print(f" ✗ Error on action {action['id']}: {e}") + + # ── Monthly daily totals (per-campaign rows → aggregate with margins) ───── + print(f"→ Fetching monthly daily totals for {config.META_CAMPAIGN_PREFIX}...") + daily_rows = meta.get_monthly_daily_totals() + _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": date, + "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), } - decision = decide(analysis) - results.append({"metrics": metrics, "analysis": analysis, "decision": decision}) + for date, d in sorted(_daily.items()) + ] + print(f" ✓ {len(daily_totals)} days with data.\n") - print(f"📢 {metrics['name'][:50]}") - print(f" Gasto: {metrics['spend']}€ | Leads: {metrics['leads']} | CPL: {metrics['cpl']}€") - print(f" Decisión: {decision['accion']} — {decision['justificacion'][:80]}") - if decision.get("alerta"): - print(f" 🚨 {decision['alerta']}") + # ── Yesterday metrics ───────────────────────────────────────────────────── + print(f"→ Fetching yesterday metrics ({config.META_CAMPAIGN_PREFIX} only, spend > 0)...") + metrics_all = meta.get_yesterday_metrics() + print(f" ✓ {len(metrics_all)} campaigns active yesterday.\n") + + # ── Analyze campaigns & propose actions ─────────────────────────────────── + actions_proposed_list = [] + campaign_details = {} # {cid: {vertical, margin, adsets, ads}} + verticals = {} # {vertical: {spend, leads, margin}} + errors = [] + + for cid, metrics in metrics_all.items(): + vertical = _extract_vertical(metrics["name"]) + max_cpl = vertical_cpls.get(vertical, config.META_TARGET_CPL) or config.META_TARGET_CPL + margin = round((max_cpl - metrics["cpl"]) * metrics["leads"], 2) if metrics["leads"] > 0 else round(-metrics["spend"], 2) + + analysis = { + "campaign_id": cid, + "name": metrics["name"], + "status": metrics["status"], + "spend": metrics["spend"], + "leads": metrics["leads"], + "cpl": metrics["cpl"], + "max_cpl": max_cpl, + "ctr": metrics["ctr"], + "cpm": metrics["cpm"], + "impressions": metrics["impressions"], + "clicks": metrics["clicks"], + } + + try: + decision = decide(analysis) + except Exception as e: + errors.append(f"{metrics['name']}: {e}") + continue + + action_type = _ACTION_MAP.get( + decision.get("action") or decision.get("accion", "MAINTAIN"), + "MAINTAIN", + ) + + print(f" {metrics['name'][:52]}") + print(f" Spend: {metrics['spend']}€ Leads: {metrics['leads']} CPL: {metrics['cpl']}€ MaxCPL: {max_cpl}€ Margen: {margin:+.2f}€") + print(f" Vertical: {vertical} Decision: {action_type} — {(decision.get('justification') or '')[:70]}") + if decision.get("alert"): + print(f" ALERT: {decision['alert']}") print() - print(f"Log guardado en: logs/{now.strftime('%Y%m%d_%H%M%S')}.log") + if action_type != "MAINTAIN": + try: + row = baserow.save_action({ + "campaign_id": cid, + "campaign_name": metrics["name"], + "action_type": action_type, + "parameter": decision.get("parameter") or 1.0, + "justification": decision.get("justification") or "", + "advice": decision.get("advice") or "", + "alert": decision.get("alert") or "", + "confidence": decision.get("confidence") or 0.0, + }) + actions_proposed_list.append({ + "campaign_name": metrics["name"], + "action_type": action_type, + "parameter": decision.get("parameter") or 1.0, + "justification": decision.get("justification") or "", + "advice": decision.get("advice") or "", + "alert": decision.get("alert") or "", + "confidence": decision.get("confidence") or 0.0, + "cpl": metrics["cpl"], + "max_cpl": max_cpl, + "row_id": row["id"], + }) + except Exception as e: + errors.append(f"Save action {metrics['name']}: {e}") + + # ── Ad set analysis ──────────────────────────────────────────────── + # ── Bid config (campaña + adsets) — antes del análisis para pasarlo a Claude ── + campaign_bid = {} + try: + campaign_bid = meta.get_campaign_bid_config(cid) + except Exception as e: + errors.append(f"Bid config {metrics['name']}: {e}") + + adset_bids = {} + try: + adset_bids = meta.get_adset_bid_configs(cid) + except Exception as e: + errors.append(f"Adset bids {metrics['name']}: {e}") + + # ── Ad set analysis (con cost_cap_eur disponible para Claude) ────────── + adsets_detail = [] + try: + for as_m in meta.get_yesterday_adset_metrics(cid)[:5]: + bid = adset_bids.get(as_m["id"], {}) + as_m["bid_strategy"] = bid.get("bid_strategy", "") + as_m["cost_cap_eur"] = bid.get("cost_cap_eur") + result = analyze_unit(as_m, "adset") + adsets_detail.append({**as_m, **result}) + print(f" [Adset] {as_m['name'][:45]} — {result.get('evaluacion','')[:60]}") + except Exception as e: + errors.append(f"Adsets {metrics['name']}: {e}") + + # ── Ad analysis ──────────────────────────────────────────────────── + ads_detail = [] + try: + for ad_m in meta.get_yesterday_ad_metrics(cid)[:5]: + result = analyze_unit(ad_m, "ad") + ads_detail.append({**ad_m, **result}) + print(f" [Ad] {ad_m['name'][:45]} — {result.get('evaluacion','')[:60]}") + except Exception as e: + errors.append(f"Ads {metrics['name']}: {e}") + + campaign_details[cid] = { + "name": metrics["name"], + "vertical": vertical, + "margin": margin, + "adsets": adsets_detail, + "ads": ads_detail, + "bid_config": campaign_bid, + } + + # ── Daily snapshot (persists analysis to Baserow for dashboard) ─────── + try: + action_info = next( + (a for a in actions_proposed_list if a["campaign_name"] == metrics["name"]), + None, + ) + baserow.save_daily_snapshot({ + "run_date": now.strftime("%Y-%m-%d"), + "campaign_id": cid, + "campaign_name": metrics["name"], + "vertical": vertical, + "spend": metrics["spend"], + "leads": metrics["leads"], + "cpl": metrics["cpl"], + "margin": margin, + "action_type": action_info["action_type"] if action_info else "MAINTAIN", + "justification": action_info["justification"] if action_info else "", + "adsets": adsets_detail, + "ads": ads_detail, + }) + except Exception as e: + errors.append(f"Snapshot {metrics['name']}: {e}") + + # ── Vertical aggregation ─────────────────────────────────────────── + if vertical not in verticals: + verticals[vertical] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": max_cpl} + verticals[vertical]["spend"] += metrics["spend"] + verticals[vertical]["leads"] += metrics["leads"] + verticals[vertical]["margin"] += margin + + # ── Creative visual analysis (Baserow storage) ───────────────────── + try: + for ad in meta.get_ads_with_creatives(cid): + if not ad["thumbnail_url"]: + continue + result = analyze_creative(ad["thumbnail_url"], ad["ad_name"]) + baserow.save_creative_analysis({ + "ad_id": ad["ad_id"], + "ad_name": ad["ad_name"], + "campaign_id": cid, + "image_url": ad["thumbnail_url"], + "analysis": result.get("analysis", ""), + "score": result.get("score", 0), + "recommendations": result.get("recommendations", ""), + }) + except Exception as e: + errors.append(f"Creatives {metrics['name']}: {e}") + + # ── Top 10 best and worst ───────────────────────────────────────────────── + with_leads = [m for m in metrics_all.values() if m["leads"] > 0] + best_10 = sorted(with_leads, key=lambda x: x["cpl"])[:10] + + all_active = list(metrics_all.values()) + worst_10 = sorted( + all_active, + key=lambda x: (x["leads"] > 0, -x["cpl"] if x["leads"] > 0 else 0), + )[:10] + + # ── Send consolidated Slack report ──────────────────────────────────────── + duration = round(time.time() - start_ts, 1) + + try: + slack_notifier.send_daily_report( + daily_totals=daily_totals, + best_campaigns=best_10, + worst_campaigns=worst_10, + actions=actions_proposed_list, + target_cpl=config.META_TARGET_CPL, + campaigns_analyzed=len(metrics_all), + mode="DRY_RUN" if config.DRY_RUN else "PRODUCTION", + verticals=verticals, + campaign_details=campaign_details, + ) + except Exception as e: + print(f" Warning: Slack notification failed: {e}") + + # ── Execution log ───────────────────────────────────────────────────────── + summary = ( + f"{len(metrics_all)} campaigns analyzed, " + f"{len(actions_proposed_list)} actions proposed, " + f"{actions_executed} executed." + ) + try: + baserow.save_execution_log({ + "mode": "DRY_RUN" if config.DRY_RUN else "PRODUCTION", + "campaigns_analyzed": len(metrics_all), + "actions_proposed": len(actions_proposed_list), + "actions_executed": actions_executed, + "errors": "\n".join(errors), + "summary": summary, + "duration_seconds": duration, + }) + except Exception as e: + print(f" Warning: could not save execution log: {e}") + + print(f"{'='*55}") + print(f" Done in {duration}s. {summary}") + if errors: + print(f" Errors ({len(errors)}): {'; '.join(errors[:3])}") + print(f"{'='*55}\n") if __name__ == "__main__": + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - log_path = os.path.join("logs", f"{timestamp}.log") + log_path = os.path.join("logs", f"{timestamp}.log") tee = Tee(log_path) sys.stdout = tee try: diff --git a/send_slack_report.py b/send_slack_report.py new file mode 100644 index 0000000..518a35d --- /dev/null +++ b/send_slack_report.py @@ -0,0 +1,173 @@ +"""Re-send yesterday's Slack report from Baserow snapshots.""" +import json +from datetime import datetime, timedelta + +import config +from meta_ads_client import MetaAdsClient +from baserow_client import BaserowClient +import slack_notifier + + +def _extract_vertical(name: str) -> str: + 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" + + +def main(): + run_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + print(f"Reenviando informe para {run_date}...") + + meta = MetaAdsClient() + baserow = BaserowClient() + + # ── Vertical CPL targets ────────────────────────────────────────────────── + 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 + print(f" ✓ Verticales: {vertical_cpls}") + except Exception as e: + print(f" ⚠ No se pudieron cargar verticales: {e}") + + # ── Monthly daily totals ────────────────────────────────────────────────── + print("Obteniendo datos mensuales de Meta...") + daily_rows = meta.get_monthly_daily_totals() + _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": date, + "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 date, d in sorted(_daily.items()) + ] + print(f" ✓ {len(daily_totals)} días con datos") + + # ── Load proposed actions (to get parameter values) ────────────────────── + action_params: dict = {} # campaign_name → parameter + try: + all_actions = baserow._get_rows(config.BASEROW_TABLE_ACTIONS, { + "filter__proposed_at__equal": run_date, + }) + for a in all_actions: + cname = a.get("campaign_name", "") + param = a.get("parameter") + if cname and param: + action_params[cname] = float(param) + print(f" ✓ {len(action_params)} parámetros de acción cargados") + except Exception as e: + print(f" Aviso: no se pudieron cargar parámetros de acción: {e}") + + # ── Load snapshots from Baserow ─────────────────────────────────────────── + print(f"Cargando snapshots de Baserow para {run_date}...") + snapshots = baserow.get_snapshots_for_date(run_date) + print(f" ✓ {len(snapshots)} snapshots encontrados") + + if not snapshots: + print("ERROR: No hay snapshots en Baserow para esta fecha. " + "Ejecuta run.py primero.") + return + + # ── Reconstruct data structures ─────────────────────────────────────────── + campaign_details: dict = {} + actions: list = [] + verticals: dict = {} + metrics_all: dict = {} + + for snap in snapshots: + cid = snap.get("campaign_id") or snap.get("campaign_name", "") + name = snap["campaign_name"] + vertical = snap.get("vertical") or _extract_vertical(name) + margin = float(snap.get("margin") or 0) + spend = float(snap.get("spend") or 0) + leads = int(snap.get("leads") or 0) + cpl = float(snap.get("cpl") or 0) + action_type = snap.get("action_type") or "MAINTAIN" + + try: + adsets = json.loads(snap.get("adsets_json") or "[]") + except Exception: + adsets = [] + try: + ads = json.loads(snap.get("ads_json") or "[]") + except Exception: + ads = [] + + campaign_details[cid] = { + "name": name, + "vertical": vertical, + "margin": margin, + "adsets": adsets, + "ads": ads, + "bid_config": {}, + } + metrics_all[cid] = {"name": name, "spend": spend, "leads": leads, "cpl": cpl} + + if action_type != "MAINTAIN": + actions.append({ + "campaign_name": name, + "action_type": action_type, + "justification": snap.get("justification") or "", + "advice": "", + "alert": "", + "confidence": 0.8, + "cpl": cpl, + "parameter": action_params.get(name, 1.0), + "row_id": snap["id"], + }) + + max_cpl = vertical_cpls.get(vertical, config.META_TARGET_CPL) + if vertical not in verticals: + verticals[vertical] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": max_cpl} + verticals[vertical]["spend"] += spend + verticals[vertical]["leads"] += leads + verticals[vertical]["margin"] += margin + + # ── Best / worst ────────────────────────────────────────────────────────── + with_leads = [m for m in metrics_all.values() if m["leads"] > 0] + best_10 = sorted(with_leads, key=lambda x: x["cpl"])[:10] + worst_10 = sorted( + list(metrics_all.values()), + key=lambda x: (x["leads"] > 0, -x["cpl"] if x["leads"] > 0 else 0), + )[:10] + + # ── Send ────────────────────────────────────────────────────────────────── + print("Enviando a Slack...") + ts = slack_notifier.send_daily_report( + daily_totals=daily_totals, + best_campaigns=best_10, + worst_campaigns=worst_10, + actions=actions, + target_cpl=config.META_TARGET_CPL, + campaigns_analyzed=len(snapshots), + mode="DRY_RUN", + verticals=verticals, + campaign_details=campaign_details, + ) + if ts: + print(f" ✓ Mensaje enviado (ts={ts})") + else: + print(" ✗ Error al enviar (revisa token y canal)") + + +if __name__ == "__main__": + main() diff --git a/setup_baserow.py b/setup_baserow.py new file mode 100644 index 0000000..402c047 --- /dev/null +++ b/setup_baserow.py @@ -0,0 +1,163 @@ +""" +One-time script: creates the meta_optimizer database in Baserow with all tables and fields. +Run once, then paste the printed IDs into your .env file. + +Usage: + python setup_baserow.py +""" +import os +import sys +import requests +from dotenv import load_dotenv + +load_dotenv() + +BASE_URL = os.environ.get("BASEROW_URL", "").rstrip("/") +EMAIL = os.environ.get("BASEROW_EMAIL", "") +PASSWORD = os.environ.get("BASEROW_PASSWORD", "") + +if not BASE_URL or not EMAIL or not PASSWORD: + print("Error: BASEROW_URL, BASEROW_EMAIL and BASEROW_PASSWORD must be set in your .env file.") + sys.exit(1) + +# Authenticate to get a JWT token +_auth = requests.post(f"{BASE_URL}/api/user/token-auth/", + json={"email": EMAIL, "password": PASSWORD}, timeout=10) +if not _auth.ok: + print(f"Auth error: {_auth.text}") + sys.exit(1) +JWT = _auth.json()["access_token"] + +HEADERS = { + "Authorization": f"JWT {JWT}", + "Content-Type": "application/json", +} + + +def api(method: str, path: str, **kwargs) -> dict: + url = f"{BASE_URL}/api{path}" + resp = requests.request(method, url, headers=HEADERS, **kwargs) + if not resp.ok: + print(f" API error {resp.status_code} {method} {path}: {resp.text[:300]}") + resp.raise_for_status() + return resp.json() + + +def setup_table(db_id: int, table_name: str, fields: list) -> int: + """Create table, rename the auto-created primary field, then add remaining fields.""" + t = api("POST", f"/database/tables/database/{db_id}/", json={"name": table_name}) + table_id = t["id"] + print(f"\n Table: {table_name} (id={table_id})") + + existing = api("GET", f"/database/fields/table/{table_id}/") + primary_id = existing[0]["id"] + + first = fields[0] + api("PATCH", f"/database/fields/{primary_id}/", json={"name": first["name"], "type": first["type"]}) + print(f" ~ primary field renamed to: {first['name']}") + + for field in fields[1:]: + api("POST", f"/database/fields/table/{table_id}/", json=field) + print(f" + {field['name']}") + + return table_id + + +# ── 1. Workspace ────────────────────────────────────────────────────────────── + +workspaces = api("GET", "/workspaces/") +if not workspaces: + print("No workspaces found. Create one in Baserow first.") + sys.exit(1) +workspace_id = workspaces[0]["id"] +print(f"Workspace: {workspaces[0]['name']} (id={workspace_id})") + +# ── 2. Database ─────────────────────────────────────────────────────────────── + +db = api("POST", f"/applications/workspace/{workspace_id}/", json={ + "type": "database", + "name": "meta_optimizer", +}) +db_id = db["id"] +print(f"Database: meta_optimizer (id={db_id})") + +# ── 3. Tables ───────────────────────────────────────────────────────────────── + +tid_campaigns = setup_table(db_id, "campaigns", [ + {"name": "campaign_id", "type": "text"}, + {"name": "name", "type": "text"}, + {"name": "max_cpl", "type": "number", "number_decimal_places": 2}, + {"name": "daily_budget", "type": "number", "number_decimal_places": 2}, + {"name": "is_active", "type": "boolean"}, + {"name": "notes", "type": "long_text"}, +]) + +tid_actions = setup_table(db_id, "proposed_actions", [ + {"name": "campaign_id", "type": "text"}, + {"name": "campaign_name", "type": "text"}, + { + "name": "action_type", + "type": "single_select", + "select_options": [ + {"value": "PAUSE", "color": "red"}, + {"value": "REDUCE_BUDGET", "color": "orange"}, + {"value": "INCREASE_BUDGET", "color": "green"}, + {"value": "MAINTAIN", "color": "blue"}, + {"value": "REVIEW_CREATIVES", "color": "purple"}, + ], + }, + {"name": "parameter", "type": "number", "number_decimal_places": 2}, + {"name": "justification", "type": "long_text"}, + {"name": "advice", "type": "long_text"}, + {"name": "alert", "type": "long_text"}, + {"name": "confidence", "type": "number", "number_decimal_places": 2}, + { + "name": "status", + "type": "single_select", + "select_options": [ + {"value": "pending", "color": "yellow"}, + {"value": "approved", "color": "green"}, + {"value": "rejected", "color": "red"}, + {"value": "executed", "color": "blue"}, + ], + }, + {"name": "proposed_at", "type": "date", "date_format": "ISO", "date_include_time": True}, + {"name": "executed_at", "type": "date", "date_format": "ISO", "date_include_time": True}, + {"name": "slack_message_ts", "type": "text"}, +]) + +tid_creatives = setup_table(db_id, "creative_analyses", [ + {"name": "ad_id", "type": "text"}, + {"name": "ad_name", "type": "text"}, + {"name": "campaign_id", "type": "text"}, + {"name": "image_url", "type": "url"}, + {"name": "analysis", "type": "long_text"}, + {"name": "score", "type": "number", "number_decimal_places": 1}, + {"name": "recommendations", "type": "long_text"}, + {"name": "created_at", "type": "date", "date_format": "ISO", "date_include_time": True}, +]) + +tid_logs = setup_table(db_id, "execution_logs", [ + {"name": "executed_at", "type": "date", "date_format": "ISO", "date_include_time": True}, + {"name": "mode", "type": "text"}, + {"name": "campaigns_analyzed", "type": "number"}, + {"name": "actions_proposed", "type": "number"}, + {"name": "actions_executed", "type": "number"}, + {"name": "errors", "type": "long_text"}, + {"name": "summary", "type": "long_text"}, + {"name": "duration_seconds", "type": "number", "number_decimal_places": 1}, +]) + +# ── 4. Output env vars ──────────────────────────────────────────────────────── + +print(f""" +{'='*55} + Add these to your .env file: +{'='*55} +BASEROW_DB_ID={db_id} +BASEROW_TABLE_CAMPAIGNS={tid_campaigns} +BASEROW_TABLE_ACTIONS={tid_actions} +BASEROW_TABLE_CREATIVES={tid_creatives} +BASEROW_TABLE_LOGS={tid_logs} +{'='*55} +""") diff --git a/slack_notifier.py b/slack_notifier.py new file mode 100644 index 0000000..4d07e7c --- /dev/null +++ b/slack_notifier.py @@ -0,0 +1,340 @@ +"""Slack notifier — Web API (Bot Token) con botones interactivos.""" +from datetime import datetime +import requests +import config + +_SLACK_API = "https://slack.com/api" + +_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_DISPLAY = { + "INCREASE_BUDGET": ("🟢", "AUMENTAR PRESUPUESTO"), + "REDUCE_BUDGET": ("🔴", "REDUCIR PRESUPUESTO"), + "PAUSE": ("⛔", "PAUSAR CAMPAÑA"), + "REVIEW_CREATIVES": ("🔍", "REVISAR CREATIVIDADES"), + "MAINTAIN": ("✅", "MANTENER"), +} + +# Solo estas acciones ejecutan algo real en Meta API → botones +_ACTIONABLE = {"INCREASE_BUDGET", "REDUCE_BUDGET", "PAUSE"} + + +def _effect_text(action: dict, budget: float | None) -> str: + """Texto que describe exactamente qué ocurrirá si se aprueba la acción.""" + atype = action.get("action_type", "") + param = float(action.get("parameter") or 1.0) + if atype == "PAUSE": + return "⚠️ *La campaña será pausada en Meta Ads*" + if atype in ("INCREASE_BUDGET", "REDUCE_BUDGET"): + pct = round((param - 1) * 100) + if pct == 0: + return "" # parameter not available, skip misleading text + sign = "+" if pct >= 0 else "" + if budget: + new_b = round(budget * param, 2) + return f"📊 Presupuesto diario: *{budget:.0f}€ → {new_b:.0f}€* ({sign}{pct}%)" + return f"📊 Ajuste de presupuesto: *{sign}{pct}%*" + return "" + + +def _post(method: str, **payload) -> dict: + resp = requests.post( + f"{_SLACK_API}/{method}", + headers={"Authorization": f"Bearer {config.SLACK_BOT_TOKEN}"}, + json=payload, + timeout=10, + ) + return resp.json() + + +def update_message(channel: str, ts: str, text: str): + """Reemplaza un mensaje con texto plano tras aprobación/rechazo.""" + _post("chat.update", channel=channel, ts=ts, text=text, blocks=[]) + + +def _adset_ad_table(items: list, label: str, show_bid: bool = False) -> str: + """Genera tabla monoespaciada de adsets o anuncios para Slack.""" + if not items: + return "" + lines = [f"*{label}*"] + lines.append("```") + if show_bid: + lines.append(f"{'Nombre':<32} {'Gasto':>6} {'Leads':>5} {'CPL':>6} {'CTR':>5} {'Cap':>7}") + lines.append("─" * 66) + else: + lines.append(f"{'Nombre':<32} {'Gasto':>6} {'Leads':>5} {'CPL':>6} {'CTR':>5}") + lines.append("─" * 58) + for it in items: + name = it["name"][:32] + leads_str = f"{it['leads']:>5}" if it["leads"] > 0 else " —" + cpl_str = f"{it['cpl']:>5.2f}€" if it["leads"] > 0 else " —" + if show_bid: + cost_cap = it.get("cost_cap_eur") + cap_str = f" {cost_cap:>5.2f}€" if cost_cap else " Auto" + else: + cap_str = "" + lines.append( + f"{name:<32} {it['spend']:>5.0f}€ {leads_str} {cpl_str} {it['ctr']:>4.1f}%{cap_str}" + ) + lines.append("```") + # Evaluations below the table + for it in items: + ev = it.get("evaluacion", "") + rec = it.get("recomendacion", "") + if ev or rec: + lines.append(f"• *{it['name'][:40]}*") + if ev: + lines.append(f" _{ev}_") + if rec: + lines.append(f" → {rec}") + return "\n".join(lines) + + +def send_daily_report( + daily_totals: list, + best_campaigns: list, + worst_campaigns: list, + actions: list, + target_cpl: float, + campaigns_analyzed: int, + mode: str = "DRY_RUN", + verticals: dict = None, + campaign_details: dict = None, +) -> str | None: + """Envía el informe diario consolidado. Devuelve el ts del mensaje.""" + now = datetime.now() + date_label = now.strftime("%d/%m/%Y") + month_name = now.strftime("%B %Y").capitalize() + prefix = config.META_CAMPAIGN_PREFIX + mode_label = "DRY RUN" if mode == "DRY_RUN" else "PRODUCCIÓN" + target_str = f"{target_cpl:.2f}€" if target_cpl > 0 else "—" + + blocks: list = [ + { + "type": "header", + "text": {"type": "plain_text", + "text": f"Meta Optimizer — {prefix} — {date_label} ({mode_label})"}, + }, + ] + + # ── Rentabilidad mensual con margen ─────────────────────────────────────── + if daily_totals: + lines = [f"{'Día':<5} {'Gasto':>7} {'Leads':>5} {'CPL':>7} {'Margen':>8} Est"] + lines.append("─" * 42) + total_spend = total_leads = total_margin = 0.0 + for d in daily_totals: + day = d["date"][8:10] + "/" + d["date"][5:7] + margin = d.get("margin", 0.0) + total_spend += d["spend"] + total_leads += d["leads"] + total_margin += margin + icon = "✅" if d["leads"] > 0 else ("❌" if d["spend"] > 0 else "—") + m_sign = f"+{margin:.0f}€" if margin >= 0 else f"{margin:.0f}€" + lines.append( + f"{day:<5} {d['spend']:>6.0f}€ {d['leads']:>5} {d['cpl']:>6.2f}€ {m_sign:>8} {icon}" + ) + lines.append("─" * 42) + total_cpl = round(total_spend / total_leads, 2) if total_leads > 0 else 0.0 + m_tot_sign = f"+{total_margin:.0f}€" if total_margin >= 0 else f"{total_margin:.0f}€" + lines.append( + f"{'TOTAL':<5} {total_spend:>6.0f}€ {int(total_leads):>5} {total_cpl:>6.2f}€ {m_tot_sign:>8}" + ) + blocks.append({ + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"*Rentabilidad {month_name}*\n```" + "\n".join(lines) + "```", + }, + }) + else: + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": "_Sin datos del mes en curso aún._"}, + }) + + # ── Resumen por vertical ────────────────────────────────────────────────── + if verticals: + blocks.append({"type": "divider"}) + lines = [f"{'Vertical':<16} {'Gasto':>7} {'Leads':>5} {'CPL':>7} {'Obj':>7} {'Margen':>9}"] + lines.append("─" * 57) + 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 + v_m = data["margin"] + 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 " —" + lines.append(f"{v:<16} {v_spend:>6.0f}€ {v_leads:>5} {v_cpl:>6.2f}€ {obj_str:>7} {m_sign:>9}") + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", + "text": "*Resumen por vertical (ayer)*\n```" + "\n".join(lines) + "```"}, + }) + + blocks.append({"type": "divider"}) + + # ── Top 10 mejores ──────────────────────────────────────────────────────── + if best_campaigns: + lines = [] + for i, m in enumerate(best_campaigns, 1): + lines.append( + f"{i}. *{m['name'][:46]}* CPL: {m['cpl']:.2f}€ | {m['leads']} leads | {m['spend']:.0f}€" + ) + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": "*Top 10 mejores (ayer)*\n" + "\n".join(lines)}, + }) + blocks.append({"type": "divider"}) + + # ── Top 10 peores ───────────────────────────────────────────────────────── + if worst_campaigns: + lines = [] + for i, m in enumerate(worst_campaigns, 1): + if m["leads"] == 0: + lines.append(f"{i}. *{m['name'][:46]}* 0 leads | {m['spend']:.0f}€ gastado") + else: + lines.append( + f"{i}. *{m['name'][:46]}* CPL: {m['cpl']:.2f}€ | {m['leads']} leads | {m['spend']:.0f}€" + ) + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": "*Top 10 peores (ayer)*\n" + "\n".join(lines)}, + }) + + blocks.append({"type": "divider"}) + + # ── Análisis por campaña: decisión + adsets + anuncios ──────────────────── + if campaign_details: + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", + "text": f"*Análisis por campaña ({campaigns_analyzed} activas ayer)*"}, + }) + # Build lookup: campaign_name → action info + action_map = {a["campaign_name"]: a for a in actions} + + for cid, detail in campaign_details.items(): + name = detail["name"] + vertical = detail["vertical"] + margin = detail["margin"] + adsets = detail.get("adsets", []) + ads = detail.get("ads", []) + bid_cfg = detail.get("bid_config", {}) + m_str = f"+{margin:.2f}€" if margin >= 0 else f"{margin:.2f}€" + action = action_map.get(name) + + strategy = bid_cfg.get("bid_strategy", "") + strategy_label = _STRATEGY_LABELS.get(strategy, strategy or "—") + budget = bid_cfg.get("daily_budget_eur") + budget_str = f"{budget:.0f}€/día" if budget else "—" + + # Icono y etiqueta de acción + atype = action["action_type"] if action else "MAINTAIN" + emoji, alabel = _ACTION_DISPLAY.get(atype, ("⚪", atype)) + + # Campaign header + header_text = ( + f"{emoji} *{name}*\n" + f"Vertical: _{vertical}_ | Margen: *{m_str}* | " + f"Estrategia: `{strategy_label}` | Presupuesto: {budget_str}\n" + f"*Decisión: {alabel}*" + ) + if action: + justification = action.get("justification", "")[:500] + header_text += f"\n_{justification}_" + if action.get("alert"): + header_text += f"\n:warning: {action['alert']}" + + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": header_text}, + }) + + # Botones solo para acciones que ejecutan algo en Meta API + if action and atype in _ACTIONABLE: + effect = _effect_text(action, budget) + if effect: + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": effect}, + }) + blocks.append({ + "type": "actions", + "elements": [ + { + "type": "button", + "text": {"type": "plain_text", "text": "✅ Aprobar"}, + "style": "primary", + "value": f"approve:{action['row_id']}", + "action_id": f"approve_{action['row_id']}", + }, + { + "type": "button", + "text": {"type": "plain_text", "text": "❌ Rechazar"}, + "style": "danger", + "value": f"reject:{action['row_id']}", + "action_id": f"reject_{action['row_id']}", + }, + ], + }) + + # Tabla adsets + anuncios + detail_parts = [] + adset_table = _adset_ad_table(adsets, "Conjuntos de anuncios", show_bid=True) + ad_table = _adset_ad_table(ads, "Anuncios") + if adset_table: + detail_parts.append(adset_table) + if ad_table: + detail_parts.append(ad_table) + if detail_parts: + combined = "\n\n".join(detail_parts) + # Split into chunks if too long (Slack block text limit: 3000 chars) + for chunk in [combined[i:i+2900] for i in range(0, len(combined), 2900)]: + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", "text": chunk}, + }) + + blocks.append({"type": "divider"}) + + if len(blocks) >= 45: # Safety margin before Slack's 50-block limit + blocks.append({ + "type": "section", + "text": {"type": "mrkdwn", + "text": "_... más campañas disponibles en Baserow._"}, + }) + break + + # ── Pie ─────────────────────────────────────────────────────────────────── + blocks.append({ + "type": "context", + "elements": [{"type": "mrkdwn", + "text": f"{campaigns_analyzed} campañas analizadas ayer"}], + }) + + result = _post( + "chat.postMessage", + channel=config.SLACK_CHANNEL_ID, + blocks=blocks, + text=f"Meta Optimizer — {prefix} — {date_label}", + ) + return result.get("ts") if result.get("ok") else None + + +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" + text = ( + f":bar_chart: *Meta Optimizer — Resumen diario* ({mode_label})\n" + f"• Campañas analizadas: {log.get('campaigns_analyzed', 0)}\n" + f"• Acciones propuestas: {log.get('actions_proposed', 0)}\n" + f"• Acciones ejecutadas: {log.get('actions_executed', 0)}\n" + f"• Duración: {log.get('duration_seconds', 0):.1f}s" + ) + _post("chat.postMessage", channel=config.SLACK_CHANNEL_ID, text=text)