""" Backfill: genera snapshots históricos con análisis Claude para un rango de fechas. Usa ventana de 1 día (no 3d/7d, los datos históricos ya están fijados) y reconstruye el capping/PPL/familia y los leads acumulados del curso tal como estaban en cada fecha histórica (as_of_date), no el estado actual. 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 airtable_client import AirtableClient, extract_cursoid from agent import decide, analyze_unit from baserow_client import BaserowClient import analyzer _ACTION_MAP = { "PAUSE": "PAUSE", "REDUCE_BUDGET": "REDUCE_BUDGET", "INCREASE_BUDGET": "INCREASE_BUDGET", "MAINTAIN": "MAINTAIN", "PAUSAR": "PAUSE", "REDUCIR_PRESUPUESTO": "REDUCE_BUDGET", "AUMENTAR_PRESUPUESTO": "INCREASE_BUDGET", "MANTENER": "MAINTAIN", } def run_backfill(date_from: str, date_to: str, skip_existing: bool = False): meta = MetaAdsClient() baserow = BaserowClient() airtable = AirtableClient() # 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 FORMACIÓN {date_from} → {date_to} ({len(dates)} días)") print(f"{'='*60}\n") total_saved = 0 total_skip = 0 _lookups_cache: dict = {} # {mes_año: (ppl_lookup, cap_lookup, familia_lookup)} for run_date in dates: print(f"\n── {run_date} ───────────────────────────────────────────────") mes_key = run_date[:7] if mes_key not in _lookups_cache: _lookups_cache[mes_key] = airtable.build_campaign_lookups(as_of_date=run_date) ppl_lookup, cap_lookup, familia_lookup = _lookups_cache[mes_key] # 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_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 cursoid = extract_cursoid(camp_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(camp_name, as_of_date=run_date) print(f" {camp_name[:55]}") print(f" Spend {metrics['spend']}€ Leads {metrics['leads']} PPL {ppl}€ " f"CPAmax {cpa_max}€ Leads mes {leads_entregados}/{cap or '∞'}") campaign_config = { "curso": camp_name, "meta_campaign_id": cid, "ppl": ppl, "cpa_maximo": cpa_max, "capping_mensual": cap, "conv_leads_lake_mes": leads_entregados, } ads_metrics = { "spend": metrics["spend"], "leads": metrics["leads"], "ctr": metrics["ctr"], "clicks": metrics["clicks"], "status": "ACTIVE", } analysis = analyzer.analyze(campaign_config, leads_entregados, ads_metrics) try: decision = decide(analysis) action_type = _ACTION_MAP.get(decision.get("action", "MAINTAIN"), "MAINTAIN") except Exception as e: print(f" ERROR decide: {e}") decision = {"action": "MAINTAIN", "justification": "", "parameter": 1.0} action_type = "MAINTAIN" print(f" Urgencia: {analysis['urgencia']} Decision: {action_type} — " f"{(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}") 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]: ad_m["ppl"] = ppl ad_m["cpa_maximo"] = cpa_max 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}") # margin en € (mismo proxy que usa run.py): leads*PPL - gasto margin = round(metrics["leads"] * ppl - metrics["spend"], 2) try: baserow.save_daily_snapshot({ "run_date": run_date, "campaign_id": cid, "campaign_name": camp_name, "familia": familia, "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(" ✓ 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 Formación 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)