meta-optimizer/agent.py

import json
import base64
import requests
import anthropic
import config

client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY)

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.

Recibirás métricas en tres ventanas temporales:
- spend/leads/cpl: ayer (1 día, contexto inmediato)
- spend_3d/leads_3d/cpl_3d: últimos 3 días (ventana PRINCIPAL para decisiones)
- spend_7d/leads_7d/cpl_7d: últimos 7 días (tendencia de fondo)

REGLAS DE DECISIÓN (basadas principalmente en cpl_3d y leads_3d):
1. cpl_3d > max_cpl → REDUCE_BUDGET (nunca PAUSE por CPL alto).
2. cpl_3d <= max_cpl con bajo volumen → INCREASE_BUDGET si hay margen.
3. Si cpl_3d < cpl_7d → tendencia mejorando: más permisivo con INCREASE_BUDGET.
4. Si cpl_3d > cpl_7d × 1.2 → tendencia empeorando: más estricto, considera REDUCE_BUDGET.
5. CTR < 1% → problema de creatividad, revisar anuncios.
6. PAUSE solo si: leads_3d == 0 Y spend_3d >= 3 × max_cpl (fallo técnico grave de tracking/píxel).
   En cualquier otro caso de bajo rendimiento usa REDUCE_BUDGET, nunca PAUSE.

Responde SOLO con JSON válido, sin texto adicional ni markdown:
{
  "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
}
"""


def decide(analysis: dict) -> dict:
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=400,
        system=DECIDE_SYSTEM,
        messages=[{
            "role": "user",
            "content": (
                "Analyze this Meta Ads campaign and return the decision as JSON:\n\n"
                + json.dumps(analysis, ensure_ascii=False, indent=2)
            ),
        }],
    )
    raw = response.content[0].text.strip()
    clean = raw.replace("```json", "").replace("```", "").strip()
    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 {
            "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 indicado.
Los datos corresponden a los últimos 3 días (ventana estándar de análisis).
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"}
"""

AD_SYSTEM = """
Eres un analista experto en Meta Ads. Analiza las métricas del anuncio indicado.
Los datos incluyen dos ventanas temporales:
- cpl_3d / leads_3d / spend_3d: últimos 3 días (puede ser volátil con poco volumen)
- cpl_7d / leads_7d: últimos 7 días (más estable, úsala como referencia principal)
Los campos spend/leads/cpl sin sufijo corresponden a 7 días.
USA SIEMPRE € como unidad de moneda. Responde SIEMPRE en español.
Responde SOLO con JSON válido (sin markdown):
{"evaluacion": "resumen del rendimiento en 2 frases usando €, mencionando diferencia 3d/7d si es relevante", "recomendacion": "una acción concreta", "accion": "PAUSE o MAINTAIN"}

Reglas para "accion": "PAUSE":
- SOLO si leads_7d == 0 Y spend de 7 días supera 3 veces el max_cpl (o 15€ si no existe max_cpl).
- Si cpl_3d es alto pero cpl_7d está dentro del objetivo, usa "MAINTAIN" (ruido estadístico de período corto).
- Si el anuncio tiene leads en 7 días aunque el CPL sea alto, usa "MAINTAIN" y recomienda optimizar.
En cualquier otro caso usa "accion": "MAINTAIN".
"""


def analyze_unit(metrics: dict, level: str = "adset") -> dict:
    """Análisis rápido de un conjunto de anuncios o anuncio individual."""
    nivel = "conjunto de anuncios" if level == "adset" else "anuncio"
    system = AD_SYSTEM if level == "ad" else UNIT_SYSTEM
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=250,
        system=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": ""}