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.

REGLAS DE DECISIÓ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.

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.
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.
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 €", "recomendacion": "una acción concreta", "accion": "PAUSE o MAINTAIN"}
Usa "accion": "PAUSE" solo si: el anuncio ha gastado más de 5€ con 0 leads, o su CPL es más del doble del objetivo sin signos de mejora.
En caso contrario 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": ""}