From 7228cf1b650eebff6a7593d605c21bc3b1f13032 Mon Sep 17 00:00:00 2001 From: Rompetechos cuenta de desarrollo Date: Thu, 9 Jul 2026 16:49:00 +0200 Subject: [PATCH] =?UTF-8?q?Integraci=C3=B3n=20con=20OpenCode?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/settings.json | 18 + .claude/settings.local.json | 28 +- AGENTS.md | 84 +++++ README.md | 160 ++++---- agent.py | 508 ++++++++++++------------- analyzer.py | 138 +++---- config.py | 52 +-- google_ads_client.py | 730 ++++++++++++++++++------------------ 8 files changed, 910 insertions(+), 808 deletions(-) create mode 100644 .claude/settings.json create mode 100644 AGENTS.md diff --git a/.claude/settings.json b/.claude/settings.json new file mode 100644 index 0000000..a162502 --- /dev/null +++ b/.claude/settings.json @@ -0,0 +1,18 @@ +{ + "permissions": { + "allow": [ + "Bash(python backfill_games_mayo.py)", + "Bash(findstr /i \"GAMes Actualiz MetricasDiarias actualizado\" \"C:\\\\Users\\\\jmgom\\\\AppData\\\\Local\\\\Temp\\\\claude\\\\c--Users-jmgom-projects-leads-optimizer\\\\2e73bc62-0b35-4293-96e9-676bded17b5f\\\\tasks\\\\bf9cfc1r9.output\")", + "Bash(findstr /i /c:\"actualiz\" \"C:\\\\Users\\\\jmgom\\\\AppData\\\\Local\\\\Temp\\\\claude\\\\c--Users-jmgom-projects-leads-optimizer\\\\2e73bc62-0b35-4293-96e9-676bded17b5f\\\\tasks\\\\bf9cfc1r9.output\")", + "Bash(python -c ' *)", + "Bash(git commit -m ' *)", + "Bash(git push *)", + "Bash(pkill -f \"streamlit run dashboard.py\")", + "Skill(run)", + "Skill(run:*)" + ], + "additionalDirectories": [ + "C:\\Users\\jmgom\\projects\\meta-optimizer" + ] + } +} diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 525b0be..c4f9e5d 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -1,14 +1,14 @@ -{ - "permissions": { - "allow": [ - "Bash(pip install *)", - "Bash(python -m py_compile run.py airtable_client.py google_ads_client.py)", - "Bash(python -m py_compile run.py airtable_client.py slack_reporter.py)", - "Bash(python -m py_compile run.py)", - "Bash(python run.py)", - "Bash(git add *)", - "Bash(python migrate_leads_field.py)", - "Bash(python backfill_leads_google_mayo.py)" - ] - } -} +{ + "permissions": { + "allow": [ + "Bash(pip install *)", + "Bash(python -m py_compile run.py airtable_client.py google_ads_client.py)", + "Bash(python -m py_compile run.py airtable_client.py slack_reporter.py)", + "Bash(python -m py_compile run.py)", + "Bash(python run.py)", + "Bash(git add *)", + "Bash(python migrate_leads_field.py)", + "Bash(python backfill_leads_google_mayo.py)" + ] + } +} diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..1cd3663 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,84 @@ +# Leads Optimizer — AGENTS.md + +## Quick start + +```bash +pip install -r requirements.txt +python run.py # main optimizer run +python weekly_report.py # weekly strategic report +streamlit run dashboard.py # Streamlit dashboard (port 15002) +``` + +Python 3.12+. No build step, no linter, no test suite. + +## Execution entry points + +| File | Trigger | What it does | +|------|---------|-------------| +| `run.py` | daily.yml (00:00 UTC) or manual | Syncs Google Ads → Airtable, analyses campaigns, applies budget decisions, reports to Slack | +| `weekly_report.py` | weekly.yml (Mon 07:00 UTC) or manual | Deeper week-over-week analysis via Anthropic, sends Slack | +| `dashboard.py` | `streamlit run dashboard.py` | UI on port 15002, reads from Airtable directly | + +## DRY_RUN mode + +`config.py` has `DRY_RUN = True` by default. When `True`, decisions are printed but **no changes are applied to Google Ads**. Must be set to `False` to apply budget/pause changes. This is a module-level constant, not an env var. + +## Required env vars (loaded via `python-dotenv` from `.env`) + +All are mandatory except `SLACK_WEBHOOK_URL` (optional, has empty default): + +- `AIRTABLE_TOKEN`, `AIRTABLE_BASE_ID` +- `Google Ads`: `GOOGLE_ADS_DEVELOPER_TOKEN`, `GOOGLE_ADS_CLIENT_ID`, `GOOGLE_ADS_CLIENT_SECRET`, `GOOGLE_ADS_REFRESH_TOKEN`, `GOOGLE_ADS_LOGIN_CUSTOMER_ID` (digits only, no dashes) +- `ANTHROPIC_API_KEY` +- `SLACK_WEBHOOK_URL` + +The `.github/workflows/` files pull these from GitHub Secrets. `run.sh` has **hardcoded secrets** — never commit changes to it. + +## Architecture + +Flat structure, no packages. Flow: + +1. **Sync**: `airtable_client.py` + `google_ads_client.py` pull campaign catalog and monthly metrics from Google Ads, write to Airtable tables `Google Ads Campaigns`, `GACampaignMes`, `Leads Lake`, `MetricasDiarias`, `GAMes`. +2. **Analyze**: `analyzer.py` computes per-campaign urgency (PAUSAR / SPRINT / ACELERAR / FRENAR / EN_RITMO). +3. **Decide**: `agent.py` calls **Anthropic Claude (claude-sonnet-4-6)** per campaign to get JSON decision with action, budget multiplier, justification, and advice. Also produces portfolio-level analysis. +4. **Apply**: `optimizer.py` mutates Google Ads campaigns (budget changes, pause/unpause) — only if `DRY_RUN = False`. +5. **Report**: `slack_reporter.py` sends formatted summary to Slack via webhook. + +## Campaign naming conventions + +Campaigns follow naming patterns that drive logic: + +- `fco_search_` — Search campaigns for formation courses +- `fco_pmx_` — PMX (Performance Max) campaigns +- `fco_leadform_` — Lead form campaigns (leads captured inside Google, never reach Airtable) + +When a course has both Search and PMX campaigns (`_search_` + `_pmx_` with same ``), Search conversions reflow to PMX. When PMX has a `_leadform` companion, leadform conversions are summed into the PMX campaign's `leads_grupo`. If a course has multiple PMX campaigns (excluding leadform), paths 4/5 are disabled to avoid double counting. + +## Month-boundary metric handling + +`run.py` has special logic (lines ~400-432) for when yesterday belongs to a different calendar month. It merges into the **previous month's** `GACampaignMes` record instead of overwriting the new month's (avoids a known bug that erased June history). Any edits to the metrics-writing section must preserve this redirect. + +## Airtable tables + +- `Google Ads Campaigns` — master campaign catalog (Curso, GoogleCampaignID, PPL, CapTotalMes, CPAMaximo, Activa) +- `Leads Lake` — individual lead records (GoogleCampaignID, FechaEntrada) +- `GACampaignMes` — per-campaign monthly snapshot; updated each run with leads, advice, criticidad, metricas_diarias +- `MetricasDiarias` — JSON field with per-day {coste, ingreso, margen, leads, leads_lake} +- `GAMes` — aggregated daily totals for all fco_ campaigns + monthly totals + +## Backfill and migration scripts + +Scripts prefixed `backfill_*` and `migrate_*` are one-off data scripts. Do not call them from normal flow; they are only for historical data repairs. + +## CI/CD + +- `daily.yml` — runs `python run.py` at 00:00 UTC (2 AM CEST / 1 AM CET) +- `weekly.yml` — runs `python weekly_report.py` Monday at 07:00 UTC (9 AM CEST) +- Both use `ubuntu-latest` + Python 3.12 + secrets from GitHub +- Logs uploaded as artifacts (30-day retention) + +## Gotchas + +- `run.py` wraps `sys.stdout` with a custom `Tee` class that writes to both console and `logs/.log`. Do not remove this. +- The Anthropic model is hardcoded to `claude-sonnet-4-6` in `agent.py`. If the model changes, update all three calls (`decide`, `portfolio_daily_analysis`, `weekly_strategic_analysis`). +- `run.sh` contains real API secrets — it should NOT be committed. Use `.env` + `python run.py` instead. diff --git a/README.md b/README.md index b814c87..7d2f518 100644 --- a/README.md +++ b/README.md @@ -1,80 +1,80 @@ -# Leads Optimizer — Formación - -Agente de optimización automática de campañas Google Ads para generación de leads de formación. -Cruza datos de Airtable (leads reales) con métricas de Google Ads y decide ajustes de presupuesto. - ---- - -## Campos requeridos en Airtable - -### Tabla: "Google Ads Campaigns" -| Campo | Tipo | Descripción | -|-------------------|---------|--------------------------------------| -| Curso | Text | Nombre del curso | -| GoogleCampaignID | Number | ID de campaña en Google Ads | -| PPL | Number | Precio por lead (€) | -| CapTotalMes | Number | Capping mensual de leads | -| CPAMaximo | Number | CPA máximo tolerable (€) | -| Activa | Boolean | TRUE para incluir en el análisis | - -### Tabla: "Leads Lake" -| Campo | Tipo | Descripción | -|-------------------|---------|--------------------------------------| -| GoogleCampaignID | Text | ID de campaña de origen | -| FechaEntrada | Date | Fecha del lead (formato YYYY-MM-DD) | - ---- - -## Variables de entorno - -```bash -export AIRTABLE_TOKEN=pat_xxxxxxxxxxxx -export AIRTABLE_BASE_ID=appXXXXXXXXXXXXXX - -export GOOGLE_ADS_DEVELOPER_TOKEN=xxxx -export GOOGLE_ADS_CLIENT_ID=xxxx.apps.googleusercontent.com -export GOOGLE_ADS_CLIENT_SECRET=xxxx -export GOOGLE_ADS_REFRESH_TOKEN=xxxx -export GOOGLE_ADS_LOGIN_CUSTOMER_ID=1234567890 # sin guiones - -export ANTHROPIC_API_KEY=sk-ant-xxxx - -export SLACK_WEBHOOK_URL=xxx -``` - ---- - -## Instalación y ejecución - -```bash -# Instalar dependencias -pip install -r requirements.txt - -# Ejecutar en modo DRY RUN (recomendado para empezar) -# DRY_RUN = True en config.py → solo muestra decisiones, no aplica cambios -python run.py - -# Cuando estés seguro, cambiar DRY_RUN = False en config.py -python run.py -``` - ---- - -## Lógica de urgencia - -| Urgencia | Condición | Acción típica | -|-------------|--------------------------------------------------------|-----------------------| -| PAUSAR | leads >= capping | Pausa campaña | -| SPRINT | ritmo muy atrasado + quedan ≤ 5 días | +30-50% presupuesto | -| ACELERAR | ritmo atrasado > 15 puntos vs ratio del mes | +10-25% presupuesto | -| FRENAR | ritmo adelantado > 15 puntos vs ratio del mes | -10-25% presupuesto | -| EN_RITMO | dentro del margen esperado | Mantener | - ---- - -## Automatización con cron - -```bash -# Ejecutar cada día a las 8:00 -0 8 * * * cd /ruta/leads-optimizer && python run.py >> logs/optimizer.log 2>&1 -``` +# Leads Optimizer — Formación + +Agente de optimización automática de campañas Google Ads para generación de leads de formación. +Cruza datos de Airtable (leads reales) con métricas de Google Ads y decide ajustes de presupuesto. + +--- + +## Campos requeridos en Airtable + +### Tabla: "Google Ads Campaigns" +| Campo | Tipo | Descripción | +|-------------------|---------|--------------------------------------| +| Curso | Text | Nombre del curso | +| GoogleCampaignID | Number | ID de campaña en Google Ads | +| PPL | Number | Precio por lead (€) | +| CapTotalMes | Number | Capping mensual de leads | +| CPAMaximo | Number | CPA máximo tolerable (€) | +| Activa | Boolean | TRUE para incluir en el análisis | + +### Tabla: "Leads Lake" +| Campo | Tipo | Descripción | +|-------------------|---------|--------------------------------------| +| GoogleCampaignID | Text | ID de campaña de origen | +| FechaEntrada | Date | Fecha del lead (formato YYYY-MM-DD) | + +--- + +## Variables de entorno + +```bash +export AIRTABLE_TOKEN=pat_xxxxxxxxxxxx +export AIRTABLE_BASE_ID=appXXXXXXXXXXXXXX + +export GOOGLE_ADS_DEVELOPER_TOKEN=xxxx +export GOOGLE_ADS_CLIENT_ID=xxxx.apps.googleusercontent.com +export GOOGLE_ADS_CLIENT_SECRET=xxxx +export GOOGLE_ADS_REFRESH_TOKEN=xxxx +export GOOGLE_ADS_LOGIN_CUSTOMER_ID=1234567890 # sin guiones + +export ANTHROPIC_API_KEY=sk-ant-xxxx + +export SLACK_WEBHOOK_URL=xxx +``` + +--- + +## Instalación y ejecución + +```bash +# Instalar dependencias +pip install -r requirements.txt + +# Ejecutar en modo DRY RUN (recomendado para empezar) +# DRY_RUN = True en config.py → solo muestra decisiones, no aplica cambios +python run.py + +# Cuando estés seguro, cambiar DRY_RUN = False en config.py +python run.py +``` + +--- + +## Lógica de urgencia + +| Urgencia | Condición | Acción típica | +|-------------|--------------------------------------------------------|-----------------------| +| PAUSAR | leads >= capping | Pausa campaña | +| SPRINT | ritmo muy atrasado + quedan ≤ 5 días | +30-50% presupuesto | +| ACELERAR | ritmo atrasado > 15 puntos vs ratio del mes | +10-25% presupuesto | +| FRENAR | ritmo adelantado > 15 puntos vs ratio del mes | -10-25% presupuesto | +| EN_RITMO | dentro del margen esperado | Mantener | + +--- + +## Automatización con cron + +```bash +# Ejecutar cada día a las 8:00 +0 8 * * * cd /ruta/leads-optimizer && python run.py >> logs/optimizer.log 2>&1 +``` diff --git a/agent.py b/agent.py index 3791ce1..59d1689 100644 --- a/agent.py +++ b/agent.py @@ -1,254 +1,254 @@ -import json -from datetime import datetime -import anthropic -import config - -client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY) - -PORTFOLIO_SYSTEM = """ -Eres un experto en marketing de performance para una agencia de generación de leads en formación. -Recibes datos agregados del portfolio de campañas de Google Ads (solo campañas fco_). -Responde siempre en español, de forma concisa y accionable. Sin markdown, sin bullet symbols especiales, usa guiones simples (-). -""" - -WEEKLY_SYSTEM = """ -Eres un consultor senior de marketing de performance especializado en generación de leads para formación. -Recibes el análisis semanal del portfolio de campañas de Google Ads (solo campañas fco_). -Tu análisis debe ser estratégico, comparando la semana actual con la anterior, identificando tendencias y proponiendo acciones concretas. -Responde siempre en español. Sin markdown, sin bullet symbols especiales, usa guiones simples (-). -""" - -SYSTEM_PROMPT = """ -Eres un agente experto en optimización de campañas de generación de leads para centros de formación. -Cada campaña corresponde a un curso concreto con un PPL (precio por lead) fijo acordado con los centros compradores. - -MODELO DE NEGOCIO: -- Ingreso = leads_entregados × PPL -- Margen = (Ingreso - Gasto Google Ads) / Ingreso -- El objetivo es maximizar leads dentro del capping mensual manteniendo margen positivo. -- El CPA máximo ya refleja el margen mínimo aceptable. - -ESTADO DE LA CAMPAÑA: -- El campo status_google indica el estado actual en Google Ads: ENABLED (activa) o PAUSED (pausada). -- Nunca recomiendes reactivar una campaña si status_google = ENABLED (ya está activa). -- Si status_google = PAUSED y la acción es AUMENTAR_PRESUPUESTO, menciona en el consejo que primero hay que reactivarla. - -CAPPING: -- Si capping = 0, significa que no hay límite de leads configurado para este mes. No menciones el capping en el consejo ni en la justificación. Toma la decisión basándote únicamente en rentabilidad y ritmo. - -REGLAS DE DECISIÓN: -1. urgencia=PAUSAR → accion=PAUSAR siempre. El capping está lleno, seguir gastando destruye margen. -2. urgencia=SPRINT → accion=AUMENTAR_PRESUPUESTO con parametro entre 1.3 y 1.5. Quedan pocos días y leads por entregar. -3. urgencia=ACELERAR y campaña rentable → accion=AUMENTAR_PRESUPUESTO con parametro entre 1.1 y 1.25. -4. urgencia=ACELERAR y campaña NO rentable → accion=MANTENER o revisar keywords (no gastar más si no convierte). -5. urgencia=FRENAR → accion=REDUCIR_PRESUPUESTO con parametro entre 0.75 y 0.9. -6. urgencia=EN_RITMO y rentable → accion=MANTENER. -7. urgencia=EN_RITMO y NO rentable → accion=REDUCIR_PRESUPUESTO con parametro 0.85. -8. alerta_tracking=true → añadir alerta sobre discrepancia de tracking aunque la acción sea otra. - -Devuelve ÚNICAMENTE un JSON válido con esta estructura exacta, sin texto adicional ni markdown: -{ - "accion": "PAUSAR | REDUCIR_PRESUPUESTO | AUMENTAR_PRESUPUESTO | MANTENER", - "parametro": 1.0, - "nuevo_budget_diario": 0.0, - "justificacion": "explicación breve del porqué de la decisión", - "consejo": "acción concreta y específica que debería tomar el gestor (keywords, pujas, anuncios, configuración, etc.)", - "alerta": "texto si hay algo crítico, null si no hay", - "confianza": 0.0 -} - -El campo nuevo_budget_diario = budget_diario_actual × parametro (calcula tú el valor final). -El campo consejo debe ser accionable y específico: qué revisar, qué cambiar, qué hacer a continuación. -""" - - -def _classify_type(curso: str) -> str: - c = curso.lower() - if "_leadform" in c: - return "leadform" - if "_pmx" in c or "pmx_" in c: - return "pmx" - if "search" in c: - return "search" - return "otro" - - -def portfolio_daily_analysis(collected: list) -> str: - """Análisis estratégico diario del portfolio fco_. Devuelve texto plano para Slack.""" - from datetime import datetime - now = datetime.now() - fco = [i for i in collected if i["campaign"]["curso"].lower().startswith("fco_")] - - tipos: dict = {} - leadforms_detail = [] - alertas_tracking = 0 - campañas_perdida = 0 - - for item in fco: - t = _classify_type(item["campaign"]["curso"]) - m = item["metrics"] - a = item["analysis"] - cost = m.get("cost", 0) - conv = a["conversiones_google"] - ppl = item["campaign"]["ppl"] - rev = a["revenue_estimado"] - margen_pct = round((rev - cost) / rev * 100, 1) if rev > 0 else 0.0 - - if t not in tipos: - tipos[t] = {"campañas": 0, "inversion": 0.0, "conversiones": 0, "ingreso": 0.0} - tipos[t]["campañas"] += 1 - tipos[t]["inversion"] += cost - tipos[t]["conversiones"] += conv - tipos[t]["ingreso"] += rev - - if a.get("alerta_tracking"): - alertas_tracking += 1 - if rev > 0 and cost > rev: - campañas_perdida += 1 - - if t == "leadform": - leadforms_detail.append({ - "curso": item["campaign"]["curso"][:40], - "cpa_google": round(cost / conv, 2) if conv > 0 else None, - "conv_google": int(conv), - "conv_airtable": item["leads"], - "margen_pct": margen_pct, - }) - - resumen_tipos = {} - for t, d in tipos.items(): - cpa = round(d["inversion"] / d["conversiones"], 2) if d["conversiones"] > 0 else None - ing = d["ingreso"] - margen = round((ing - d["inversion"]) / ing * 100, 1) if ing > 0 else 0.0 - resumen_tipos[t] = { - "campañas": d["campañas"], - "inversion": round(d["inversion"], 2), - "conversiones": int(d["conversiones"]), - "cpa_medio": cpa, - "margen_pct": margen, - } - - data = { - "fecha": now.strftime("%d/%m/%Y"), - "dia_del_mes": now.day, - "campañas_totales": len(fco), - "campañas_en_perdida": campañas_perdida, - "alertas_tracking": alertas_tracking, - "rendimiento_por_tipo": resumen_tipos, - "detalle_leadforms": leadforms_detail, - } - - try: - response = client.messages.create( - model="claude-sonnet-4-6", - max_tokens=800, - system=PORTFOLIO_SYSTEM, - messages=[{ - "role": "user", - "content": ( - "Analiza estos datos del portfolio y proporciona:\n" - "1. Diagnóstico en 2 frases\n" - "2. Problemas principales (máx 3, con guión)\n" - "3. Acciones prioritarias (máx 3, muy concretas, con guión)\n" - "Si hay campañas leadform, evalúa específicamente su situación.\n\n" - f"{json.dumps(data, ensure_ascii=False, indent=2)}" - ), - }], - ) - return response.content[0].text.strip() - except Exception as e: - return f"Error generando análisis: {e}" - - -def weekly_strategic_analysis(games_md_this: dict, games_md_prev_week: dict, - collected: list, mes_nombre: str) -> str: - """ - Análisis estratégico semanal profundo. - games_md_this: MetricasDiarias de GAMes de los últimos 7 días (esta semana). - games_md_prev_week: MetricasDiarias de GAMes de los 7 días anteriores. - collected: lista de campañas del optimizer. - """ - def _week_summary(md: dict) -> dict: - coste = ing = leads = leads_lake = 0.0 - for v in md.values(): - coste += v.get("coste", 0) - ing += v.get("ingreso_sum", 0) - leads += v.get("leads", 0) - leads_lake += v.get("leads_lake", 0) - margen = round((ing - coste) / ing * 100, 1) if ing > 0 else 0.0 - cpa = round(coste / leads, 2) if leads > 0 else None - return {"coste": round(coste, 2), "ingreso": round(ing, 2), - "leads_google": int(leads), "leads_airtable": int(leads_lake), - "margen_pct": margen, "cpa": cpa} - - fco = [i for i in collected if i["campaign"]["curso"].lower().startswith("fco_")] - - # Top 5 peores por CPA del mes - peores = sorted( - [{"curso": i["campaign"]["curso"][:40], - "cpa": i["analysis"]["cpa_actual"], - "conv": int(i["analysis"]["conversiones_google"]), - "margen_pct": round(i["analysis"]["margen"] * 100, 1)} - for i in fco if i["analysis"]["cpa_actual"] > 0], - key=lambda x: x["cpa"], reverse=True - )[:5] - - data = { - "mes": mes_nombre, - "semana_actual": _week_summary(games_md_this), - "semana_anterior": _week_summary(games_md_prev_week), - "top5_peor_cpa_mes": peores, - "leadforms_activos": sum(1 for i in fco if "_leadform" in i["campaign"]["curso"].lower()), - "campañas_totales": len(fco), - } - - try: - response = client.messages.create( - model="claude-sonnet-4-6", - max_tokens=900, - system=WEEKLY_SYSTEM, - messages=[{ - "role": "user", - "content": ( - "Genera el informe estratégico semanal con:\n" - "1. Resumen ejecutivo (3 frases comparando esta semana con la anterior)\n" - "2. Tendencias clave detectadas (máx 4, con guión)\n" - "3. Situación campañas leadform y qué hacer con ellas\n" - "4. Acciones estratégicas prioritarias para la próxima semana (máx 4, muy concretas, con guión)\n" - "5. Una frase de conclusión sobre si el portfolio va en la dirección correcta\n\n" - f"{json.dumps(data, ensure_ascii=False, indent=2)}" - ), - }], - ) - return response.content[0].text.strip() - except Exception as e: - return f"Error generando análisis semanal: {e}" - - -def decide(analysis: dict) -> dict: - response = client.messages.create( - model="claude-sonnet-4-6", - max_tokens=750, - system=SYSTEM_PROMPT, - messages=[{ - "role": "user", - "content": ( - f"Analiza esta campaña y devuelve la decisión en JSON:\n\n" - f"{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: - # Fallback seguro si el modelo no devuelve JSON limpio - return { - "accion": "MANTENER", - "parametro": 1.0, - "nuevo_budget_diario": analysis.get("budget_diario_actual", 0), - "justificacion": "Error parseando respuesta del agente. Revisión manual recomendada.", - "alerta": f"JSON inválido recibido: {raw[:200]}", - "confianza": 0.0, - } +import json +from datetime import datetime +import anthropic +import config + +client = anthropic.Anthropic(api_key=config.ANTHROPIC_API_KEY) + +PORTFOLIO_SYSTEM = """ +Eres un experto en marketing de performance para una agencia de generación de leads en formación. +Recibes datos agregados del portfolio de campañas de Google Ads (solo campañas fco_). +Responde siempre en español, de forma concisa y accionable. Sin markdown, sin bullet symbols especiales, usa guiones simples (-). +""" + +WEEKLY_SYSTEM = """ +Eres un consultor senior de marketing de performance especializado en generación de leads para formación. +Recibes el análisis semanal del portfolio de campañas de Google Ads (solo campañas fco_). +Tu análisis debe ser estratégico, comparando la semana actual con la anterior, identificando tendencias y proponiendo acciones concretas. +Responde siempre en español. Sin markdown, sin bullet symbols especiales, usa guiones simples (-). +""" + +SYSTEM_PROMPT = """ +Eres un agente experto en optimización de campañas de generación de leads para centros de formación. +Cada campaña corresponde a un curso concreto con un PPL (precio por lead) fijo acordado con los centros compradores. + +MODELO DE NEGOCIO: +- Ingreso = leads_entregados × PPL +- Margen = (Ingreso - Gasto Google Ads) / Ingreso +- El objetivo es maximizar leads dentro del capping mensual manteniendo margen positivo. +- El CPA máximo ya refleja el margen mínimo aceptable. + +ESTADO DE LA CAMPAÑA: +- El campo status_google indica el estado actual en Google Ads: ENABLED (activa) o PAUSED (pausada). +- Nunca recomiendes reactivar una campaña si status_google = ENABLED (ya está activa). +- Si status_google = PAUSED y la acción es AUMENTAR_PRESUPUESTO, menciona en el consejo que primero hay que reactivarla. + +CAPPING: +- Si capping = 0, significa que no hay límite de leads configurado para este mes. No menciones el capping en el consejo ni en la justificación. Toma la decisión basándote únicamente en rentabilidad y ritmo. + +REGLAS DE DECISIÓN: +1. urgencia=PAUSAR → accion=PAUSAR siempre. El capping está lleno, seguir gastando destruye margen. +2. urgencia=SPRINT → accion=AUMENTAR_PRESUPUESTO con parametro entre 1.3 y 1.5. Quedan pocos días y leads por entregar. +3. urgencia=ACELERAR y campaña rentable → accion=AUMENTAR_PRESUPUESTO con parametro entre 1.1 y 1.25. +4. urgencia=ACELERAR y campaña NO rentable → accion=MANTENER o revisar keywords (no gastar más si no convierte). +5. urgencia=FRENAR → accion=REDUCIR_PRESUPUESTO con parametro entre 0.75 y 0.9. +6. urgencia=EN_RITMO y rentable → accion=MANTENER. +7. urgencia=EN_RITMO y NO rentable → accion=REDUCIR_PRESUPUESTO con parametro 0.85. +8. alerta_tracking=true → añadir alerta sobre discrepancia de tracking aunque la acción sea otra. + +Devuelve ÚNICAMENTE un JSON válido con esta estructura exacta, sin texto adicional ni markdown: +{ + "accion": "PAUSAR | REDUCIR_PRESUPUESTO | AUMENTAR_PRESUPUESTO | MANTENER", + "parametro": 1.0, + "nuevo_budget_diario": 0.0, + "justificacion": "explicación breve del porqué de la decisión", + "consejo": "acción concreta y específica que debería tomar el gestor (keywords, pujas, anuncios, configuración, etc.)", + "alerta": "texto si hay algo crítico, null si no hay", + "confianza": 0.0 +} + +El campo nuevo_budget_diario = budget_diario_actual × parametro (calcula tú el valor final). +El campo consejo debe ser accionable y específico: qué revisar, qué cambiar, qué hacer a continuación. +""" + + +def _classify_type(curso: str) -> str: + c = curso.lower() + if "_leadform" in c: + return "leadform" + if "_pmx" in c or "pmx_" in c: + return "pmx" + if "search" in c: + return "search" + return "otro" + + +def portfolio_daily_analysis(collected: list) -> str: + """Análisis estratégico diario del portfolio fco_. Devuelve texto plano para Slack.""" + from datetime import datetime + now = datetime.now() + fco = [i for i in collected if i["campaign"]["curso"].lower().startswith("fco_")] + + tipos: dict = {} + leadforms_detail = [] + alertas_tracking = 0 + campañas_perdida = 0 + + for item in fco: + t = _classify_type(item["campaign"]["curso"]) + m = item["metrics"] + a = item["analysis"] + cost = m.get("cost", 0) + conv = a["conversiones_google"] + ppl = item["campaign"]["ppl"] + rev = a["revenue_estimado"] + margen_pct = round((rev - cost) / rev * 100, 1) if rev > 0 else 0.0 + + if t not in tipos: + tipos[t] = {"campañas": 0, "inversion": 0.0, "conversiones": 0, "ingreso": 0.0} + tipos[t]["campañas"] += 1 + tipos[t]["inversion"] += cost + tipos[t]["conversiones"] += conv + tipos[t]["ingreso"] += rev + + if a.get("alerta_tracking"): + alertas_tracking += 1 + if rev > 0 and cost > rev: + campañas_perdida += 1 + + if t == "leadform": + leadforms_detail.append({ + "curso": item["campaign"]["curso"][:40], + "cpa_google": round(cost / conv, 2) if conv > 0 else None, + "conv_google": int(conv), + "conv_airtable": item["leads"], + "margen_pct": margen_pct, + }) + + resumen_tipos = {} + for t, d in tipos.items(): + cpa = round(d["inversion"] / d["conversiones"], 2) if d["conversiones"] > 0 else None + ing = d["ingreso"] + margen = round((ing - d["inversion"]) / ing * 100, 1) if ing > 0 else 0.0 + resumen_tipos[t] = { + "campañas": d["campañas"], + "inversion": round(d["inversion"], 2), + "conversiones": int(d["conversiones"]), + "cpa_medio": cpa, + "margen_pct": margen, + } + + data = { + "fecha": now.strftime("%d/%m/%Y"), + "dia_del_mes": now.day, + "campañas_totales": len(fco), + "campañas_en_perdida": campañas_perdida, + "alertas_tracking": alertas_tracking, + "rendimiento_por_tipo": resumen_tipos, + "detalle_leadforms": leadforms_detail, + } + + try: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=800, + system=PORTFOLIO_SYSTEM, + messages=[{ + "role": "user", + "content": ( + "Analiza estos datos del portfolio y proporciona:\n" + "1. Diagnóstico en 2 frases\n" + "2. Problemas principales (máx 3, con guión)\n" + "3. Acciones prioritarias (máx 3, muy concretas, con guión)\n" + "Si hay campañas leadform, evalúa específicamente su situación.\n\n" + f"{json.dumps(data, ensure_ascii=False, indent=2)}" + ), + }], + ) + return response.content[0].text.strip() + except Exception as e: + return f"Error generando análisis: {e}" + + +def weekly_strategic_analysis(games_md_this: dict, games_md_prev_week: dict, + collected: list, mes_nombre: str) -> str: + """ + Análisis estratégico semanal profundo. + games_md_this: MetricasDiarias de GAMes de los últimos 7 días (esta semana). + games_md_prev_week: MetricasDiarias de GAMes de los 7 días anteriores. + collected: lista de campañas del optimizer. + """ + def _week_summary(md: dict) -> dict: + coste = ing = leads = leads_lake = 0.0 + for v in md.values(): + coste += v.get("coste", 0) + ing += v.get("ingreso_sum", 0) + leads += v.get("leads", 0) + leads_lake += v.get("leads_lake", 0) + margen = round((ing - coste) / ing * 100, 1) if ing > 0 else 0.0 + cpa = round(coste / leads, 2) if leads > 0 else None + return {"coste": round(coste, 2), "ingreso": round(ing, 2), + "leads_google": int(leads), "leads_airtable": int(leads_lake), + "margen_pct": margen, "cpa": cpa} + + fco = [i for i in collected if i["campaign"]["curso"].lower().startswith("fco_")] + + # Top 5 peores por CPA del mes + peores = sorted( + [{"curso": i["campaign"]["curso"][:40], + "cpa": i["analysis"]["cpa_actual"], + "conv": int(i["analysis"]["conversiones_google"]), + "margen_pct": round(i["analysis"]["margen"] * 100, 1)} + for i in fco if i["analysis"]["cpa_actual"] > 0], + key=lambda x: x["cpa"], reverse=True + )[:5] + + data = { + "mes": mes_nombre, + "semana_actual": _week_summary(games_md_this), + "semana_anterior": _week_summary(games_md_prev_week), + "top5_peor_cpa_mes": peores, + "leadforms_activos": sum(1 for i in fco if "_leadform" in i["campaign"]["curso"].lower()), + "campañas_totales": len(fco), + } + + try: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=900, + system=WEEKLY_SYSTEM, + messages=[{ + "role": "user", + "content": ( + "Genera el informe estratégico semanal con:\n" + "1. Resumen ejecutivo (3 frases comparando esta semana con la anterior)\n" + "2. Tendencias clave detectadas (máx 4, con guión)\n" + "3. Situación campañas leadform y qué hacer con ellas\n" + "4. Acciones estratégicas prioritarias para la próxima semana (máx 4, muy concretas, con guión)\n" + "5. Una frase de conclusión sobre si el portfolio va en la dirección correcta\n\n" + f"{json.dumps(data, ensure_ascii=False, indent=2)}" + ), + }], + ) + return response.content[0].text.strip() + except Exception as e: + return f"Error generando análisis semanal: {e}" + + +def decide(analysis: dict) -> dict: + response = client.messages.create( + model="claude-sonnet-4-6", + max_tokens=750, + system=SYSTEM_PROMPT, + messages=[{ + "role": "user", + "content": ( + f"Analiza esta campaña y devuelve la decisión en JSON:\n\n" + f"{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: + # Fallback seguro si el modelo no devuelve JSON limpio + return { + "accion": "MANTENER", + "parametro": 1.0, + "nuevo_budget_diario": analysis.get("budget_diario_actual", 0), + "justificacion": "Error parseando respuesta del agente. Revisión manual recomendada.", + "alerta": f"JSON inválido recibido: {raw[:200]}", + "confianza": 0.0, + } diff --git a/analyzer.py b/analyzer.py index 3e5d720..ac0a13a 100644 --- a/analyzer.py +++ b/analyzer.py @@ -1,69 +1,69 @@ -from datetime import datetime -import calendar - - -def analyze(campaign_config: dict, leads_entregados: int, ads_metrics: dict) -> dict: - now = datetime.now() - dias_mes = calendar.monthrange(now.year, now.month)[1] - dia_actual = now.day - ratio_mes = dia_actual / dias_mes - - capping = campaign_config["capping_mensual"] - ppl = campaign_config["ppl"] - cpa_max = campaign_config["cpa_maximo"] - margen_objetivo = campaign_config.get("margen_objetivo", 0) - gasto = ads_metrics.get("cost", 0) - conversiones_google = ads_metrics.get("conversions", 0) - - ratio_leads = leads_entregados / capping if capping > 0 else 0 - cpa_actual = gasto / leads_entregados if leads_entregados > 0 else 0 - revenue = leads_entregados * ppl - margen = (revenue - gasto) / revenue if revenue > 0 else 0 - leads_restantes = capping - leads_entregados - dias_restantes = dias_mes - dia_actual - ritmo = ratio_leads - ratio_mes # positivo = adelantado, negativo = atrasado - - # Urgencia - if ratio_leads >= 1.0: - urgencia = "PAUSAR" - elif capping > 0 and ratio_leads < ratio_mes - 0.15 and dias_restantes <= 5: - urgencia = "SPRINT" - elif ritmo < -0.15: - urgencia = "ACELERAR" - elif ritmo > 0.15: - urgencia = "FRENAR" - else: - urgencia = "EN_RITMO" - - # Discrepancia en ambas direcciones: Google > Airtable puede indicar conversiones falsas - conv_leads_lake_mes = campaign_config.get("conv_leads_lake_mes", leads_entregados) - discrepancia = abs(conversiones_google - conv_leads_lake_mes) - - return { - "curso": campaign_config["curso"], - "campaign_id": campaign_config["google_campaign_id"], - "ppl": ppl, - "cpa_maximo": cpa_max, - "margen_objetivo": margen_objetivo, - "margen_ok": margen >= margen_objetivo if margen_objetivo > 0 else True, - "capping": capping, - "leads_entregados": leads_entregados, - "leads_restantes": leads_restantes, - "dias_restantes": dias_restantes, - "ratio_leads": round(ratio_leads, 3), - "ratio_mes": round(ratio_mes, 3), - "ritmo": round(ritmo, 3), - "urgencia": urgencia, - "cpa_actual": round(cpa_actual, 2), - "rentable": cpa_actual <= cpa_max if cpa_actual > 0 else True, - "margen": round(margen, 3), - "revenue_estimado": round(revenue, 2), - "gasto_acumulado": round(gasto, 2), - "budget_diario_actual": ads_metrics.get("budget_daily", 0), - "ctr": ads_metrics.get("ctr", 0), - "clicks": ads_metrics.get("clicks", 0), - "conversiones_google": conversiones_google, - "discrepancia_tracking": discrepancia, - "alerta_tracking": discrepancia > 10, - "status_google": ads_metrics.get("status", "UNKNOWN"), - } +from datetime import datetime +import calendar + + +def analyze(campaign_config: dict, leads_entregados: int, ads_metrics: dict) -> dict: + now = datetime.now() + dias_mes = calendar.monthrange(now.year, now.month)[1] + dia_actual = now.day + ratio_mes = dia_actual / dias_mes + + capping = campaign_config["capping_mensual"] + ppl = campaign_config["ppl"] + cpa_max = campaign_config["cpa_maximo"] + margen_objetivo = campaign_config.get("margen_objetivo", 0) + gasto = ads_metrics.get("cost", 0) + conversiones_google = ads_metrics.get("conversions", 0) + + ratio_leads = leads_entregados / capping if capping > 0 else 0 + cpa_actual = gasto / leads_entregados if leads_entregados > 0 else 0 + revenue = leads_entregados * ppl + margen = (revenue - gasto) / revenue if revenue > 0 else 0 + leads_restantes = capping - leads_entregados + dias_restantes = dias_mes - dia_actual + ritmo = ratio_leads - ratio_mes # positivo = adelantado, negativo = atrasado + + # Urgencia + if ratio_leads >= 1.0: + urgencia = "PAUSAR" + elif capping > 0 and ratio_leads < ratio_mes - 0.15 and dias_restantes <= 5: + urgencia = "SPRINT" + elif ritmo < -0.15: + urgencia = "ACELERAR" + elif ritmo > 0.15: + urgencia = "FRENAR" + else: + urgencia = "EN_RITMO" + + # Discrepancia en ambas direcciones: Google > Airtable puede indicar conversiones falsas + conv_leads_lake_mes = campaign_config.get("conv_leads_lake_mes", leads_entregados) + discrepancia = abs(conversiones_google - conv_leads_lake_mes) + + return { + "curso": campaign_config["curso"], + "campaign_id": campaign_config["google_campaign_id"], + "ppl": ppl, + "cpa_maximo": cpa_max, + "margen_objetivo": margen_objetivo, + "margen_ok": margen >= margen_objetivo if margen_objetivo > 0 else True, + "capping": capping, + "leads_entregados": leads_entregados, + "leads_restantes": leads_restantes, + "dias_restantes": dias_restantes, + "ratio_leads": round(ratio_leads, 3), + "ratio_mes": round(ratio_mes, 3), + "ritmo": round(ritmo, 3), + "urgencia": urgencia, + "cpa_actual": round(cpa_actual, 2), + "rentable": cpa_actual <= cpa_max if cpa_actual > 0 else True, + "margen": round(margen, 3), + "revenue_estimado": round(revenue, 2), + "gasto_acumulado": round(gasto, 2), + "budget_diario_actual": ads_metrics.get("budget_daily", 0), + "ctr": ads_metrics.get("ctr", 0), + "clicks": ads_metrics.get("clicks", 0), + "conversiones_google": conversiones_google, + "discrepancia_tracking": discrepancia, + "alerta_tracking": discrepancia > 10, + "status_google": ads_metrics.get("status", "UNKNOWN"), + } diff --git a/config.py b/config.py index 810c0b0..769129a 100644 --- a/config.py +++ b/config.py @@ -1,26 +1,26 @@ -import os -from dotenv import load_dotenv - -load_dotenv() - -# Airtable -AIRTABLE_TOKEN = os.environ["AIRTABLE_TOKEN"] -AIRTABLE_BASE_ID = os.environ["AIRTABLE_BASE_ID"] -LEADS_TABLE = "Leads Lake" -CAMPAIGNS_TABLE = "Google Ads Campaigns" - -# Google Ads -GOOGLE_ADS_DEVELOPER_TOKEN = os.environ["GOOGLE_ADS_DEVELOPER_TOKEN"] -GOOGLE_ADS_CLIENT_ID = os.environ["GOOGLE_ADS_CLIENT_ID"] -GOOGLE_ADS_CLIENT_SECRET = os.environ["GOOGLE_ADS_CLIENT_SECRET"] -GOOGLE_ADS_REFRESH_TOKEN = os.environ["GOOGLE_ADS_REFRESH_TOKEN"] -GOOGLE_ADS_LOGIN_CUSTOMER_ID = os.environ["GOOGLE_ADS_LOGIN_CUSTOMER_ID"] - -# Anthropic -ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"] - -# Slack -SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "") - -# Operación -DRY_RUN = True # True = solo sugiere, no aplica cambios en Google Ads +import os +from dotenv import load_dotenv + +load_dotenv() + +# Airtable +AIRTABLE_TOKEN = os.environ["AIRTABLE_TOKEN"] +AIRTABLE_BASE_ID = os.environ["AIRTABLE_BASE_ID"] +LEADS_TABLE = "Leads Lake" +CAMPAIGNS_TABLE = "Google Ads Campaigns" + +# Google Ads +GOOGLE_ADS_DEVELOPER_TOKEN = os.environ["GOOGLE_ADS_DEVELOPER_TOKEN"] +GOOGLE_ADS_CLIENT_ID = os.environ["GOOGLE_ADS_CLIENT_ID"] +GOOGLE_ADS_CLIENT_SECRET = os.environ["GOOGLE_ADS_CLIENT_SECRET"] +GOOGLE_ADS_REFRESH_TOKEN = os.environ["GOOGLE_ADS_REFRESH_TOKEN"] +GOOGLE_ADS_LOGIN_CUSTOMER_ID = os.environ["GOOGLE_ADS_LOGIN_CUSTOMER_ID"] + +# Anthropic +ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"] + +# Slack +SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "") + +# Operación +DRY_RUN = True # True = solo sugiere, no aplica cambios en Google Ads diff --git a/google_ads_client.py b/google_ads_client.py index 0c817f3..98475c2 100644 --- a/google_ads_client.py +++ b/google_ads_client.py @@ -1,365 +1,365 @@ -from google.ads.googleads.client import GoogleAdsClient as GAdsClient -from google.ads.googleads.errors import GoogleAdsException -from datetime import datetime -import config - - -class GoogleAdsClient: - def __init__(self): - self.client = GAdsClient.load_from_dict({ - "developer_token": config.GOOGLE_ADS_DEVELOPER_TOKEN, - "client_id": config.GOOGLE_ADS_CLIENT_ID, - "client_secret": config.GOOGLE_ADS_CLIENT_SECRET, - "refresh_token": config.GOOGLE_ADS_REFRESH_TOKEN, - "login_customer_id": config.GOOGLE_ADS_LOGIN_CUSTOMER_ID, - "use_proto_plus": True, - }) - self.customer_id = config.GOOGLE_ADS_LOGIN_CUSTOMER_ID - - def get_all_campaigns(self) -> list[dict]: - """Obtiene todas las campañas no eliminadas de la cuenta.""" - ga_service = self.client.get_service("GoogleAdsService") - query = """ - SELECT - campaign.id, - campaign.name, - campaign.status - FROM campaign - WHERE campaign.status != 'REMOVED' - ORDER BY campaign.name - """ - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - return [ - { - "id": str(row.campaign.id), - "name": row.campaign.name, - "status": row.campaign.status.name, - } - for row in response - ] - except GoogleAdsException as e: - print(f" ❌ Error obteniendo campañas de Google Ads: {e}") - return [] - - def get_monthly_metrics_all(self) -> dict: - """ - Devuelve métricas del mes actual para TODAS las campañas en una sola query. - Retorna dict {campaign_id: {conversions, cost}}. - """ - ga_service = self.client.get_service("GoogleAdsService") - query = """ - SELECT - campaign.id, - metrics.conversions, - metrics.cost_micros - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date DURING THIS_MONTH - """ - result = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - cid = str(row.campaign.id) - result[cid] = { - "conversions": row.metrics.conversions, - "cost": row.metrics.cost_micros / 1_000_000, - } - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas mensuales: {e}") - return result - - def get_yesterday_metrics_all(self) -> dict: - """ - Devuelve métricas del día anterior para TODAS las campañas en una sola query. - Retorna dict {campaign_id: {conversions, cost}}. - """ - from datetime import timedelta - yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") - ga_service = self.client.get_service("GoogleAdsService") - query = f""" - SELECT - campaign.id, - metrics.conversions, - metrics.cost_micros - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date = '{yesterday}' - """ - result = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - cid = str(row.campaign.id) - result[cid] = { - "conversions": row.metrics.conversions, - "cost": row.metrics.cost_micros / 1_000_000, - } - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas de hoy: {e}") - return result - - def get_metrics_for_date(self, date_str: str) -> dict: - """ - Devuelve métricas de una fecha concreta ('YYYY-MM-DD') para TODAS las - campañas en una sola query. Retorna dict {campaign_id: {conversions, cost}}. - """ - ga_service = self.client.get_service("GoogleAdsService") - query = f""" - SELECT - campaign.id, - metrics.conversions, - metrics.cost_micros - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date = '{date_str}' - """ - result = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - cid = str(row.campaign.id) - result[cid] = { - "conversions": row.metrics.conversions, - "cost": row.metrics.cost_micros / 1_000_000, - } - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas del {date_str}: {e}") - return result - - def get_daily_metrics_for_month(self, year: int, month: int) -> dict: - """ - Devuelve coste y conversiones diarias de todas las campañas para un mes - completo. Retorna dict {date_str ('YYYY-MM-DD'): {campaign_id: {cost, conversions}}}. - """ - start = f"{year}-{month:02d}-01" - if month == 12: - end = f"{year + 1}-01-01" - else: - end = f"{year}-{month + 1:02d}-01" - ga_service = self.client.get_service("GoogleAdsService") - query = f""" - SELECT - campaign.id, - segments.date, - metrics.conversions, - metrics.cost_micros - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date >= '{start}' - AND segments.date < '{end}' - """ - result: dict = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - date_str = row.segments.date - cid = str(row.campaign.id) - result.setdefault(date_str, {})[cid] = { - "conversions": row.metrics.conversions, - "cost": row.metrics.cost_micros / 1_000_000, - } - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas diarias del mes: {e}") - return result - - def get_daily_conversions_for_month(self, year: int, month: int) -> dict: - """ - Devuelve conversiones diarias de todas las campañas para un mes completo. - Retorna dict {date_str ('YYYY-MM-DD'): {campaign_id: conversions}}. - """ - start = f"{year}-{month:02d}-01" - # último día del mes - if month == 12: - end = f"{year + 1}-01-01" - else: - end = f"{year}-{month + 1:02d}-01" - ga_service = self.client.get_service("GoogleAdsService") - query = f""" - SELECT - campaign.id, - segments.date, - metrics.conversions - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date >= '{start}' - AND segments.date < '{end}' - """ - result: dict = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - date_str = row.segments.date - cid = str(row.campaign.id) - result.setdefault(date_str, {})[cid] = row.metrics.conversions - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas mensuales: {e}") - return result - - def get_monthly_metrics_all(self) -> dict: - """ - Métricas del mes en curso para TODAS las campañas en una sola query. - Retorna dict {campaign_id: {cost, conversions, clicks, impressions, ctr, - status, budget_daily, budget_resource_name, name}}. - """ - ga_service = self.client.get_service("GoogleAdsService") - query = """ - SELECT - campaign.id, - campaign.name, - campaign.status, - campaign_budget.amount_micros, - campaign_budget.resource_name, - metrics.cost_micros, - metrics.conversions, - metrics.clicks, - metrics.impressions - FROM campaign - WHERE campaign.status != 'REMOVED' - AND segments.date DURING THIS_MONTH - """ - raw: dict = {} - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - for row in response: - cid = str(row.campaign.id) - if cid not in raw: - raw[cid] = { - "name": row.campaign.name, - "status": row.campaign.status.name, - "budget_daily": row.campaign_budget.amount_micros / 1_000_000, - "budget_resource_name": row.campaign_budget.resource_name, - "cost": 0.0, "conversions": 0.0, "clicks": 0, "impressions": 0, - } - m = row.metrics - raw[cid]["cost"] += m.cost_micros / 1_000_000 - raw[cid]["conversions"] += m.conversions - raw[cid]["clicks"] += m.clicks - raw[cid]["impressions"] += m.impressions - except GoogleAdsException as e: - print(f" ❌ Error obteniendo métricas mensuales bulk: {e}") - - result = {} - for cid, d in raw.items(): - imp = d["impressions"] - result[cid] = { - "campaign_id": cid, - "name": d["name"], - "status": d["status"], - "budget_daily": round(d["budget_daily"], 2), - "budget_resource_name": d["budget_resource_name"], - "cost": round(d["cost"], 2), - "conversions": d["conversions"], - "clicks": d["clicks"], - "impressions": imp, - "ctr": round(d["clicks"] / imp * 100, 2) if imp > 0 else 0.0, - } - return result - - def get_campaign_metrics(self, campaign_id: str) -> dict: - """Métricas del mes en curso para una campaña concreta (acumulado mensual).""" - ga_service = self.client.get_service("GoogleAdsService") - query = f""" - SELECT - campaign.id, - campaign.name, - campaign.status, - campaign_budget.amount_micros, - campaign_budget.resource_name, - metrics.cost_micros, - metrics.conversions, - metrics.clicks, - metrics.impressions - FROM campaign - WHERE campaign.id = {campaign_id} - AND segments.date DURING THIS_MONTH - """ - try: - response = ga_service.search(customer_id=self.customer_id, query=query) - cost = conversions = clicks = impressions = 0 - meta = {} - for row in response: - m = row.metrics - cost += m.cost_micros / 1_000_000 - conversions += m.conversions - clicks += m.clicks - impressions += m.impressions - if not meta: - meta = { - "name": row.campaign.name, - "status": row.campaign.status.name, - "budget_daily": row.campaign_budget.amount_micros / 1_000_000, - "budget_resource_name": row.campaign_budget.resource_name, - } - if not meta: - return {} - ctr = round(clicks / impressions * 100, 2) if impressions > 0 else 0.0 - return { - "campaign_id": campaign_id, - "name": meta["name"], - "status": meta["status"], - "budget_daily": meta["budget_daily"], - "budget_resource_name": meta["budget_resource_name"], - "cost": round(cost, 2), - "conversions": conversions, - "clicks": clicks, - "impressions": impressions, - "ctr": ctr, - } - except GoogleAdsException as e: - print(f" ❌ Error Google Ads para campaña {campaign_id}: {e}") - return {} - return {} - - def set_campaign_budget(self, budget_resource_name: str, new_daily_budget: float): - """Ajusta el presupuesto diario de una campaña.""" - if config.DRY_RUN: - print(f" [DRY RUN] Nuevo presupuesto diario → {new_daily_budget:.2f}€") - return True - - try: - budget_service = self.client.get_service("CampaignBudgetService") - campaign_budget = self.client.get_type("CampaignBudget") - campaign_budget.resource_name = budget_resource_name - campaign_budget.amount_micros = int(new_daily_budget * 1_000_000) - - operation = self.client.get_type("CampaignBudgetOperation") - operation.update = campaign_budget - operation.update_mask.paths.append("amount_micros") - - budget_service.mutate_campaign_budgets( - customer_id=self.customer_id, - operations=[operation] - ) - return True - except GoogleAdsException as e: - print(f" ❌ Error ajustando presupuesto: {e}") - return False - - def pause_campaign(self, campaign_id: str): - """Pausa una campaña.""" - if config.DRY_RUN: - print(f" [DRY RUN] Pausar campaña {campaign_id}") - return True - - try: - campaign_service = self.client.get_service("CampaignService") - campaign = self.client.get_type("Campaign") - campaign.resource_name = campaign_service.campaign_path( - self.customer_id, campaign_id - ) - campaign.status = self.client.enums.CampaignStatusEnum.PAUSED - - operation = self.client.get_type("CampaignOperation") - operation.update = campaign - operation.update_mask.paths.append("status") - - campaign_service.mutate_campaigns( - customer_id=self.customer_id, - operations=[operation] - ) - return True - except GoogleAdsException as e: - print(f" ❌ Error pausando campaña: {e}") - return False +from google.ads.googleads.client import GoogleAdsClient as GAdsClient +from google.ads.googleads.errors import GoogleAdsException +from datetime import datetime +import config + + +class GoogleAdsClient: + def __init__(self): + self.client = GAdsClient.load_from_dict({ + "developer_token": config.GOOGLE_ADS_DEVELOPER_TOKEN, + "client_id": config.GOOGLE_ADS_CLIENT_ID, + "client_secret": config.GOOGLE_ADS_CLIENT_SECRET, + "refresh_token": config.GOOGLE_ADS_REFRESH_TOKEN, + "login_customer_id": config.GOOGLE_ADS_LOGIN_CUSTOMER_ID, + "use_proto_plus": True, + }) + self.customer_id = config.GOOGLE_ADS_LOGIN_CUSTOMER_ID + + def get_all_campaigns(self) -> list[dict]: + """Obtiene todas las campañas no eliminadas de la cuenta.""" + ga_service = self.client.get_service("GoogleAdsService") + query = """ + SELECT + campaign.id, + campaign.name, + campaign.status + FROM campaign + WHERE campaign.status != 'REMOVED' + ORDER BY campaign.name + """ + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + return [ + { + "id": str(row.campaign.id), + "name": row.campaign.name, + "status": row.campaign.status.name, + } + for row in response + ] + except GoogleAdsException as e: + print(f" ❌ Error obteniendo campañas de Google Ads: {e}") + return [] + + def get_monthly_metrics_all(self) -> dict: + """ + Devuelve métricas del mes actual para TODAS las campañas en una sola query. + Retorna dict {campaign_id: {conversions, cost}}. + """ + ga_service = self.client.get_service("GoogleAdsService") + query = """ + SELECT + campaign.id, + metrics.conversions, + metrics.cost_micros + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date DURING THIS_MONTH + """ + result = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + cid = str(row.campaign.id) + result[cid] = { + "conversions": row.metrics.conversions, + "cost": row.metrics.cost_micros / 1_000_000, + } + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas mensuales: {e}") + return result + + def get_yesterday_metrics_all(self) -> dict: + """ + Devuelve métricas del día anterior para TODAS las campañas en una sola query. + Retorna dict {campaign_id: {conversions, cost}}. + """ + from datetime import timedelta + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + ga_service = self.client.get_service("GoogleAdsService") + query = f""" + SELECT + campaign.id, + metrics.conversions, + metrics.cost_micros + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date = '{yesterday}' + """ + result = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + cid = str(row.campaign.id) + result[cid] = { + "conversions": row.metrics.conversions, + "cost": row.metrics.cost_micros / 1_000_000, + } + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas de hoy: {e}") + return result + + def get_metrics_for_date(self, date_str: str) -> dict: + """ + Devuelve métricas de una fecha concreta ('YYYY-MM-DD') para TODAS las + campañas en una sola query. Retorna dict {campaign_id: {conversions, cost}}. + """ + ga_service = self.client.get_service("GoogleAdsService") + query = f""" + SELECT + campaign.id, + metrics.conversions, + metrics.cost_micros + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date = '{date_str}' + """ + result = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + cid = str(row.campaign.id) + result[cid] = { + "conversions": row.metrics.conversions, + "cost": row.metrics.cost_micros / 1_000_000, + } + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas del {date_str}: {e}") + return result + + def get_daily_metrics_for_month(self, year: int, month: int) -> dict: + """ + Devuelve coste y conversiones diarias de todas las campañas para un mes + completo. Retorna dict {date_str ('YYYY-MM-DD'): {campaign_id: {cost, conversions}}}. + """ + start = f"{year}-{month:02d}-01" + if month == 12: + end = f"{year + 1}-01-01" + else: + end = f"{year}-{month + 1:02d}-01" + ga_service = self.client.get_service("GoogleAdsService") + query = f""" + SELECT + campaign.id, + segments.date, + metrics.conversions, + metrics.cost_micros + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date >= '{start}' + AND segments.date < '{end}' + """ + result: dict = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + date_str = row.segments.date + cid = str(row.campaign.id) + result.setdefault(date_str, {})[cid] = { + "conversions": row.metrics.conversions, + "cost": row.metrics.cost_micros / 1_000_000, + } + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas diarias del mes: {e}") + return result + + def get_daily_conversions_for_month(self, year: int, month: int) -> dict: + """ + Devuelve conversiones diarias de todas las campañas para un mes completo. + Retorna dict {date_str ('YYYY-MM-DD'): {campaign_id: conversions}}. + """ + start = f"{year}-{month:02d}-01" + # último día del mes + if month == 12: + end = f"{year + 1}-01-01" + else: + end = f"{year}-{month + 1:02d}-01" + ga_service = self.client.get_service("GoogleAdsService") + query = f""" + SELECT + campaign.id, + segments.date, + metrics.conversions + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date >= '{start}' + AND segments.date < '{end}' + """ + result: dict = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + date_str = row.segments.date + cid = str(row.campaign.id) + result.setdefault(date_str, {})[cid] = row.metrics.conversions + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas mensuales: {e}") + return result + + def get_monthly_metrics_all(self) -> dict: + """ + Métricas del mes en curso para TODAS las campañas en una sola query. + Retorna dict {campaign_id: {cost, conversions, clicks, impressions, ctr, + status, budget_daily, budget_resource_name, name}}. + """ + ga_service = self.client.get_service("GoogleAdsService") + query = """ + SELECT + campaign.id, + campaign.name, + campaign.status, + campaign_budget.amount_micros, + campaign_budget.resource_name, + metrics.cost_micros, + metrics.conversions, + metrics.clicks, + metrics.impressions + FROM campaign + WHERE campaign.status != 'REMOVED' + AND segments.date DURING THIS_MONTH + """ + raw: dict = {} + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + for row in response: + cid = str(row.campaign.id) + if cid not in raw: + raw[cid] = { + "name": row.campaign.name, + "status": row.campaign.status.name, + "budget_daily": row.campaign_budget.amount_micros / 1_000_000, + "budget_resource_name": row.campaign_budget.resource_name, + "cost": 0.0, "conversions": 0.0, "clicks": 0, "impressions": 0, + } + m = row.metrics + raw[cid]["cost"] += m.cost_micros / 1_000_000 + raw[cid]["conversions"] += m.conversions + raw[cid]["clicks"] += m.clicks + raw[cid]["impressions"] += m.impressions + except GoogleAdsException as e: + print(f" ❌ Error obteniendo métricas mensuales bulk: {e}") + + result = {} + for cid, d in raw.items(): + imp = d["impressions"] + result[cid] = { + "campaign_id": cid, + "name": d["name"], + "status": d["status"], + "budget_daily": round(d["budget_daily"], 2), + "budget_resource_name": d["budget_resource_name"], + "cost": round(d["cost"], 2), + "conversions": d["conversions"], + "clicks": d["clicks"], + "impressions": imp, + "ctr": round(d["clicks"] / imp * 100, 2) if imp > 0 else 0.0, + } + return result + + def get_campaign_metrics(self, campaign_id: str) -> dict: + """Métricas del mes en curso para una campaña concreta (acumulado mensual).""" + ga_service = self.client.get_service("GoogleAdsService") + query = f""" + SELECT + campaign.id, + campaign.name, + campaign.status, + campaign_budget.amount_micros, + campaign_budget.resource_name, + metrics.cost_micros, + metrics.conversions, + metrics.clicks, + metrics.impressions + FROM campaign + WHERE campaign.id = {campaign_id} + AND segments.date DURING THIS_MONTH + """ + try: + response = ga_service.search(customer_id=self.customer_id, query=query) + cost = conversions = clicks = impressions = 0 + meta = {} + for row in response: + m = row.metrics + cost += m.cost_micros / 1_000_000 + conversions += m.conversions + clicks += m.clicks + impressions += m.impressions + if not meta: + meta = { + "name": row.campaign.name, + "status": row.campaign.status.name, + "budget_daily": row.campaign_budget.amount_micros / 1_000_000, + "budget_resource_name": row.campaign_budget.resource_name, + } + if not meta: + return {} + ctr = round(clicks / impressions * 100, 2) if impressions > 0 else 0.0 + return { + "campaign_id": campaign_id, + "name": meta["name"], + "status": meta["status"], + "budget_daily": meta["budget_daily"], + "budget_resource_name": meta["budget_resource_name"], + "cost": round(cost, 2), + "conversions": conversions, + "clicks": clicks, + "impressions": impressions, + "ctr": ctr, + } + except GoogleAdsException as e: + print(f" ❌ Error Google Ads para campaña {campaign_id}: {e}") + return {} + return {} + + def set_campaign_budget(self, budget_resource_name: str, new_daily_budget: float): + """Ajusta el presupuesto diario de una campaña.""" + if config.DRY_RUN: + print(f" [DRY RUN] Nuevo presupuesto diario → {new_daily_budget:.2f}€") + return True + + try: + budget_service = self.client.get_service("CampaignBudgetService") + campaign_budget = self.client.get_type("CampaignBudget") + campaign_budget.resource_name = budget_resource_name + campaign_budget.amount_micros = int(new_daily_budget * 1_000_000) + + operation = self.client.get_type("CampaignBudgetOperation") + operation.update = campaign_budget + operation.update_mask.paths.append("amount_micros") + + budget_service.mutate_campaign_budgets( + customer_id=self.customer_id, + operations=[operation] + ) + return True + except GoogleAdsException as e: + print(f" ❌ Error ajustando presupuesto: {e}") + return False + + def pause_campaign(self, campaign_id: str): + """Pausa una campaña.""" + if config.DRY_RUN: + print(f" [DRY RUN] Pausar campaña {campaign_id}") + return True + + try: + campaign_service = self.client.get_service("CampaignService") + campaign = self.client.get_type("Campaign") + campaign.resource_name = campaign_service.campaign_path( + self.customer_id, campaign_id + ) + campaign.status = self.client.enums.CampaignStatusEnum.PAUSED + + operation = self.client.get_type("CampaignOperation") + operation.update = campaign + operation.update_mask.paths.append("status") + + campaign_service.mutate_campaigns( + customer_id=self.customer_id, + operations=[operation] + ) + return True + except GoogleAdsException as e: + print(f" ❌ Error pausando campaña: {e}") + return False