José Manuel Gómez
9edd74b9ad
- Detect Search+PMX companion campaigns by course number - Add PMX attribution warning to Log field for Search campaigns with active PMX - Add ConvLeadsLakeMesGrupo: for PMX with Search companion uses Search leads count Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
340 lines
12 KiB
Python
340 lines
12 KiB
Python
import sys
|
|
import io
|
|
import os
|
|
import re
|
|
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
|
|
from optimizer import apply_decision
|
|
import config
|
|
from datetime import datetime
|
|
|
|
|
|
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()
|
|
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")
|
|
|
|
# === 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)
|
|
at.update_gacampaignmes_leads_lake(campaign["airtable_id"], lead_ids)
|
|
|
|
metrics = gads.get_campaign_metrics(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,
|
|
))
|
|
|
|
analysis = analyze(campaign, leads, metrics)
|
|
decision = decide(analysis)
|
|
|
|
collected.append({
|
|
"campaign": campaign,
|
|
"leads": leads,
|
|
"metrics": metrics,
|
|
"analysis": analysis,
|
|
"decision": decision,
|
|
})
|
|
|
|
# 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)
|
|
|
|
# Detectar cursos con campaña Search Y PMX simultáneas (posible reatribución PMX)
|
|
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: dict[str, set] = {}
|
|
for item in collected:
|
|
name = item["campaign"]["curso"]
|
|
num = _course_num(name)
|
|
if num:
|
|
course_types.setdefault(num, set())
|
|
if "pmx" in name.lower():
|
|
course_types[num].add("pmx")
|
|
elif "search" in name.lower():
|
|
course_types[num].add("search")
|
|
courses_with_both = {num for num, types in course_types.items() if "pmx" in types and "search" in types}
|
|
|
|
# === SEGUNDA PASADA: imprimir en orden + aplicar decisiones ===
|
|
resumen = []
|
|
advice_updates = [] # (gcm_record_id, consejo, criticidad) para batch update final
|
|
last_priority = -1
|
|
|
|
for item in collected:
|
|
campaign = item["campaign"]
|
|
leads = item["leads"]
|
|
metrics = item["metrics"]
|
|
analysis = item["analysis"]
|
|
decision = item["decision"]
|
|
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"
|
|
|
|
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.")
|
|
|
|
# Snapshot diario: ConvLeadsLakeMesFinal + ConvLeadsLakeMesGrupo
|
|
# Para PMX con Search companion del mismo curso, Grupo = leads del Search
|
|
course_search_leads = {
|
|
_course_num(item["campaign"]["curso"]): item["leads"]
|
|
for item in collected
|
|
if "search" in item["campaign"]["curso"].lower()
|
|
and _course_num(item["campaign"]["curso"])
|
|
}
|
|
final_leads_data = []
|
|
for item in collected:
|
|
name = item["campaign"]["curso"]
|
|
num = _course_num(name)
|
|
if num and "pmx" in name.lower() and num in courses_with_both:
|
|
grupo = course_search_leads.get(num, item["leads"])
|
|
else:
|
|
grupo = item["leads"]
|
|
final_leads_data.append({
|
|
"airtable_id": item["campaign"]["airtable_id"],
|
|
"conv_leads_lake_mes": item["leads"],
|
|
"conv_leads_lake_mes_grupo": grupo,
|
|
})
|
|
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()
|
|
|
|
|
|
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)
|