"""Meta Optimizer Formación — main entry point.""" import sys import io import os import time from datetime import datetime, timedelta import config from meta_ads_client import MetaAdsClient from airtable_client import AirtableClient, extract_cursoid from agent import decide, analyze_unit from baserow_client import BaserowClient import analyzer import slack_notifier _ACTION_MAP = { "PAUSE": "PAUSE", "REDUCE_BUDGET": "REDUCE_BUDGET", "INCREASE_BUDGET": "INCREASE_BUDGET", "MAINTAIN": "MAINTAIN", # legacy Spanish names, por si el modelo responde en español "PAUSAR": "PAUSE", "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", "MANTENER": "MAINTAIN", } def _criticidad(urgencia: str, action_type: str) -> str: if urgencia in ("PAUSAR", "SPRINT"): return "Crítico" if action_type != "MAINTAIN": return "Peligro" return "Mantener" def _priority(urgencia: str, action_type: str) -> int: if urgencia in ("PAUSAR", "SPRINT"): return 0 if action_type != "MAINTAIN": return 1 return 2 class Tee: def __init__(self, filepath: str): 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() def _execute_action(meta: MetaAdsClient, action: dict): """Apply an approved action via Meta API.""" action_type = action.get("action_type", "") cid = action.get("campaign_id", "") parameter = float(action.get("parameter") or 1.0) if action_type == "PAUSE": if cid.startswith("ad:"): meta.pause_ad(cid[3:]) else: meta.pause_campaign(cid) elif action_type in ("REDUCE_BUDGET", "INCREASE_BUDGET"): try: bid_cfg = meta.get_campaign_bid_config(cid) current_budget = bid_cfg.get("daily_budget_eur") if current_budget: meta.set_campaign_budget(cid, int(current_budget * parameter * 100)) except Exception: pass def run(): start_ts = time.time() now = datetime.now() print(f"\n{'='*55}") print(f" META OPTIMIZER FORMACIÓN — {now.strftime('%d/%m/%Y %H:%M')}") print(f" Prefix: {config.META_CAMPAIGN_PREFIX} | Modelo: PPL + capping mensual por curso") print(f" Mode: {'DRY RUN (no changes)' if config.DRY_RUN else 'PRODUCTION'}") print(f"{'='*55}\n") meta = MetaAdsClient() baserow = BaserowClient() airtable = AirtableClient() # ── Lookups de negocio (PPL, capping, familia) desde Airtable ────────────── print("→ Cargando PPL/capping/familia por curso desde Airtable...") ppl_lookup, cap_lookup, familia_lookup = airtable.build_campaign_lookups() print(f" ✓ {len(ppl_lookup)} cursos con PPL, {len(cap_lookup)} con capping este mes.\n") # ── Execute previously approved actions ─────────────────────────────────── actions_executed = 0 if not config.DRY_RUN: approved = baserow.get_approved_actions() print(f"→ Executing {len(approved)} approved actions...\n") for action in approved: try: _execute_action(meta, action) baserow.update_action_status(action["id"], "executed") actions_executed += 1 print(f" ✓ {action.get('campaign_name')} — {action.get('action_type')}") except Exception as e: print(f" ✗ Error on action {action['id']}: {e}") # ── Catálogo Meta -> Airtable (Meta Ads Campaigns) ───────────────────────── print(f"→ Sincronizando catálogo de campañas {config.META_CAMPAIGN_PREFIX} con Airtable...") meta_campaigns = meta.get_all_campaigns() sync_result = airtable.sync_campaigns_from_meta_ads(meta_campaigns, ppl_lookup) at_by_cid = sync_result["at_by_cid"] print(f" ✓ {len(sync_result['created'])} creadas, {len(sync_result['updated'])} actualizadas " f"(de {len(meta_campaigns)} campañas totales).\n") # ── Métricas mes-a-la-fecha (para MetaCampaignMes y para el análisis) ────── print("→ Fetching month-to-date metrics...") month_start = f"{now.year}-{now.month:02d}-01" yesterday = (now - timedelta(days=1)).strftime("%Y-%m-%d") monthly_metrics_meta = meta.get_campaign_metrics(month_start, yesterday) print(f" ✓ {len(monthly_metrics_meta)} campañas con gasto este mes.\n") mcm_sync = airtable.sync_metacampaignmes( meta_campaigns, monthly_metrics_meta, ppl_lookup, cap_lookup, at_by_cid, ) print(f"→ MetaCampaignMes: {mcm_sync['created']} creadas, {mcm_sync['updated']} actualizadas.\n") mcm_by_meta_cid = {r["meta_campaign_id"]: r for r in airtable.get_active_metacampaignmes()} # ── Monthly daily totals (per-campaign rows → agregado por familia) ──────── print(f"→ Fetching monthly daily totals for {config.META_CAMPAIGN_PREFIX}...") daily_rows = meta.get_daily_campaign_rows(month_start, yesterday) _daily: dict = {} monthly_familias: dict = {} for row in daily_rows: cursoid = extract_cursoid(row["campaign_name"]) or "" familia = familia_lookup.get(cursoid, "Sin familia") ppl = ppl_lookup.get(cursoid, 0) margin = round(row["leads"] * ppl - row["spend"], 2) d = _daily.setdefault(row["date"], {"spend": 0.0, "leads": 0, "margin": 0.0, "f_margins": {}}) d["spend"] += row["spend"] d["leads"] += row["leads"] d["margin"] += margin d["f_margins"][familia] = d["f_margins"].get(familia, 0.0) + margin mf = monthly_familias.setdefault(familia, {"spend": 0.0, "leads": 0, "margin": 0.0}) mf["spend"] += row["spend"] mf["leads"] += row["leads"] mf["margin"] += margin daily_totals = [ { "date": date, "spend": round(d["spend"], 2), "leads": int(d["leads"]), "cpl": round(d["spend"] / d["leads"], 2) if d["leads"] > 0 else 0.0, "margin": round(d["margin"], 2), "f_margins": {f: round(m, 0) for f, m in d["f_margins"].items()}, } for date, d in sorted(_daily.items()) ] print(f" ✓ {len(daily_totals)} days with data.\n") # ── Yesterday metrics (contexto 1d para el informe) ──────────────────────── print(f"→ Fetching yesterday metrics ({config.META_CAMPAIGN_PREFIX} only, spend > 0)...") metrics_yesterday = meta.get_yesterday_metrics() print(f" ✓ {len(metrics_yesterday)} campaigns active yesterday.\n") # ── 3-day and 7-day metrics (capa táctica adset/anuncio) ─────────────────── print("→ Fetching 3-day and 7-day metrics...") metrics_3d = meta.get_period_campaign_metrics(days=3) metrics_7d = meta.get_period_campaign_metrics(days=7) print(" ✓ Multi-window data ready.\n") # ── Analyze active campaigns & propose actions ───────────────────────────── active_campaigns = [mc for mc in meta_campaigns if mc["status"] == "ACTIVE"] actions_proposed_list = [] campaign_details = {} # {cid: {familia, margin, adsets, ads, ...}} familias = {} # {familia: {spend, leads, margin}} advice_updates = [] # [(mcm_id, consejo, criticidad, log)] final_leads_updates = [] # [(mcm_id, leads_entregados)] errors = [] for mc in active_campaigns: cid, name = mc["id"], mc["name"] cursoid = extract_cursoid(name) or "" familia = familia_lookup.get(cursoid, "Sin familia") ppl = ppl_lookup.get(cursoid, 0) cap = cap_lookup.get(cursoid, 0) cpa_max = round(ppl * 0.70, 2) leads_entregados, _ = airtable.get_leads_this_month_meta(name) m1 = metrics_yesterday.get(cid, {}) mmes = monthly_metrics_meta.get(cid, {}) campaign_bid = {} try: campaign_bid = meta.get_campaign_bid_config(cid) except Exception as e: errors.append(f"Bid config {name}: {e}") ads_metrics = { "spend": mmes.get("spend", 0.0), "leads": mmes.get("leads", 0), "ctr": m1.get("ctr", 0.0), "clicks": m1.get("clicks", 0), "budget_daily": campaign_bid.get("daily_budget_eur", 0) or 0, "status": mc["status"], } campaign_config = { "curso": name, "meta_campaign_id": cid, "ppl": ppl, "cpa_maximo": cpa_max, "capping_mensual": cap, "conv_leads_lake_mes": leads_entregados, } analysis = analyzer.analyze(campaign_config, leads_entregados, ads_metrics) try: decision = decide(analysis) except Exception as e: errors.append(f"{name}: {e}") continue action_type = _ACTION_MAP.get(decision.get("action", "MAINTAIN"), "MAINTAIN") adset_bids = {} try: adset_bids = meta.get_adset_bid_configs(cid) except Exception as e: errors.append(f"Adset bids {name}: {e}") # ABO campaigns (presupuesto solo a nivel de conjunto): omitir ajustes de campaña is_cbo = campaign_bid.get("daily_budget_eur") is not None if action_type in ("INCREASE_BUDGET", "REDUCE_BUDGET") and not is_cbo: action_type = "MAINTAIN" print(f" {name[:52]}") print(f" Curso: {cursoid} Familia: {familia} PPL: {ppl}€ CPAmax: {cpa_max}€") print(f" Urgencia: {analysis['urgencia']} Ritmo: {analysis['ritmo']:+.2f} " f"Leads mes: {leads_entregados}/{cap or '∞'} Margen: {analysis['margen']*100:.0f}%") print(f" Decision: {action_type} — {(decision.get('justification') or '')[:70]}") if decision.get("alert"): print(f" ALERT: {decision['alert']}") print() if action_type != "MAINTAIN": try: row = baserow.save_action({ "campaign_id": cid, "campaign_name": name, "action_type": action_type, "parameter": decision.get("parameter") or 1.0, "justification": decision.get("justification") or "", "advice": decision.get("advice") or "", "alert": decision.get("alert") or "", "confidence": decision.get("confidence") or 0.0, }) actions_proposed_list.append({ "campaign_name": name, "action_type": action_type, "parameter": decision.get("parameter") or 1.0, "justification": decision.get("justification") or "", "advice": decision.get("advice") or "", "alert": decision.get("alert") or "", "confidence": decision.get("confidence") or 0.0, "cpa_actual": analysis["cpa_actual"], "cpa_maximo": cpa_max, "row_id": row["id"], }) except Exception as e: errors.append(f"Save action {name}: {e}") # ── Ad set analysis (3d) ──────────────────────────────────────────── adsets_detail = [] try: for as_m in meta.get_period_adset_metrics(cid, days=3)[:5]: bid = adset_bids.get(as_m["id"], {}) as_m["bid_strategy"] = bid.get("bid_strategy", "") as_m["cost_cap_eur"] = bid.get("cost_cap_eur") result = analyze_unit(as_m, "adset") adsets_detail.append({**as_m, **result}) print(f" [Adset] {as_m['name'][:45]} — {result.get('evaluacion','')[:60]}") except Exception as e: errors.append(f"Adsets {name}: {e}") # ── Ad analysis (3d + 7d merged) ──────────────────────────────────── ads_detail = [] try: ads_3d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, days=3)} ads_7d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, days=7)} ordered_ids = list(dict.fromkeys( [a["id"] for a in sorted(ads_7d.values(), key=lambda x: -x["spend"])] + [a["id"] for a in sorted(ads_3d.values(), key=lambda x: -x["spend"])] ))[:5] for ad_id in ordered_ids: a3 = ads_3d.get(ad_id, {}) a7 = ads_7d.get(ad_id, {}) ad_m = dict(a7) if a7 else dict(a3) ad_m["cpl_3d"] = a3.get("cpl", 0.0) ad_m["leads_3d"] = a3.get("leads", 0) ad_m["spend_3d"] = a3.get("spend", 0.0) ad_m["cpl_7d"] = a7.get("cpl", 0.0) ad_m["leads_7d"] = a7.get("leads", 0) ad_m["ppl"] = ppl ad_m["cpa_maximo"] = cpa_max result = analyze_unit(ad_m, "ad") ad_entry = {**ad_m, **result} if result.get("accion") == "PAUSE": try: ad_row = baserow.save_action({ "campaign_id": f"ad:{ad_m['id']}", "campaign_name": ad_m["name"], "action_type": "PAUSE", "parameter": 1.0, "justification": result.get("recomendacion", ""), "advice": result.get("evaluacion", ""), "confidence": 0.8, }) ad_entry["row_id"] = ad_row["id"] except Exception as e: errors.append(f"Ad action {ad_m['name']}: {e}") ads_detail.append(ad_entry) action_tag = " ⛔PAUSE" if result.get("accion") == "PAUSE" else "" print(f" [Ad] {ad_m['name'][:45]} — {result.get('evaluacion','')[:60]}{action_tag}") except Exception as e: errors.append(f"Ads {name}: {e}") # margin_eur: proxy diario de rentabilidad (leads*PPL - gasto), igual unidad # que las tablas de Slack; margen_pct: rentabilidad acumulada del mes (analyzer). margin_eur = round(m1.get("leads", 0) * ppl - m1.get("spend", 0.0), 2) campaign_details[cid] = { "name": name, "familia": familia, "urgencia": analysis["urgencia"], "margen_pct": analysis["margen"], "margin": margin_eur, "leads_mes": leads_entregados, "capping": cap, "ppl": ppl, "spend_1d": m1.get("spend", 0.0), "leads_1d": m1.get("leads", 0), "adsets": adsets_detail, "ads": ads_detail, "bid_config": campaign_bid, } # ── Daily snapshot (persists analysis to Baserow for dashboard) ─────── try: baserow.save_daily_snapshot({ "run_date": now.strftime("%Y-%m-%d"), "campaign_id": cid, "campaign_name": name, "familia": familia, "spend": m1.get("spend", 0.0), "leads": m1.get("leads", 0), "cpl": m1.get("cpl", 0.0), "margin": margin_eur, "action_type": action_type, "justification": decision.get("justification") or "", "adsets": adsets_detail, "ads": ads_detail, }) except Exception as e: errors.append(f"Snapshot {name}: {e}") # ── Familia aggregation ────────────────────────────────────────────── f = familias.setdefault(familia, {"spend": 0.0, "leads": 0, "margin": 0.0}) f["spend"] += m1.get("spend", 0.0) f["leads"] += m1.get("leads", 0) f["margin"] += margin_eur # ── MetaCampaignMes: consejo/criticidad/log + leads confirmados ─────── mcm = mcm_by_meta_cid.get(cid) if mcm: criticidad = _criticidad(analysis["urgencia"], action_type) log_text = decision.get("alert") or "" advice_updates.append((mcm["airtable_id"], decision.get("advice") or "", criticidad, log_text)) final_leads_updates.append((mcm["airtable_id"], leads_entregados)) if advice_updates: airtable.batch_update_metacampaignmes_advice(advice_updates) if final_leads_updates: airtable.batch_update_metacampaignmes_final_leads(final_leads_updates) # ── Top 10 best and worst (por CPL de ayer) ───────────────────────────────── with_leads = [m for m in metrics_yesterday.values() if m["leads"] > 0] best_10 = sorted(with_leads, key=lambda x: x["cpl"])[:10] all_active = list(metrics_yesterday.values()) worst_10 = sorted( all_active, key=lambda x: (x["leads"] > 0, -x["cpl"] if x["leads"] > 0 else 0), )[:10] # ── Send consolidated Slack report ──────────────────────────────────────── duration = round(time.time() - start_ts, 1) try: slack_notifier.send_daily_report( daily_totals=daily_totals, best_campaigns=best_10, worst_campaigns=worst_10, actions=actions_proposed_list, campaigns_analyzed=len(active_campaigns), mode="DRY_RUN" if config.DRY_RUN else "PRODUCTION", familias=familias, campaign_details=campaign_details, monthly_familias=monthly_familias, ) except Exception as e: print(f" Warning: Slack notification failed: {e}") # ── Execution log ───────────────────────────────────────────────────────── summary = ( f"{len(active_campaigns)} campaigns analyzed, " f"{len(actions_proposed_list)} actions proposed, " f"{actions_executed} executed." ) try: baserow.save_execution_log({ "mode": "DRY_RUN" if config.DRY_RUN else "PRODUCTION", "campaigns_analyzed": len(active_campaigns), "actions_proposed": len(actions_proposed_list), "actions_executed": actions_executed, "errors": "\n".join(errors), "summary": summary, "duration_seconds": duration, }) except Exception as e: print(f" Warning: could not save execution log: {e}") print(f"{'='*55}") print(f" Done in {duration}s. {summary}") if errors: print(f" Errors ({len(errors)}): {'; '.join(errors[:3])}") print(f"{'='*55}\n") if __name__ == "__main__": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) 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() sys.stdout = tee._stdout