José Manuel Gómez
b30a075c59
- Add get_monthly_metrics_all() bulk method to GoogleAdsClient - Use bulk fetch in first pass instead of N individual API calls - Pre-compute leadform conversions per course number - For PMX campaigns with leadform companion, leads_grupo = PMX conv + leadform conv - fco_pmx_60 now correctly shows 334 leads at 61% margin (was negative) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
462 lines
18 KiB
Python
462 lines
18 KiB
Python
import sys
|
||
import io
|
||
import os
|
||
import re
|
||
import json
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True)
|
||
|
||
from airtable_client import AirtableClient
|
||
from google_ads_client import GoogleAdsClient
|
||
from analyzer import analyze
|
||
from agent import decide, portfolio_daily_analysis
|
||
from optimizer import apply_decision
|
||
from slack_reporter import build_and_send
|
||
import config
|
||
from datetime import datetime, timedelta
|
||
|
||
|
||
class Tee:
|
||
"""Escribe simultáneamente en consola y en archivo de log."""
|
||
def __init__(self, filepath):
|
||
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
||
self._file = open(filepath, "w", encoding="utf-8")
|
||
self._stdout = sys.stdout
|
||
|
||
def write(self, data):
|
||
self._stdout.write(data)
|
||
self._file.write(data)
|
||
|
||
def flush(self):
|
||
self._stdout.flush()
|
||
if not self._file.closed:
|
||
self._file.flush()
|
||
|
||
def close(self):
|
||
self._file.close()
|
||
|
||
|
||
ICONOS = {
|
||
"PAUSAR": "⛔",
|
||
"SPRINT": "🚀",
|
||
"ACELERAR": "📈",
|
||
"FRENAR": "📉",
|
||
"EN_RITMO": "✅",
|
||
}
|
||
|
||
ACCION_ICONOS = {
|
||
"PAUSAR": "⛔",
|
||
"AUMENTAR_PRESUPUESTO": "📈",
|
||
"REDUCIR_PRESUPUESTO": "📉",
|
||
"MANTENER": "✅",
|
||
}
|
||
|
||
# Google Ads customer ID (sin guiones) para construir enlaces directos
|
||
_CUSTOMER_ID = config.GOOGLE_ADS_LOGIN_CUSTOMER_ID.replace("-", "")
|
||
|
||
|
||
def _ads_link(campaign_id: str) -> str:
|
||
return f"https://ads.google.com/aw/campaigns?campaignId={campaign_id}&__c={_CUSTOMER_ID}"
|
||
|
||
|
||
def _priority(item: dict) -> int:
|
||
"""
|
||
0 — Crítica: urgencia PAUSAR o SPRINT
|
||
1 — No crítica: accion cambia presupuesto pero urgencia no es crítica
|
||
2 — Mantener: accion MANTENER
|
||
"""
|
||
urgencia = item["analysis"]["urgencia"]
|
||
accion = item["decision"]["accion"]
|
||
if urgencia in ("PAUSAR", "SPRINT"):
|
||
return 0
|
||
if accion != "MANTENER":
|
||
return 1
|
||
return 2
|
||
|
||
|
||
SECCION_LABELS = {
|
||
0: "ACCIONES CRÍTICAS (PAUSAR / AUMENTAR)",
|
||
1: "ACCIONES NO CRÍTICAS",
|
||
2: "SIN CAMBIOS (MANTENER)",
|
||
}
|
||
|
||
CRITICIDAD_MAP = {
|
||
0: "Crítico",
|
||
1: "Peligro",
|
||
2: "Mantener",
|
||
}
|
||
|
||
|
||
def run():
|
||
print(f"\n{'='*55}")
|
||
print(f" LEADS OPTIMIZER — {datetime.now().strftime('%d/%m/%Y %H:%M')}")
|
||
print(f" Modo: {'DRY RUN (sin cambios)' if config.DRY_RUN else '⚡ PRODUCCIÓN'}")
|
||
print(f"{'='*55}\n")
|
||
|
||
at = AirtableClient()
|
||
gads = GoogleAdsClient()
|
||
|
||
# Sincronizar catálogo de campañas desde Google Ads → Airtable
|
||
print("→ Sincronizando campañas desde Google Ads...")
|
||
google_campaigns = gads.get_all_campaigns()
|
||
monthly_metrics = gads.get_monthly_metrics_all()
|
||
today_metrics = gads.get_yesterday_metrics_all()
|
||
_ayer_date = (datetime.now() - timedelta(days=1))
|
||
leads_yesterday = at.get_leads_by_campaign_on_date(_ayer_date.strftime("%Y-%m-%d"))
|
||
print(" Calculando PPL y CapTotalMes...")
|
||
ppl_lookup, cap_lookup = at.build_campaign_lookups()
|
||
sync_result = at.sync_campaigns_from_google_ads(google_campaigns, monthly_metrics, ppl_lookup, cap_lookup)
|
||
if sync_result["created"]:
|
||
print(f" ✅ Campañas nuevas importadas ({len(sync_result['created'])}):")
|
||
for c in sync_result["created"]:
|
||
print(f" + [{c['id']}] {c['name']} → {c['status']}")
|
||
if sync_result["updated"]:
|
||
print(f" 🔄 Campañas actualizadas ({len(sync_result['updated'])}):")
|
||
for c in sync_result["updated"]:
|
||
for field, val in c["changes"].items():
|
||
print(f" ~ [{c['id']}] {c['name']} | {field}: '{val['antes']}' → '{val['ahora']}'")
|
||
if not sync_result["created"] and not sync_result["updated"]:
|
||
print(" ✓ Sin cambios en el catálogo.")
|
||
|
||
# Sincronizar GACampaignMes (campañas con actividad este mes)
|
||
print(" Sincronizando GACampaignMes...")
|
||
gcm_result = at.sync_gacampaignmes(
|
||
google_campaigns, monthly_metrics, ppl_lookup, cap_lookup,
|
||
sync_result["at_by_gid"]
|
||
)
|
||
print(f" ✓ GACampaignMes: {gcm_result['created']} nuevas, {gcm_result['updated']} actualizadas.")
|
||
print()
|
||
|
||
campaigns = at.get_active_gacampaignmes()
|
||
print(f"→ {len(campaigns)} campañas con actividad este mes")
|
||
print("→ Analizando...\n")
|
||
|
||
# Detección anticipada de cursos con Search Y PMX simultáneos
|
||
def _course_num(name: str) -> str | None:
|
||
m = re.search(r'fco_(?:search|pmx)_(\d+)', name, re.IGNORECASE)
|
||
return m.group(1) if m else None
|
||
|
||
course_types_pre: dict[str, set] = {}
|
||
for c in campaigns:
|
||
num = _course_num(c["curso"])
|
||
if num:
|
||
course_types_pre.setdefault(num, set())
|
||
if "pmx" in c["curso"].lower():
|
||
course_types_pre[num].add("pmx")
|
||
elif "search" in c["curso"].lower():
|
||
course_types_pre[num].add("search")
|
||
courses_with_both = {
|
||
num for num, types in course_types_pre.items()
|
||
if "pmx" in types and "search" in types
|
||
}
|
||
|
||
# Detección anticipada de cursos con campaña _leadform activa
|
||
courses_with_leadform = {
|
||
_course_num(c["curso"])
|
||
for c in campaigns
|
||
if "_leadform" in c["curso"].lower() and _course_num(c["curso"])
|
||
}
|
||
|
||
# Precomputar conversiones Google de cada campaña leadform por número de curso
|
||
leadform_conv_by_course: dict[str, int] = {}
|
||
for c in campaigns:
|
||
if "_leadform" in c["curso"].lower():
|
||
num = _course_num(c["curso"])
|
||
if num:
|
||
lf_conv = int(monthly_metrics.get(c["google_campaign_id"], {}).get("conversions", 0))
|
||
leadform_conv_by_course[num] = leadform_conv_by_course.get(num, 0) + lf_conv
|
||
|
||
# === PRIMERA PASADA: recopilar datos de todas las campañas ===
|
||
collected = []
|
||
skipped = []
|
||
status_updates = [] # (gcm_id, campaign_at_id, google_status) para batch update
|
||
|
||
for campaign in campaigns:
|
||
cid = campaign["google_campaign_id"]
|
||
|
||
leads, lead_ids = at.get_leads_this_month_gads(cid, campaign["curso"])
|
||
at.update_gacampaignmes_leads_lake(campaign["airtable_id"], lead_ids)
|
||
|
||
metrics = monthly_metrics.get(cid)
|
||
if not metrics:
|
||
skipped.append(f"[{cid}] {campaign['curso']}")
|
||
continue
|
||
|
||
# Recopilar status para actualizar ambas tablas al final del pase
|
||
google_status = metrics.get("status", "")
|
||
if google_status:
|
||
status_updates.append((
|
||
campaign["airtable_id"],
|
||
campaign.get("campaign_at_id", ""),
|
||
google_status,
|
||
))
|
||
|
||
course_num = _course_num(campaign["curso"])
|
||
is_pmx = "pmx" in campaign["curso"].lower() and "_leadform" not in campaign["curso"].lower()
|
||
|
||
# PMX con companion Search: conversiones Google del PMX (search fluye hacia PMX)
|
||
is_pmx_with_companion = is_pmx and course_num in courses_with_both
|
||
# PMX con companion Leadform: conv PMX + conv leadform (leads no llegan a Airtable)
|
||
is_pmx_with_leadform = is_pmx and course_num in courses_with_leadform
|
||
|
||
if is_pmx_with_leadform:
|
||
leads_grupo = int(metrics.get("conversions", 0)) + leadform_conv_by_course.get(course_num, 0)
|
||
elif is_pmx_with_companion:
|
||
leads_grupo = int(metrics.get("conversions", 0))
|
||
else:
|
||
leads_grupo = leads
|
||
|
||
analysis = analyze(campaign, leads_grupo, metrics)
|
||
decision = decide(analysis)
|
||
|
||
collected.append({
|
||
"campaign": campaign,
|
||
"leads": leads,
|
||
"leads_grupo": leads_grupo,
|
||
"metrics": metrics,
|
||
"analysis": analysis,
|
||
"decision": decision,
|
||
"today_metrics": today_metrics.get(cid, {}),
|
||
"is_pmx_with_leadform": is_pmx_with_leadform,
|
||
})
|
||
|
||
# Actualizar status en GACampaignMes y Google Ads Campaigns
|
||
if status_updates:
|
||
at.batch_update_status(status_updates)
|
||
|
||
if skipped:
|
||
print(f" ⚠️ {len(skipped)} campañas sin métricas omitidas:")
|
||
for s in skipped:
|
||
print(f" · {s}")
|
||
print()
|
||
|
||
# Ordenar: 0=críticas → 1=no críticas → 2=mantener
|
||
collected.sort(key=_priority)
|
||
|
||
# === SEGUNDA PASADA: imprimir en orden + aplicar decisiones ===
|
||
resumen = []
|
||
advice_updates = [] # (gcm_record_id, consejo, criticidad) para batch update final
|
||
metricas_updates = [] # {airtable_id, metricas_json} para MetricasDiarias
|
||
games_agg: dict = {} # dia_hoy → {coste, ingreso_sum, ingreso_lxp, leads, leads_lake}
|
||
games_mes = {"coste_mes": 0.0, "conv_mes": 0, "ppl_leads": 0.0, "leads_total": 0}
|
||
ayer = datetime.now() - timedelta(days=1)
|
||
dia_hoy = ayer.strftime("%d")
|
||
cambio_mes = ayer.month != datetime.now().month
|
||
last_priority = -1
|
||
|
||
for item in collected:
|
||
campaign = item["campaign"]
|
||
leads = item["leads"]
|
||
metrics = item["metrics"]
|
||
analysis = item["analysis"]
|
||
decision = item["decision"]
|
||
today_m = item["today_metrics"]
|
||
cid = campaign["google_campaign_id"]
|
||
|
||
p = _priority(item)
|
||
if p != last_priority:
|
||
print(f"\n{'━'*55}")
|
||
print(f" {SECCION_LABELS[p]}")
|
||
print(f"{'━'*55}")
|
||
last_priority = p
|
||
|
||
icono = ICONOS.get(analysis["urgencia"], "❓")
|
||
accion_icono = ACCION_ICONOS.get(decision["accion"], "")
|
||
|
||
print(f"{'─'*55}")
|
||
print(f"📚 {campaign['curso']}")
|
||
print(f" ID: {cid} | PPL: {campaign['ppl']}€ | Cap: {campaign['capping_mensual']} leads")
|
||
print(f" 🔗 {_ads_link(cid)}")
|
||
print(f" Leads mes: {leads}/{campaign['capping_mensual']} "
|
||
f"({analysis['ratio_leads']*100:.0f}% cap) | "
|
||
f"Ratio mes: {analysis['ratio_mes']*100:.0f}%")
|
||
print(f" CPA actual: {analysis['cpa_actual']}€ | "
|
||
f"CPA máximo: {analysis['cpa_maximo']}€ | "
|
||
f"Margen: {analysis['margen']*100:.0f}%")
|
||
print(f" Urgencia: {icono} {analysis['urgencia']} | "
|
||
f"Rentable: {'✅' if analysis['rentable'] else '❌'}")
|
||
|
||
if analysis["alerta_tracking"]:
|
||
print(f" 🚨 ALERTA TRACKING: {analysis['discrepancia_tracking']} leads de diferencia "
|
||
f"entre Airtable ({campaign['conv_leads_lake_mes']}) y Google Ads ({int(analysis['conversiones_google'])})")
|
||
|
||
print(f" Decisión: {accion_icono} {decision['accion']} "
|
||
f"(confianza: {decision['confianza']*100:.0f}%)")
|
||
print(f" Justificación: {decision['justificacion']}")
|
||
|
||
if decision.get("consejo"):
|
||
print(f" 💡 Consejo: {decision['consejo']}")
|
||
|
||
if decision.get("alerta"):
|
||
print(f" 🚨 {decision['alerta']}")
|
||
|
||
apply_decision(campaign, decision, metrics, gads)
|
||
|
||
conv_google = int(analysis["conversiones_google"])
|
||
discrepancia_fresh = abs(conv_google - leads)
|
||
if discrepancia_fresh > 10:
|
||
log_text = (
|
||
f"⚠️ DISCREPANCIA: {leads} leads Airtable vs {conv_google} conv Google Ads "
|
||
f"(Δ {discrepancia_fresh})"
|
||
)
|
||
else:
|
||
log_text = f"OK: {leads} leads Airtable | {conv_google} conv Google Ads"
|
||
|
||
# Nota de reatribución PMX para campañas Search con companion PMX del mismo curso
|
||
course_num = _course_num(campaign["curso"])
|
||
if "search" in campaign["curso"].lower() and course_num in courses_with_both:
|
||
log_text += " | ⚠️ PMX ATTRIBUTION: campaña Search con companion PMX activo — parte de las conversiones de Google pueden estar reatribuidas a la campaña PMX"
|
||
|
||
# Nota leadform: los leads se capturan en Google sin pasar por la web
|
||
is_leadform = "_leadform" in campaign["curso"].lower()
|
||
if is_leadform:
|
||
log_text += " | ℹ️ LEADFORM: leads capturados directamente en Google (sin visitar la web) — no llegan a Airtable"
|
||
elif item.get("is_pmx_with_leadform"):
|
||
lf_conv = leadform_conv_by_course.get(course_num, 0)
|
||
log_text += f" | ℹ️ LEADFORM COMPANION: {lf_conv} conv leadform sumadas a leads_grupo para este curso"
|
||
|
||
# Métricas diarias: coste hoy, ingreso (conversiones × PPL) y margen
|
||
coste_hoy = round(today_m.get("cost", 0), 2)
|
||
conv_hoy = today_m.get("conversions", 0)
|
||
ingreso_hoy = round(conv_hoy * campaign["ppl"], 2)
|
||
margen_hoy = round(ingreso_hoy - coste_hoy, 2)
|
||
try:
|
||
md = json.loads(campaign["metricas_diarias"])
|
||
except (json.JSONDecodeError, TypeError):
|
||
md = {}
|
||
leads_lake_hoy = leads_yesterday.get(cid, 0)
|
||
md[dia_hoy] = {"coste": coste_hoy, "ingreso": ingreso_hoy, "margen": margen_hoy, "leads": int(conv_hoy), "leads_lake": leads_lake_hoy}
|
||
campaign["metricas_diarias"] = json.dumps(md, ensure_ascii=False)
|
||
|
||
if campaign["curso"].lower().startswith("fco_"):
|
||
if dia_hoy not in games_agg:
|
||
games_agg[dia_hoy] = {"coste": 0.0, "ingreso_sum": 0.0, "ingreso_lxp": 0.0, "leads": 0, "leads_lake": 0}
|
||
games_agg[dia_hoy]["coste"] += coste_hoy
|
||
games_agg[dia_hoy]["ingreso_sum"] += ingreso_hoy
|
||
games_agg[dia_hoy]["ingreso_lxp"] += leads_lake_hoy * campaign["ppl"]
|
||
games_agg[dia_hoy]["leads"] += int(conv_hoy)
|
||
games_agg[dia_hoy]["leads_lake"] += leads_lake_hoy
|
||
# Totales mes (fuente: Google Ads acumulado mensual)
|
||
games_mes["coste_mes"] += metrics.get("cost", 0)
|
||
games_mes["conv_mes"] += int(analysis["conversiones_google"])
|
||
# PPLMedio ponderado por leads Airtable
|
||
games_mes["ppl_leads"] += campaign["ppl"] * leads
|
||
games_mes["leads_total"] += leads
|
||
metricas_updates.append({
|
||
"airtable_id": campaign["airtable_id"],
|
||
"metricas_json": json.dumps(md, ensure_ascii=False),
|
||
})
|
||
|
||
advice_updates.append((
|
||
campaign["airtable_id"],
|
||
decision.get("consejo", ""),
|
||
CRITICIDAD_MAP[p],
|
||
log_text,
|
||
))
|
||
|
||
resumen.append({
|
||
"curso": campaign["curso"],
|
||
"urgencia": analysis["urgencia"],
|
||
"accion": decision["accion"],
|
||
"leads": f"{leads}/{campaign['capping_mensual']}",
|
||
"cpa": analysis["cpa_actual"],
|
||
"margen": f"{analysis['margen']*100:.0f}%",
|
||
"consejo": decision.get("consejo", ""),
|
||
"link": _ads_link(cid),
|
||
"prioridad": p,
|
||
})
|
||
|
||
print()
|
||
|
||
# Guardar consejos y criticidad en GACampaignMes
|
||
if advice_updates:
|
||
print(f"→ Guardando consejos y criticidad en GACampaignMes ({len(advice_updates)} registros)...")
|
||
at.batch_update_gacampaignmes_advice(advice_updates)
|
||
print(" ✓ Consejos y criticidad guardados.")
|
||
|
||
# Guardar métricas diarias en MetricasDiarias
|
||
if metricas_updates:
|
||
print(f"→ Actualizando MetricasDiarias ({len(metricas_updates)} registros)...")
|
||
if cambio_mes:
|
||
# Ayer pertenece al mes anterior: redirigir escritura al GACampaignMes correcto
|
||
prev_map = at.get_gcm_id_map_for_month(ayer.month, ayer.year)
|
||
for u in metricas_updates:
|
||
gid = next(
|
||
(item["campaign"]["google_campaign_id"]
|
||
for item in collected if item["campaign"]["airtable_id"] == u["airtable_id"]),
|
||
None,
|
||
)
|
||
if gid and gid in prev_map:
|
||
u["airtable_id"] = prev_map[gid]
|
||
at.batch_update_metricas_diarias(metricas_updates)
|
||
print(" ✓ MetricasDiarias actualizado.")
|
||
|
||
# Actualizar GAMes con el agregado diario fco_
|
||
if games_agg:
|
||
print("→ Actualizando GAMes...")
|
||
games_rid = at.get_or_create_games_record(ayer.month, ayer.year)
|
||
games_md = at.get_games_metricas(games_rid)
|
||
for d, vals in games_agg.items():
|
||
games_md[d] = {k: round(v, 2) if isinstance(v, float) else v for k, v in vals.items()}
|
||
coste_mes = round(games_mes["coste_mes"], 2)
|
||
conv_mes = games_mes["conv_mes"]
|
||
ppl_medio = round(games_mes["ppl_leads"] / games_mes["leads_total"], 2) if games_mes["leads_total"] > 0 else 0.0
|
||
cpa_medio = round(coste_mes / conv_mes, 2) if conv_mes > 0 else 0.0
|
||
totales = {
|
||
"CosteMes": coste_mes,
|
||
"ConvMes": conv_mes,
|
||
"PPLMedio": ppl_medio,
|
||
"CPAMedio": cpa_medio,
|
||
}
|
||
at.update_games_metricas(games_rid, json.dumps(games_md, ensure_ascii=False), totales)
|
||
print(" ✓ GAMes actualizado.")
|
||
|
||
# Snapshot diario: ConvLeadsLakeMesFinal + ConvLeadsLakeMesGrupo
|
||
# PMX con companion Search → Grupo = conversiones Google (ya calculado en leads_grupo)
|
||
final_leads_data = [
|
||
{
|
||
"airtable_id": item["campaign"]["airtable_id"],
|
||
"conv_leads_lake_mes": item["leads"],
|
||
"conv_leads_lake_mes_grupo": item["leads_grupo"],
|
||
}
|
||
for item in collected
|
||
]
|
||
if final_leads_data:
|
||
print(f"→ Actualizando ConvLeadsLakeMesFinal ({len(final_leads_data)} registros)...")
|
||
at.batch_update_gacampaignmes_final_leads(final_leads_data)
|
||
print(" ✓ ConvLeadsLakeMesFinal actualizado.")
|
||
|
||
# Resumen final ordenado por prioridad (ya está ordenado)
|
||
print(f"{'='*55}")
|
||
print("RESUMEN FINAL")
|
||
print(f"{'='*55}")
|
||
last_p = -1
|
||
for r in resumen:
|
||
if r["prioridad"] != last_p:
|
||
print(f"\n --- {SECCION_LABELS[r['prioridad']]} ---")
|
||
last_p = r["prioridad"]
|
||
print(f" {r['curso'][:35]:<35} | {r['urgencia']:<12} | {r['accion']:<25} | {r['leads']} leads | {r['margen']} margen")
|
||
print(f" {'':35} {r['link']}")
|
||
if r["consejo"]:
|
||
print(f" {'':35} 💡 {r['consejo']}")
|
||
print()
|
||
|
||
# Enviar resumen a Slack
|
||
print("→ Generando análisis estratégico del portfolio...")
|
||
portfolio_text = portfolio_daily_analysis(collected)
|
||
print("→ Enviando resumen a Slack...")
|
||
prev_month_metricas = at.get_metricas_diarias_prev_month() if (cambio_mes or datetime.now().day <= 5) else {}
|
||
build_and_send(collected, config.DRY_RUN, prev_month_metricas, portfolio_text)
|
||
print(" ✓ Resumen enviado a Slack.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
log_path = os.path.join("logs", f"{timestamp}.log")
|
||
tee = Tee(log_path)
|
||
sys.stdout = tee
|
||
try:
|
||
run()
|
||
finally:
|
||
tee.close()
|
||
print(f"\nLog guardado en: {log_path}", file=tee._stdout)
|