José Manuel Gómez
9239e2f67f
Ports meta-optimizer's Meta Ads execution/approval/creative-analysis layer (agent.py, meta_ads_client.py, baserow_client.py, slack_notifier.py, approval_server.py) and replaces the per-vertical CPL model with the PPL + monthly-capping-per-course model already used by leads-optimizer, via a new airtable_client.py that shares Cursos/Familias/CentroCurso/ CursoMes/Leads Lake with that project and adds Meta Ads Campaigns / MetaCampaignMes alongside its Google Ads Campaigns / GACampaignMes.
209 lines
7.9 KiB
Python
209 lines
7.9 KiB
Python
"""
|
||
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()
|