José Manuel Gómez
92786e94a8
- meta_ads_client.py: Meta Marketing API client (facebook-business SDK) - agent.py: Claude-powered campaign decision engine - run.py: main orchestration script - config.py: environment variables - .github/workflows/daily.yml: GitHub Actions cron (8am CEST) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""
|
|
Meta Optimizer — punto de entrada principal.
|
|
Analiza campañas de Meta Ads y publica resumen en Slack.
|
|
"""
|
|
import sys
|
|
import io
|
|
import os
|
|
import json
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", line_buffering=True)
|
|
|
|
from meta_ads_client import MetaAdsClient
|
|
from agent import decide
|
|
import config
|
|
from datetime import datetime
|
|
|
|
|
|
class Tee:
|
|
def __init__(self, filepath):
|
|
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 run():
|
|
now = datetime.now()
|
|
print(f"\n{'='*55}")
|
|
print(f" META OPTIMIZER — {now.strftime('%d/%m/%Y %H:%M')}")
|
|
print(f" Modo: {'DRY RUN (sin cambios)' if config.DRY_RUN else 'PRODUCCIÓN'}")
|
|
print(f"{'='*55}\n")
|
|
|
|
meta = MetaAdsClient()
|
|
|
|
print("→ Obteniendo métricas del mes desde Meta Ads...")
|
|
metrics_all = meta.get_monthly_metrics_all()
|
|
print(f" ✓ {len(metrics_all)} campañas encontradas.\n")
|
|
|
|
results = []
|
|
for cid, metrics in metrics_all.items():
|
|
analysis = {
|
|
"campaign_id": cid,
|
|
"name": metrics["name"],
|
|
"status": metrics["status"],
|
|
"spend": metrics["spend"],
|
|
"leads": metrics["leads"],
|
|
"cpl": metrics["cpl"],
|
|
"cpl_maximo": 0, # TODO: cargar desde Airtable o config por campaña
|
|
"ctr": metrics["ctr"],
|
|
"cpm": metrics["cpm"],
|
|
"impressions": metrics["impressions"],
|
|
"clicks": metrics["clicks"],
|
|
}
|
|
decision = decide(analysis)
|
|
results.append({"metrics": metrics, "analysis": analysis, "decision": decision})
|
|
|
|
print(f"📢 {metrics['name'][:50]}")
|
|
print(f" Gasto: {metrics['spend']}€ | Leads: {metrics['leads']} | CPL: {metrics['cpl']}€")
|
|
print(f" Decisión: {decision['accion']} — {decision['justificacion'][:80]}")
|
|
if decision.get("alerta"):
|
|
print(f" 🚨 {decision['alerta']}")
|
|
print()
|
|
|
|
print(f"Log guardado en: logs/{now.strftime('%Y%m%d_%H%M%S')}.log")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
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
|