José Manuel Gómez
769d86c896
- Broaden Airtable lead counting to attr_utm_source IN ('Lead ads','landingpage')
— the 'landingpage' leads (100% fbclid, 0% gclid) were being missed entirely,
undercounting real leads for '_web' suffixed campaigns and skewing
capping/pacing decisions since yesterday's first production run.
- Add airtable_client.get_meta_leads_bulk() for day/curso-level aggregation.
- Drop per-familia Slack sectioning in favor of a single Formación block,
chunked by campaign batches instead.
- Add daily AT-vs-Meta table, per-curso PPL/CPL contrast table (leadform vs
landing breakdown), and a Claude-generated portfolio strategic diagnosis
(ported from leads-optimizer's portfolio_daily_analysis).
- Persist daily aggregate totals to a new Baserow table (daily_metrics) so
the dashboard and future reports don't depend on Meta's historical API
access remaining available indefinitely.
- Surface adset/ad-level recommendations in the campaign cards instead of
only numeric tables.
251 lines
11 KiB
Python
251 lines
11 KiB
Python
"""Baserow REST API client for meta_optimizer_formacion tables.
|
|
|
|
A diferencia de meta-optimizer (Viviful), aquí NO hay tablas 'campaigns' ni
|
|
'verticals' — esa capa de negocio (PPL, capping, familia) vive en Airtable
|
|
(ver airtable_client.py). Baserow solo guarda lo operativo de Meta: acciones
|
|
propuestas, snapshots diarios, análisis de creatividades y logs de ejecución.
|
|
"""
|
|
import requests
|
|
from datetime import datetime
|
|
import config
|
|
|
|
|
|
class BaserowClient:
|
|
def __init__(self):
|
|
self._base = config.BASEROW_URL.rstrip("/")
|
|
self._headers = {
|
|
"Authorization": f"Token {config.BASEROW_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
def _url(self, table_id: int, row_id: int = None) -> str:
|
|
path = f"/api/database/rows/table/{table_id}/"
|
|
if row_id:
|
|
path += f"{row_id}/"
|
|
return self._base + path
|
|
|
|
def _get_rows(self, table_id: int, filters: dict = None) -> list:
|
|
params = {"user_field_names": "true"}
|
|
if filters:
|
|
params.update(filters)
|
|
try:
|
|
resp = requests.get(self._url(table_id), headers=self._headers, params=params, timeout=15)
|
|
if not resp.ok:
|
|
return []
|
|
return resp.json().get("results", [])
|
|
except requests.RequestException:
|
|
return []
|
|
|
|
def _create_row(self, table_id: int, data: dict) -> dict:
|
|
resp = requests.post(
|
|
self._url(table_id),
|
|
headers=self._headers,
|
|
params={"user_field_names": "true"},
|
|
json=data,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def _delete_row(self, table_id: int, row_id: int):
|
|
resp = requests.delete(
|
|
self._url(table_id, row_id),
|
|
headers=self._headers,
|
|
params={"user_field_names": "true"},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
|
|
def _update_row(self, table_id: int, row_id: int, data: dict) -> dict:
|
|
resp = requests.patch(
|
|
self._url(table_id, row_id),
|
|
headers=self._headers,
|
|
params={"user_field_names": "true"},
|
|
json=data,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
# ── proposed_actions ──────────────────────────────────────────────────────
|
|
|
|
def save_action(self, action: dict) -> dict:
|
|
return self._create_row(config.BASEROW_TABLE_ACTIONS, {
|
|
"campaign_id": action["campaign_id"],
|
|
"campaign_name": action["campaign_name"],
|
|
"action_type": action["action_type"],
|
|
"parameter": action.get("parameter", 1.0),
|
|
"justification": action.get("justification", ""),
|
|
"advice": action.get("advice", ""),
|
|
"alert": action.get("alert") or "",
|
|
"confidence": action.get("confidence", 0.0),
|
|
"status": "pending",
|
|
"proposed_at": datetime.now().strftime("%Y-%m-%d"),
|
|
})
|
|
|
|
def get_approved_actions(self) -> list:
|
|
return self._get_rows(
|
|
config.BASEROW_TABLE_ACTIONS,
|
|
{"filter__status__single_select_equal": "approved"},
|
|
)
|
|
|
|
def update_action_status(
|
|
self, row_id: int, status: str, slack_message_ts: str = None
|
|
) -> dict:
|
|
data: dict = {"status": status}
|
|
if status == "executed":
|
|
data["executed_at"] = datetime.now().strftime("%Y-%m-%d")
|
|
if slack_message_ts is not None:
|
|
data["slack_message_ts"] = slack_message_ts
|
|
return self._update_row(config.BASEROW_TABLE_ACTIONS, row_id, data)
|
|
|
|
# ── creative_analyses ─────────────────────────────────────────────────────
|
|
|
|
def get_all_creative_analyses(self, filters: dict = None) -> list:
|
|
"""Returns creative analyses from Baserow, up to 200 rows."""
|
|
params = {"user_field_names": "true", "page_size": 200, "order_by": "-created_at"}
|
|
if filters:
|
|
params.update(filters)
|
|
try:
|
|
resp = requests.get(
|
|
self._url(config.BASEROW_TABLE_CREATIVES),
|
|
headers=self._headers, params=params, timeout=15,
|
|
)
|
|
if not resp.ok:
|
|
return []
|
|
return resp.json().get("results", [])
|
|
except requests.RequestException:
|
|
return []
|
|
|
|
def get_creative_history_by_ad(self, ad_id: str) -> list:
|
|
"""Returns all analyses for an ad_id sorted by date ascending (for score evolution)."""
|
|
rows = self._get_rows(
|
|
config.BASEROW_TABLE_CREATIVES,
|
|
{"filter__ad_id__equal": ad_id},
|
|
)
|
|
return sorted(rows, key=lambda r: r.get("created_at", ""))
|
|
|
|
def save_creative_analysis(self, analysis: dict) -> dict:
|
|
return self._create_row(config.BASEROW_TABLE_CREATIVES, {
|
|
"ad_id": analysis["ad_id"],
|
|
"ad_name": analysis["ad_name"],
|
|
"campaign_id": analysis["campaign_id"],
|
|
"image_url": analysis["image_url"],
|
|
"analysis": analysis.get("analysis", ""),
|
|
"score": analysis.get("score", 0),
|
|
"recommendations": analysis.get("recommendations", ""),
|
|
"created_at": datetime.now().strftime("%Y-%m-%d"),
|
|
})
|
|
|
|
# ── daily_snapshots ───────────────────────────────────────────────────────
|
|
|
|
def save_daily_snapshot(self, snapshot: dict) -> dict:
|
|
import json
|
|
|
|
# Remove existing snapshots for same day + campaign before saving new one
|
|
existing = self._get_rows(
|
|
config.BASEROW_TABLE_SNAPSHOTS,
|
|
{
|
|
"filter__run_date__equal": snapshot["run_date"],
|
|
"filter__campaign_name__equal": snapshot["campaign_name"],
|
|
},
|
|
)
|
|
for row in existing:
|
|
try:
|
|
self._delete_row(config.BASEROW_TABLE_SNAPSHOTS, row["id"])
|
|
except Exception:
|
|
pass
|
|
|
|
def _safe_json(items: list) -> str:
|
|
cleaned = []
|
|
for item in items:
|
|
cleaned.append({
|
|
k: (str(v)[:500] if isinstance(v, str) else v)
|
|
for k, v in item.items()
|
|
if isinstance(v, (str, int, float, bool, type(None)))
|
|
})
|
|
s = json.dumps(cleaned, ensure_ascii=False)
|
|
return s[:60000] # Baserow long_text practical limit
|
|
|
|
resp = requests.post(
|
|
self._url(config.BASEROW_TABLE_SNAPSHOTS),
|
|
headers=self._headers,
|
|
params={"user_field_names": "true"},
|
|
json={
|
|
"run_date": snapshot["run_date"],
|
|
"campaign_id": snapshot["campaign_id"],
|
|
"campaign_name": snapshot["campaign_name"],
|
|
"familia": snapshot["familia"],
|
|
"spend": float(snapshot["spend"]),
|
|
"leads": int(snapshot["leads"]),
|
|
"cpl": float(snapshot["cpl"]),
|
|
"margin": float(snapshot["margin"]),
|
|
"action_type": snapshot.get("action_type", "MAINTAIN"),
|
|
"justification": (snapshot.get("justification") or "")[:2000],
|
|
"adsets_json": _safe_json(snapshot.get("adsets", [])),
|
|
"ads_json": _safe_json(snapshot.get("ads", [])),
|
|
},
|
|
timeout=15,
|
|
)
|
|
if not resp.ok:
|
|
raise Exception(f"{resp.status_code} {resp.text[:300]}")
|
|
return resp.json()
|
|
|
|
def get_snapshots_for_date(self, run_date: str) -> list:
|
|
return self._get_rows(
|
|
config.BASEROW_TABLE_SNAPSHOTS,
|
|
{"filter__run_date__equal": run_date},
|
|
)
|
|
|
|
def get_snapshot_dates(self) -> list:
|
|
"""Return sorted list of distinct run_date values that have snapshots."""
|
|
rows = self._get_rows(config.BASEROW_TABLE_SNAPSHOTS)
|
|
return sorted({r["run_date"] for r in rows if r.get("run_date")}, reverse=True)
|
|
|
|
# ── daily_metrics (totales agregados del bloque Formación, uno por día) ────
|
|
# Se persisten para no depender de poder re-pedirle a Meta el histórico
|
|
# diario más adelante, y para que el dashboard los lea sin llamar a la API.
|
|
|
|
def save_daily_metrics(self, row: dict) -> dict:
|
|
existing = self._get_rows(
|
|
config.BASEROW_TABLE_DAILY_METRICS,
|
|
{"filter__date__equal": row["date"]},
|
|
)
|
|
for r in existing:
|
|
try:
|
|
self._delete_row(config.BASEROW_TABLE_DAILY_METRICS, r["id"])
|
|
except Exception:
|
|
pass
|
|
return self._create_row(config.BASEROW_TABLE_DAILY_METRICS, {
|
|
"date": row["date"],
|
|
"spend": float(row.get("spend", 0)),
|
|
"leads_meta": int(row.get("leads_meta", 0)),
|
|
"leads_at": int(row.get("leads_at", 0)),
|
|
"ing_meta": float(row.get("ing_meta", 0)),
|
|
"ing_at": float(row.get("ing_at", 0)),
|
|
"margin": float(row.get("margin", 0)),
|
|
"margin_pct": float(row.get("margin_pct", 0)),
|
|
})
|
|
|
|
def get_daily_metrics(self, date_from: str = None, date_to: str = None) -> list:
|
|
rows = self._get_rows(config.BASEROW_TABLE_DAILY_METRICS)
|
|
if date_from:
|
|
rows = [r for r in rows if r.get("date", "") >= date_from]
|
|
if date_to:
|
|
rows = [r for r in rows if r.get("date", "") <= date_to]
|
|
return sorted(rows, key=lambda r: r.get("date", ""))
|
|
|
|
# ── execution_logs ────────────────────────────────────────────────────────
|
|
|
|
def save_execution_log(self, log: dict) -> dict:
|
|
return self._create_row(config.BASEROW_TABLE_LOGS, {
|
|
"executed_at": datetime.now().strftime("%Y-%m-%d"),
|
|
"mode": log.get("mode", "DRY_RUN"),
|
|
"campaigns_analyzed": log.get("campaigns_analyzed", 0),
|
|
"actions_proposed": log.get("actions_proposed", 0),
|
|
"actions_executed": log.get("actions_executed", 0),
|
|
"errors": log.get("errors", ""),
|
|
"summary": log.get("summary", ""),
|
|
"duration_seconds": log.get("duration_seconds", 0.0),
|
|
})
|