""" Backfill: genera snapshots históricos con análisis Claude para un rango de fechas. Uso: python backfill.py # mes en curso → ayer python backfill.py --from 2026-06-01 --to 2026-06-04 python backfill.py --skip-existing # no reprocesa días ya guardados """ import sys import io import argparse sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True) from datetime import datetime, timedelta import config from meta_ads_client import MetaAdsClient from agent import decide, analyze_unit from baserow_client import BaserowClient _ACTION_MAP = { "PAUSE": "PAUSE", "REDUCE_BUDGET": "REDUCE_BUDGET", "INCREASE_BUDGET": "INCREASE_BUDGET", "MAINTAIN": "MAINTAIN", "REVIEW_CREATIVES": "REVIEW_CREATIVES", "PAUSAR": "PAUSE", "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", "MANTENER": "MAINTAIN", "REVISAR_CREATIVIDADES": "REVIEW_CREATIVES", } def _extract_vertical(name: str) -> str: 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" def run_backfill(date_from: str, date_to: str, skip_existing: bool = False): meta = MetaAdsClient() baserow = BaserowClient() # Vertical CPL targets 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 as e: print(f"Warning: could not fetch verticals: {e}") # Build date list d = datetime.strptime(date_from, "%Y-%m-%d") d_end = datetime.strptime(date_to, "%Y-%m-%d") dates = [] while d <= d_end: dates.append(d.strftime("%Y-%m-%d")) d += timedelta(days=1) print(f"\n{'='*60}") print(f" BACKFILL {date_from} → {date_to} ({len(dates)} días)") print(f"{'='*60}\n") total_saved = 0 total_skip = 0 for run_date in dates: print(f"\n── {run_date} ───────────────────────────────────────────────") # Pre-load existing snapshots for this date if skip_existing existing_names: set = set() if skip_existing: try: for r in baserow.get_snapshots_for_date(run_date): existing_names.add(r.get("campaign_name", "")) except Exception: pass campaign_metrics = meta.get_campaign_metrics(run_date, run_date) if not campaign_metrics: print(" Sin campañas con gasto.") continue print(f" {len(campaign_metrics)} campañas activas.") # Adset bid configs (current — bid strategy doesn't change day to day) adset_bids_cache: dict = {} for cid, metrics in campaign_metrics.items(): camp_name = metrics["name"] if skip_existing and camp_name in existing_names: print(f" SKIP {camp_name[:55]}") total_skip += 1 continue vertical = _extract_vertical(camp_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) print(f" {camp_name[:55]}") print(f" Spend {metrics['spend']}€ Leads {metrics['leads']} CPL {metrics['cpl']}€ MaxCPL {max_cpl}€ Margen {margin:+.2f}€") # ── Claude: decisión ──────────────────────────────────────────── analysis = { "campaign_id": cid, "name": camp_name, "status": "ACTIVE", "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"], } try: decision = decide(analysis) action_type = _ACTION_MAP.get( decision.get("action") or decision.get("accion", "MAINTAIN"), "MAINTAIN", ) except Exception as e: print(f" ERROR decide: {e}") decision = {"action": "MAINTAIN", "justification": "", "parameter": 1.0} action_type = "MAINTAIN" print(f" Decision: {action_type} — {(decision.get('justification') or '')[:70]}") # ── Claude: adsets ────────────────────────────────────────────── adsets_detail = [] try: for as_m in meta.get_adset_metrics(cid, run_date, run_date)[:5]: result = analyze_unit(as_m, "adset") adsets_detail.append({**as_m, **result}) print(f" [Adset] {as_m['name'][:45]} — {result.get('evaluacion','')[:50]}") except Exception as e: print(f" ERROR adsets: {e}") # Add bid configs (cached per campaign) if cid not in adset_bids_cache: try: adset_bids_cache[cid] = meta.get_adset_bid_configs(cid) except Exception: adset_bids_cache[cid] = {} for adset in adsets_detail: b = adset_bids_cache[cid].get(adset["id"], {}) adset["cost_cap_eur"] = b.get("cost_cap_eur") adset["bid_strategy"] = b.get("bid_strategy", "") # ── Claude: anuncios ──────────────────────────────────────────── ads_detail = [] try: for ad_m in meta.get_ad_metrics(cid, run_date, run_date)[:5]: result = analyze_unit(ad_m, "ad") ads_detail.append({**ad_m, **result}) print(f" [Ad] {ad_m['name'][:45]} — {result.get('evaluacion','')[:50]}") except Exception as e: print(f" ERROR ads: {e}") # ── Guardar snapshot ──────────────────────────────────────────── try: baserow.save_daily_snapshot({ "run_date": run_date, "campaign_id": cid, "campaign_name": camp_name, "vertical": vertical, "spend": metrics["spend"], "leads": metrics["leads"], "cpl": metrics["cpl"], "margin": margin, "action_type": action_type, "justification": decision.get("justification") or "", "adsets": adsets_detail, "ads": ads_detail, }) print(f" ✓ Snapshot guardado") total_saved += 1 except Exception as e: print(f" ERROR snapshot: {e}") print(f"\n{'='*60}") print(f" Backfill completo. Guardados: {total_saved} Saltados: {total_skip}") print(f"{'='*60}\n") if __name__ == "__main__": now = datetime.now() default_from = f"{now.year}-{now.month:02d}-01" default_to = (now - timedelta(days=1)).strftime("%Y-%m-%d") parser = argparse.ArgumentParser(description="Backfill Meta Optimizer snapshots") parser.add_argument("--from", dest="date_from", default=default_from, help=f"Fecha inicio YYYY-MM-DD (default: {default_from})") parser.add_argument("--to", dest="date_to", default=default_to, help=f"Fecha fin YYYY-MM-DD (default: {default_to})") parser.add_argument("--skip-existing", action="store_true", help="No reprocesa campañas que ya tienen snapshot ese día") args = parser.parse_args() run_backfill(args.date_from, args.date_to, args.skip_existing)