""" Análisis profundo de creatividades de Meta Ads. Analiza visualmente cada anuncio activo, correlaciona con métricas de rendimiento, detecta fatiga creativa y compara anuncios dentro del mismo adset. Uso: python analyze_creatives.py # todas las campañas python analyze_creatives.py --campaign RoiFormacion_884 # filtrar por nombre python analyze_creatives.py --no-slack # sin envío a Slack """ import argparse import sys import time from datetime import datetime sys.stdout.reconfigure(encoding="utf-8", errors="replace") import config from meta_ads_client import MetaAdsClient from airtable_client import AirtableClient, extract_cursoid from baserow_client import BaserowClient from agent import analyze_creative_deep, compare_adset_creatives from slack_notifier import send_creative_analysis_report def main(): parser = argparse.ArgumentParser(description="Deep creative analysis for Meta Ads") parser.add_argument("--campaign", help="Filter campaigns by name substring (case-insensitive)") parser.add_argument("--no-slack", action="store_true", help="Skip Slack report") args = parser.parse_args() meta = MetaAdsClient() db = BaserowClient() airtable = AirtableClient() print(f"\n{'='*60}") print(f" ANÁLISIS DE CREATIVIDADES FORMACIÓN — {datetime.now().strftime('%d/%m/%Y %H:%M')}") print(f"{'='*60}\n") ppl_lookup, _, _ = airtable.build_campaign_lookups() # Active campaigns (last 7 days) campaigns = meta.get_period_campaign_metrics(7) if args.campaign: campaigns = {k: v for k, v in campaigns.items() if args.campaign.upper() in v["name"].upper()} if not campaigns: print("No hay campañas activas en los últimos 7 días.") return print(f"→ {len(campaigns)} campañas a analizar\n") all_results: dict = {} total_analyzed = 0 total_errors = 0 for cid, camp_metrics in campaigns.items(): campaign_name = camp_metrics["name"] ppl = ppl_lookup.get(extract_cursoid(campaign_name) or "", 0) cpa_maximo = round(ppl * 0.70, 2) if ppl else 0.0 print(f" ● {campaign_name} (PPL: {ppl:.2f}€ · CPA máximo: {cpa_maximo:.2f}€)") ads_with_creatives = meta.get_ads_with_creatives(cid) if not ads_with_creatives: print(" — sin anuncios activos con creatividades, omitiendo\n") continue # Metrics for both windows ads_7d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, 7)} ads_3d = {a["id"]: a for a in meta.get_period_ad_metrics(cid, 3)} # Adset name lookup from 7d metrics adset_names = {a["id"]: a["name"] for a in meta.get_period_adset_metrics(cid, 7)} # Group ads by adset adset_groups: dict = {} for ad in ads_with_creatives: if not ad.get("thumbnail_url"): continue ad_id = ad["ad_id"] adset_id = ad.get("adset_id", "unknown") adset_name = adset_names.get(adset_id, adset_id) m7 = ads_7d.get(ad_id, {}) m3 = ads_3d.get(ad_id, {}) metrics = { "spend_7d": m7.get("spend", 0), "leads_7d": m7.get("leads", 0), "cpl_7d": m7.get("cpl", 0), "ctr_7d": m7.get("ctr", 0), "spend_3d": m3.get("spend", 0), "leads_3d": m3.get("leads", 0), "cpl_3d": m3.get("cpl", 0), "ctr_3d": m3.get("ctr", 0), "cpa_maximo": cpa_maximo, } if adset_id not in adset_groups: adset_groups[adset_id] = {"name": adset_name, "ads": []} adset_groups[adset_id]["ads"].append({ "ad_id": ad_id, "ad_name": ad["ad_name"], "campaign_id": cid, "adset_id": adset_id, "adset_name": adset_name, # Fallback chain: signed thumbnail → permanent video picture → static image_url "image_url": [ad["thumbnail_url"], ad["video_thumbnail_url"], ad["image_url"]], **metrics, }) # Analyze each ad individually analyzed_adsets: dict = {} for adset_id, adset_data in adset_groups.items(): adset_name = adset_data["name"] analyzed_ads = [] for ad in adset_data["ads"]: short_name = ad["ad_name"][:50] print(f" [{adset_name[:30]}] {short_name}...", end=" ", flush=True) result = analyze_creative_deep( image_url=ad["image_url"], ad_name=ad["ad_name"], metrics={k: ad[k] for k in ( "spend_7d", "leads_7d", "cpl_7d", "ctr_7d", "spend_3d", "leads_3d", "cpl_3d", "ctr_3d", "cpa_maximo" )}, ) score = result.get("score", 0) fatigue_flag = " ⚠️FATIGA" if result.get("fatigue") else "" print(f"score={score:.1f}{fatigue_flag}") ad_result = {**ad, **result} analyzed_ads.append(ad_result) # Save to Baserow analysis_text = result.get("analysis", "") if result.get("fatigue") and result.get("fatigue_reason"): analysis_text += f"\n\n⚠️ FATIGA CREATIVA: {result['fatigue_reason']}" try: urls = ad["image_url"] saved_url = urls[0] if isinstance(urls, list) else urls db.save_creative_analysis({ "ad_id": ad["ad_id"], "ad_name": ad["ad_name"], "campaign_id": cid, "image_url": saved_url, "analysis": analysis_text, "score": score, "recommendations": result.get("recommendations", ""), }) total_analyzed += 1 except Exception as e: print(f" [WARN] Baserow: {e}") total_errors += 1 time.sleep(0.3) # Comparative analysis for adsets with 2+ ads comparison = None if len(analyzed_ads) >= 2: print(f" [comparativa] {adset_name[:40]}...", end=" ", flush=True) comparison = compare_adset_creatives(analyzed_ads) winner = comparison.get("winner", "") print(f"ganador: {winner[:40]}") analyzed_adsets[adset_id] = { "name": adset_name, "ads": analyzed_ads, "comparison": comparison, } all_results[cid] = { "name": campaign_name, "cpa_maximo": cpa_maximo, "adsets": analyzed_adsets, } print() # Summary total_ads = sum(len(as_d["ads"]) for c in all_results.values() for as_d in c["adsets"].values()) total_fatigue = sum( 1 for c in all_results.values() for as_d in c["adsets"].values() for ad in as_d["ads"] if ad.get("fatigue") ) print(f"{'='*60}") print(f" Finalizado: {len(all_results)} campañas, {total_ads} anuncios analizados") if total_fatigue: print(f" ⚠️ {total_fatigue} anuncios con fatiga creativa detectada") if total_errors: print(f" ⚠️ {total_errors} errores al guardar en Baserow") print(f"{'='*60}\n") # Slack report if not args.no_slack and all_results: print("→ Enviando informe a Slack...") send_creative_analysis_report(all_results) print(" ✓ Informe enviado.") if __name__ == "__main__": main()