375 lines
15 KiB
Python
375 lines
15 KiB
Python
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 = """
|
||
IDIOMA: Responde SIEMPRE en español. Todos los campos del JSON deben estar en español.
|
||
|
||
Eres un experto en análisis de creatividades de Meta Ads.
|
||
Analiza la imagen publicitaria y devuelve SOLO JSON válido sin markdown:
|
||
{
|
||
"score": 7.5,
|
||
"analysis": "análisis conciso en español: mensaje, diseño, CTA, atractivo visual",
|
||
"recommendations": "mejoras concretas en español para mejorar CTR y conversiones"
|
||
}
|
||
Score 1-10: 1-3 crítico, 4-5 bajo, 6-7 aceptable, 8-9 bueno, 10 excelente.
|
||
"""
|
||
|
||
|
||
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": ""}
|
||
|
||
|
||
CREATIVE_DEEP_SYSTEM = """
|
||
IDIOMA: Responde SIEMPRE en español. Todos los campos del JSON deben estar en español.
|
||
|
||
Eres un experto en análisis de creatividades de Meta Ads con conocimiento de neuromarketing y diseño persuasivo.
|
||
Recibirás una imagen publicitaria junto con sus métricas de rendimiento reales.
|
||
|
||
Evalúa considerando:
|
||
1. CALIDAD VISUAL: claridad del mensaje, jerarquía visual, CTA, copy, atractivo y relevancia
|
||
2. CORRELACIÓN CON RENDIMIENTO: ¿el CTR y CPL reales son consistentes con la calidad visual?
|
||
3. SEÑAL DE FATIGA: si CTR 3d < CTR 7d × 0.75 indica saturación de audiencia
|
||
4. RECOMENDACIONES: mejoras concretas y priorizadas para mejorar CTR y conversiones
|
||
|
||
Devuelve SOLO JSON válido sin markdown (todos los textos en español):
|
||
{
|
||
"score": 7.5,
|
||
"analysis": "análisis conciso en español: qué funciona, qué no, correlación con rendimiento real",
|
||
"recommendations": "mejoras concretas en español en orden de impacto esperado",
|
||
"fatigue": false,
|
||
"fatigue_reason": null
|
||
}
|
||
|
||
Score 1-10: 1-3 crítico (pausar), 4-5 bajo, 6-7 aceptable, 8-9 bueno, 10 excelente.
|
||
Si el anuncio tiene buen rendimiento real (CPL bajo, CTR alto) pero diseño mediocre, sube el score.
|
||
Si el diseño parece bueno pero el rendimiento es pobre, baja el score y explica la desconexión.
|
||
"""
|
||
|
||
CREATIVE_COMPARE_SYSTEM = """
|
||
IDIOMA: Responde SIEMPRE en español. Todos los campos del JSON deben estar en español.
|
||
|
||
Eres un experto en análisis comparativo de creatividades de Meta Ads.
|
||
Recibirás varios anuncios del mismo adset con sus imágenes y métricas de rendimiento.
|
||
Evalúa cuál funciona mejor considerando tanto calidad visual como rendimiento real.
|
||
|
||
Devuelve SOLO JSON válido sin markdown (todos los textos en español):
|
||
{
|
||
"winner": "nombre exacto del anuncio ganador",
|
||
"ranking": [
|
||
{"name": "nombre completo del anuncio", "rank": 1, "reason": "razón en español"}
|
||
],
|
||
"insights": "observación comparativa clave en español: ¿qué diferencia visualmente al ganador del resto?"
|
||
}
|
||
"""
|
||
|
||
|
||
def _download_image(image_url) -> tuple | None:
|
||
"""Returns (base64_data, media_type) or None. Accepts str or list of URLs (tries in order)."""
|
||
urls = [image_url] if isinstance(image_url, str) else image_url
|
||
for url in urls:
|
||
if not url:
|
||
continue
|
||
try:
|
||
resp = requests.get(url, timeout=15)
|
||
resp.raise_for_status()
|
||
content_type = resp.headers.get("content-type", "image/jpeg")
|
||
if not content_type.startswith("image/"):
|
||
continue
|
||
data = base64.standard_b64encode(resp.content).decode("utf-8")
|
||
return data, content_type.split(";")[0]
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _parse_json_response(raw: str) -> dict:
|
||
import re
|
||
clean = re.sub(r"```json\s*", "", raw.strip())
|
||
clean = re.sub(r"```\s*", "", clean).strip()
|
||
try:
|
||
return json.loads(clean)
|
||
except json.JSONDecodeError:
|
||
start, end = clean.find("{"), clean.rfind("}")
|
||
if start != -1 and end > start:
|
||
try:
|
||
return json.loads(clean[start:end + 1])
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def analyze_creative_deep(image_url: str, ad_name: str, metrics: dict) -> dict:
|
||
"""Deep creative analysis combining visual quality with performance data and fatigue detection."""
|
||
_default = {"score": 0, "analysis": "", "recommendations": "", "fatigue": False, "fatigue_reason": None}
|
||
|
||
downloaded = _download_image(image_url)
|
||
if not downloaded:
|
||
return {**_default, "analysis": "Error descargando imagen."}
|
||
image_data, media_type = downloaded
|
||
|
||
ctr_7d = metrics.get("ctr_7d", 0)
|
||
ctr_3d = metrics.get("ctr_3d", 0)
|
||
fatigue_hint = ""
|
||
if ctr_7d > 0 and ctr_3d > 0 and ctr_3d < ctr_7d * 0.75:
|
||
fatigue_hint = f"\n⚠️ SEÑAL DE FATIGA DETECTADA: CTR cayó de {ctr_7d:.2f}% (7d) a {ctr_3d:.2f}% (3d) — posible saturación"
|
||
|
||
context = (
|
||
f'Anuncio: "{ad_name}"\n\n'
|
||
f"Métricas reales:\n"
|
||
f"- 7 días: gasto {metrics.get('spend_7d', 0):.0f}€, "
|
||
f"{metrics.get('leads_7d', 0)} leads, "
|
||
f"CPL {metrics.get('cpl_7d', 0):.2f}€, "
|
||
f"CTR {ctr_7d:.2f}%\n"
|
||
f"- 3 días: gasto {metrics.get('spend_3d', 0):.0f}€, "
|
||
f"{metrics.get('leads_3d', 0)} leads, "
|
||
f"CPL {metrics.get('cpl_3d', 0):.2f}€, "
|
||
f"CTR {ctr_3d:.2f}%\n"
|
||
f"- Objetivo CPL máximo: {metrics.get('max_cpl', 0):.2f}€"
|
||
f"{fatigue_hint}\n\n"
|
||
f"Analiza esta creatividad:"
|
||
)
|
||
|
||
try:
|
||
response = client.messages.create(
|
||
model="claude-sonnet-4-6",
|
||
max_tokens=700,
|
||
system=CREATIVE_DEEP_SYSTEM,
|
||
messages=[{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "image", "source": {"type": "base64", "media_type": media_type, "data": image_data}},
|
||
{"type": "text", "text": context},
|
||
],
|
||
}],
|
||
)
|
||
result = _parse_json_response(response.content[0].text)
|
||
if not result:
|
||
return {**_default, "analysis": "Error parseando respuesta."}
|
||
result.setdefault("fatigue", False)
|
||
result.setdefault("fatigue_reason", None)
|
||
return result
|
||
except Exception as e:
|
||
return {**_default, "analysis": f"Error en análisis: {e}"}
|
||
|
||
|
||
def compare_adset_creatives(ads: list) -> dict:
|
||
"""Compare up to 4 ads within the same adset. ads: list of analyzed ad dicts."""
|
||
_default = {"winner": "", "ranking": [], "insights": "Sin datos suficientes para comparar."}
|
||
|
||
top_ads = sorted(ads, key=lambda x: -x.get("spend_7d", 0))[:4]
|
||
content_blocks = []
|
||
|
||
for i, ad in enumerate(top_ads, 1):
|
||
downloaded = _download_image(ad["image_url"])
|
||
if downloaded:
|
||
image_data, media_type = downloaded
|
||
content_blocks.append({
|
||
"type": "image",
|
||
"source": {"type": "base64", "media_type": media_type, "data": image_data},
|
||
})
|
||
fatigue_note = f" ⚠️ Fatiga: {ad.get('fatigue_reason','')}" if ad.get("fatigue") else ""
|
||
content_blocks.append({
|
||
"type": "text",
|
||
"text": (
|
||
f"Anuncio {i}: \"{ad['ad_name']}\"\n"
|
||
f"Score: {ad.get('score', 0):.1f}/10 | "
|
||
f"CTR 7d: {ad.get('ctr_7d', 0):.2f}% | "
|
||
f"CPL 7d: {ad.get('cpl_7d', 0):.2f}€ | "
|
||
f"Leads 7d: {ad.get('leads_7d', 0)} | "
|
||
f"Gasto 7d: {ad.get('spend_7d', 0):.0f}€"
|
||
f"{fatigue_note}"
|
||
),
|
||
})
|
||
|
||
if not content_blocks:
|
||
return _default
|
||
|
||
try:
|
||
response = client.messages.create(
|
||
model="claude-sonnet-4-6",
|
||
max_tokens=600,
|
||
system=CREATIVE_COMPARE_SYSTEM,
|
||
messages=[{"role": "user", "content": content_blocks}],
|
||
)
|
||
result = _parse_json_response(response.content[0].text)
|
||
return result if result else _default
|
||
except Exception as e:
|
||
return {**_default, "insights": f"Error en comparativa: {e}"}
|