"""Meta Optimizer — main entry point.""" import sys import io import os import time from datetime import datetime import config from meta_ads_client import MetaAdsClient from agent import decide, analyze_creative, analyze_unit from baserow_client import BaserowClient import slack_notifier _ACTION_MAP = { "PAUSE": "PAUSE", "REDUCE_BUDGET": "REDUCE_BUDGET", "INCREASE_BUDGET": "INCREASE_BUDGET", "MAINTAIN": "MAINTAIN", "REVIEW_CREATIVES": "REVIEW_CREATIVES", # legacy Spanish names "PAUSAR": "PAUSE", "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", "MANTENER": "MAINTAIN", "REVISAR_CREATIVIDADES":"REVIEW_CREATIVES", } def _extract_vertical(name: str) -> str: """VIVIFUL_13_telefonia_leadads → telefonia""" prefix = config.META_CAMPAIGN_PREFIX rest = name[len(prefix):].lstrip("_") parts = rest.split("_") start = 1 if parts and parts[0].isdigit() else 0 return parts[start].lower() if start < len(parts) else "otros" 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 — {now.strftime('%d/%m/%Y %H:%M')}") print(f" Prefix: {config.META_CAMPAIGN_PREFIX} | Target CPL: {config.META_TARGET_CPL}€") print(f" Mode: {'DRY RUN (no changes)' if config.DRY_RUN else 'PRODUCTION'}") print(f"{'='*55}\n") meta = MetaAdsClient() baserow = BaserowClient() # ── Fetch all vertical CPL targets upfront ──────────────────────────────── vertical_cpls: dict = {} try: for v in baserow.get_all_verticals(): name = (v.get("Nombre") or "").strip().lower() cpl = float(v.get("target_cpl") or 0) if name and cpl: vertical_cpls[name] = cpl except Exception: pass # ── 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}") # ── Monthly daily totals (per-campaign rows → aggregate with margins) ───── print(f"→ Fetching monthly daily totals for {config.META_CAMPAIGN_PREFIX}...") daily_rows = meta.get_monthly_daily_totals() _daily: dict = {} monthly_verticals: dict = {} for row in daily_rows: v = _extract_vertical(row["campaign_name"]) target = vertical_cpls.get(v, config.META_TARGET_CPL) margin = round((target - row["spend"] / row["leads"]) * row["leads"], 2) if row["leads"] > 0 else round(-row["spend"], 2) d = _daily.setdefault(row["date"], {"spend": 0.0, "leads": 0, "margin": 0.0, "v_margins": {}}) d["spend"] += row["spend"] d["leads"] += row["leads"] d["margin"] += margin d["v_margins"][v] = d["v_margins"].get(v, 0.0) + margin mv = monthly_verticals.setdefault(v, {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": target}) mv["spend"] += row["spend"] mv["leads"] += row["leads"] mv["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), "v_margins": {v: round(m, 0) for v, m in d["v_margins"].items()}, } for date, d in sorted(_daily.items()) ] print(f" ✓ {len(daily_totals)} days with data.\n") # ── Yesterday metrics ───────────────────────────────────────────────────── print(f"→ Fetching yesterday metrics ({config.META_CAMPAIGN_PREFIX} only, spend > 0)...") metrics_all = meta.get_yesterday_metrics() print(f" ✓ {len(metrics_all)} campaigns active yesterday.\n") # ── 3-day and 7-day metrics for trend analysis ──────────────────────────── print(f"→ 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(f" ✓ Multi-window data ready.\n") # ── Analyze campaigns & propose actions ─────────────────────────────────── actions_proposed_list = [] campaign_details = {} # {cid: {vertical, margin, adsets, ads}} verticals = {} # {vertical: {spend, leads, margin}} errors = [] for cid, metrics in metrics_all.items(): vertical = _extract_vertical(metrics["name"]) max_cpl = vertical_cpls.get(vertical, config.META_TARGET_CPL) or config.META_TARGET_CPL margin = round((max_cpl - metrics["cpl"]) * metrics["leads"], 2) if metrics["leads"] > 0 else round(-metrics["spend"], 2) m3 = metrics_3d.get(cid, {}) m7 = metrics_7d.get(cid, {}) analysis = { "campaign_id": cid, "name": metrics["name"], "status": metrics["status"], # Yesterday (1d) — contexto inmediato "spend": metrics["spend"], "leads": metrics["leads"], "cpl": metrics["cpl"], "max_cpl": max_cpl, "ctr": metrics["ctr"], "cpm": metrics["cpm"], "impressions": metrics["impressions"], "clicks": metrics["clicks"], # 3 días — ventana principal para decisiones "spend_3d": m3.get("spend", 0.0), "leads_3d": m3.get("leads", 0), "cpl_3d": m3.get("cpl", 0.0), # 7 días — tendencia de fondo "spend_7d": m7.get("spend", 0.0), "leads_7d": m7.get("leads", 0), "cpl_7d": m7.get("cpl", 0.0), } try: decision = decide(analysis) except Exception as e: errors.append(f"{metrics['name']}: {e}") continue action_type = _ACTION_MAP.get( decision.get("action") or decision.get("accion", "MAINTAIN"), "MAINTAIN", ) # ── Bid config (campaña + adsets) ───────────────────────────────── campaign_bid = {} try: campaign_bid = meta.get_campaign_bid_config(cid) except Exception as e: errors.append(f"Bid config {metrics['name']}: {e}") adset_bids = {} try: adset_bids = meta.get_adset_bid_configs(cid) except Exception as e: errors.append(f"Adset bids {metrics['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" {metrics['name'][:52]}") print(f" CPL: 1d={metrics['cpl']}€ / 3d={m3.get('cpl',0.0):.2f}€ / 7d={m7.get('cpl',0.0):.2f}€ MaxCPL: {max_cpl}€ Margen: {margin:+.2f}€") print(f" Leads: 1d={metrics['leads']} / 3d={m3.get('leads',0)} / 7d={m7.get('leads',0)} Spend 1d: {metrics['spend']}€") print(f" Vertical: {vertical} 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": metrics["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": metrics["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, "cpl": metrics["cpl"], "max_cpl": max_cpl, "row_id": row["id"], }) except Exception as e: errors.append(f"Save action {metrics['name']}: {e}") # ── Ad set analysis ──────────────────────────────────────────────── 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 {metrics['name']}: {e}") # ── Ad analysis ──────────────────────────────────────────────────── ads_detail = [] try: for ad_m in meta.get_period_ad_metrics(cid, days=3)[:5]: ad_m["max_cpl"] = max_cpl result = analyze_unit(ad_m, "ad") ad_entry = {**ad_m, **result} # Propose ad pause if Claude recommends it 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 {metrics['name']}: {e}") campaign_details[cid] = { "name": metrics["name"], "vertical": vertical, "margin": margin, "adsets": adsets_detail, "ads": ads_detail, "bid_config": campaign_bid, } # ── Daily snapshot (persists analysis to Baserow for dashboard) ─────── try: action_info = next( (a for a in actions_proposed_list if a["campaign_name"] == metrics["name"]), None, ) baserow.save_daily_snapshot({ "run_date": now.strftime("%Y-%m-%d"), "campaign_id": cid, "campaign_name": metrics["name"], "vertical": vertical, "spend": metrics["spend"], "leads": metrics["leads"], "cpl": metrics["cpl"], "margin": margin, "action_type": action_info["action_type"] if action_info else "MAINTAIN", "justification": action_info["justification"] if action_info else "", "adsets": adsets_detail, "ads": ads_detail, }) except Exception as e: errors.append(f"Snapshot {metrics['name']}: {e}") # ── Vertical aggregation ─────────────────────────────────────────── if vertical not in verticals: verticals[vertical] = {"spend": 0.0, "leads": 0, "margin": 0.0, "target_cpl": max_cpl} verticals[vertical]["spend"] += metrics["spend"] verticals[vertical]["leads"] += metrics["leads"] verticals[vertical]["margin"] += margin # ── Creative visual analysis (Baserow storage) ───────────────────── try: for ad in meta.get_ads_with_creatives(cid): if not ad["thumbnail_url"]: continue result = analyze_creative(ad["thumbnail_url"], ad["ad_name"]) baserow.save_creative_analysis({ "ad_id": ad["ad_id"], "ad_name": ad["ad_name"], "campaign_id": cid, "image_url": ad["thumbnail_url"], "analysis": result.get("analysis", ""), "score": result.get("score", 0), "recommendations": result.get("recommendations", ""), }) except Exception as e: errors.append(f"Creatives {metrics['name']}: {e}") # ── Top 10 best and worst ───────────────────────────────────────────────── with_leads = [m for m in metrics_all.values() if m["leads"] > 0] best_10 = sorted(with_leads, key=lambda x: x["cpl"])[:10] all_active = list(metrics_all.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, target_cpl=config.META_TARGET_CPL, campaigns_analyzed=len(metrics_all), mode="DRY_RUN" if config.DRY_RUN else "PRODUCTION", verticals=verticals, campaign_details=campaign_details, monthly_verticals=monthly_verticals, ) except Exception as e: print(f" Warning: Slack notification failed: {e}") # ── Execution log ───────────────────────────────────────────────────────── summary = ( f"{len(metrics_all)} 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(metrics_all), "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