import sys import io import os 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)", } 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 = [] 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 analysis = analyze(campaign, leads, metrics) decision = decide(analysis) collected.append({ "campaign": campaign, "leads": leads, "metrics": metrics, "analysis": analysis, "decision": decision, }) 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 = [] consejo_updates = [] # (gcm_record_id, consejo) 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 ({leads}) 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) if decision.get("consejo"): consejo_updates.append((campaign["airtable_id"], decision["consejo"])) 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 en GACampaignMes if consejo_updates: print(f"→ Guardando {len(consejo_updates)} consejos en GACampaignMes...") at.batch_update_gacampaignmes_consejos(consejo_updates) print(" ✓ Consejos guardados.") # 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)