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.
385 lines
17 KiB
Python
385 lines
17 KiB
Python
"""
|
|
Client for the shared Airtable base (same base used by leads-optimizer).
|
|
|
|
Reused as-is from leads-optimizer: Cursos / CentroCurso / CursoMes / Leads Lake
|
|
are READ-ONLY here — the course catalog, PPL per center and monthly capping are
|
|
maintained externally (manually, or by leads-optimizer). This client only
|
|
creates/updates "Meta Ads Campaigns" and "MetaCampaignMes", the Meta-specific
|
|
tables that sit alongside "Google Ads Campaigns" / "GACampaignMes".
|
|
"""
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from pyairtable import Api
|
|
import config
|
|
|
|
# Los leads de Meta llegan a Leads Lake por dos vías, ambas confirmadas 100%
|
|
# atribuibles a Meta (fbclid presente, 0% gclid) aunque la web las etiqueta
|
|
# distinto: 'Lead ads' = leadform nativo de Meta (attr_referer = nombre de
|
|
# campaña); 'landingpage' = clic a landing (attr_referer = URL con fbclid).
|
|
META_UTM_SOURCES = ("Lead ads", "landingpage")
|
|
|
|
MESES_ES = {
|
|
1: "Enero", 2: "Febrero", 3: "Marzo", 4: "Abril",
|
|
5: "Mayo", 6: "Junio", 7: "Julio", 8: "Agosto",
|
|
9: "Septiembre", 10: "Octubre", 11: "Noviembre", 12: "Diciembre",
|
|
}
|
|
|
|
|
|
def extract_cursoid(campaign_name: str) -> str | None:
|
|
"""
|
|
RoiFormacion_884_Curso_Desarrollador_leadads -> "884"
|
|
RoiFormacion_1281Instaladores_24_leadads -> "1281" (sin guion bajo tras el id)
|
|
Tolera también la variante con tilde (RoiFormación) por si reaparece.
|
|
"""
|
|
m = re.search(r'roiformaci[oó]n_?(\d+)', campaign_name, re.IGNORECASE)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
class AirtableClient:
|
|
def __init__(self):
|
|
self.api = Api(config.AIRTABLE_TOKEN)
|
|
self.leads = self.api.table(config.AIRTABLE_BASE_ID, config.LEADS_TABLE)
|
|
self.cursos = self.api.table(config.AIRTABLE_BASE_ID, config.CURSOS_TABLE)
|
|
self.centrocurso = self.api.table(config.AIRTABLE_BASE_ID, config.CENTROCURSO_TABLE)
|
|
self.cursomes = self.api.table(config.AIRTABLE_BASE_ID, config.CURSOMES_TABLE)
|
|
self.campaigns = self.api.table(config.AIRTABLE_BASE_ID, config.META_CAMPAIGNS_TABLE)
|
|
self.metacampaignmes = self.api.table(config.AIRTABLE_BASE_ID, config.META_CAMPAIGNMES_TABLE)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Lookups de negocio (PPL, capping, familia) — solo lectura #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def build_campaign_lookups(self, as_of_date: str = None) -> tuple[dict, dict, dict]:
|
|
"""
|
|
3 llamadas bulk, igual que leads-optimizer, más el lookup de Familia
|
|
(no existe en leads-optimizer porque Google no lo necesitaba para nada
|
|
operativo; aquí sustituye al concepto de "vertical" de Viviful).
|
|
|
|
as_of_date (YYYY-MM-DD): usa el capping del mes de esa fecha en vez del
|
|
mes en curso (lo usa backfill.py para reconstruir estados pasados).
|
|
|
|
Devuelve (ppl_lookup, cap_lookup, familia_lookup), todos keyed por
|
|
cursoid_text.
|
|
"""
|
|
ref = datetime.strptime(as_of_date, "%Y-%m-%d") if as_of_date else datetime.now()
|
|
mes_nombre = MESES_ES[ref.month]
|
|
anio_str = str(ref.year)
|
|
|
|
cc_records = self.centrocurso.all(
|
|
formula="{Estado ROI}='ABIERTO'",
|
|
fields=["PPL", "% invalidación (from Centros)"],
|
|
)
|
|
cc_data = {}
|
|
for r in cc_records:
|
|
ppl_val = float(r["fields"].get("PPL") or 0)
|
|
pct_raw = r["fields"].get("% invalidación (from Centros)", 0)
|
|
pct = float(pct_raw[0]) if isinstance(pct_raw, list) and pct_raw else float(pct_raw or 0)
|
|
cc_data[r["id"]] = {"ppl": ppl_val, "pct": pct}
|
|
|
|
cursos_records = self.cursos.all(
|
|
fields=["CursoID", "CentroCurso", "Familia (from Familias)"]
|
|
)
|
|
curso_by_recordid = {}
|
|
curso_to_cc = {}
|
|
familia_lookup = {}
|
|
for r in cursos_records:
|
|
cid = r["fields"].get("CursoID")
|
|
if cid is None:
|
|
continue
|
|
cursoid_text = str(int(cid))
|
|
curso_by_recordid[r["id"]] = cursoid_text
|
|
curso_to_cc[cursoid_text] = r["fields"].get("CentroCurso", [])
|
|
familia = r["fields"].get("Familia (from Familias)")
|
|
familia_lookup[cursoid_text] = familia[0] if isinstance(familia, list) and familia else (familia or "Sin familia")
|
|
|
|
ppl_lookup = {}
|
|
for cursoid_text, cc_ids in curso_to_cc.items():
|
|
total_ppl = sum(
|
|
cc_data[cc_id]["ppl"] * cc_data[cc_id]["pct"]
|
|
for cc_id in cc_ids if cc_id in cc_data
|
|
)
|
|
ppl_lookup[cursoid_text] = round(total_ppl, 2)
|
|
|
|
cap_formula = f"AND({{Mes}}='{mes_nombre}',{{Año}}='{anio_str}')"
|
|
cursomes_records = self.cursomes.all(formula=cap_formula, fields=["CursoID", "Caping Admitido"])
|
|
cap_lookup = {}
|
|
for r in cursomes_records:
|
|
cap = int(r["fields"].get("Caping Admitido") or 0)
|
|
for curso_rec_id in r["fields"].get("CursoID", []):
|
|
cursoid_text = curso_by_recordid.get(curso_rec_id)
|
|
if cursoid_text:
|
|
cap_lookup[cursoid_text] = cap
|
|
|
|
return ppl_lookup, cap_lookup, familia_lookup
|
|
|
|
def get_leads_this_month_meta(self, campaign_name: str, as_of_date: str = None) -> tuple[int, list[str]]:
|
|
"""
|
|
Leads acumulados en el mes atribuidos a un curso vía Meta (leadform +
|
|
landing), hasta as_of_date (YYYY-MM-DD) inclusive, o hasta hoy si no se
|
|
indica (as_of_date lo usa backfill.py para reconstruir el estado
|
|
histórico del mes en una fecha pasada).
|
|
"""
|
|
course_num = extract_cursoid(campaign_name)
|
|
if not course_num:
|
|
return 0, []
|
|
ref = datetime.strptime(as_of_date, "%Y-%m-%d") if as_of_date else datetime.now()
|
|
mes_inicio = f"{ref.year}-{ref.month:02d}-01"
|
|
fin_clause = f",{{creado}}<'{(ref).strftime('%Y-%m-%d')}T23:59:59.999Z'" if as_of_date else ""
|
|
utm_clause = "OR(" + ",".join(f"{{attr_utm_source}}='{s}'" for s in META_UTM_SOURCES) + ")"
|
|
formula = (
|
|
f"AND({utm_clause},"
|
|
f"{{attr_cursoid}}='{course_num}',"
|
|
f"{{creado}}>='{mes_inicio}'{fin_clause})"
|
|
)
|
|
records = self.leads.all(formula=formula, fields=["attr_cursoid"])
|
|
ids = [r["id"] for r in records]
|
|
return len(ids), ids
|
|
|
|
def get_meta_leads_bulk(self, date_from: str, date_to: str) -> list[dict]:
|
|
"""
|
|
Todos los leads de Meta (leadform + landing) creados en [date_from, date_to]
|
|
(inclusive), sin restringir a un curso concreto — una sola llamada bulk
|
|
para poder agregar por día y/o por curso en el informe (igual patrón que
|
|
get_leads_by_campaign_on_date en leads-optimizer).
|
|
Devuelve [{"cursoid": str, "date": "YYYY-MM-DD", "utm_source": str}].
|
|
"""
|
|
next_day = (datetime.strptime(date_to, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d")
|
|
utm_clause = "OR(" + ",".join(f"{{attr_utm_source}}='{s}'" for s in META_UTM_SOURCES) + ")"
|
|
formula = f"AND({utm_clause},{{creado}}>='{date_from}',{{creado}}<'{next_day}')"
|
|
records = self.leads.all(formula=formula, fields=["attr_cursoid", "attr_utm_source", "creado"])
|
|
result = []
|
|
for r in records:
|
|
f = r["fields"]
|
|
cursoid = f.get("attr_cursoid")
|
|
if cursoid is None:
|
|
continue
|
|
cursoid_text = str(int(cursoid)) if isinstance(cursoid, (int, float)) else str(cursoid).strip()
|
|
creado = f.get("creado", "")
|
|
if not cursoid_text or not creado:
|
|
continue
|
|
result.append({
|
|
"cursoid": cursoid_text,
|
|
"date": creado[:10],
|
|
"utm_source": f.get("attr_utm_source", ""),
|
|
})
|
|
return result
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Meta Ads Campaigns (catálogo) #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def sync_campaigns_from_meta_ads(self, meta_campaigns: list[dict], ppl_lookup: dict = None) -> dict:
|
|
"""
|
|
Sincroniza el catálogo de campañas de Meta -> Airtable.
|
|
meta_campaigns: [{id, name, status}] (status: "ACTIVE"|"PAUSED"|...)
|
|
A diferencia de "Google Ads Campaigns" (donde 'CursoID Text' ya existe
|
|
como campo pre-poblado), aquí lo calculamos y escribimos nosotros mismos
|
|
al crear la tabla desde cero.
|
|
"""
|
|
STATUS_MAP = {"ACTIVE": "Activa"}
|
|
|
|
all_records = self.campaigns.all()
|
|
at_by_cid = {
|
|
str(r["fields"].get("CampaignID", "")).strip(): r
|
|
for r in all_records if r["fields"].get("CampaignID")
|
|
}
|
|
|
|
created, updated = [], []
|
|
to_create, to_update = [], []
|
|
|
|
for mc in meta_campaigns:
|
|
cid = mc["id"]
|
|
at_status = STATUS_MAP.get(mc["status"], "Pausada")
|
|
at_record = at_by_cid.get(cid)
|
|
cursoid_text = extract_cursoid(mc["name"]) or ""
|
|
ppl = round((ppl_lookup or {}).get(cursoid_text, 0), 2)
|
|
|
|
if at_record is None:
|
|
fields = {
|
|
"Campaign Name": mc["name"],
|
|
"CampaignID": cid,
|
|
"Status": at_status,
|
|
"CursoID Text": cursoid_text,
|
|
}
|
|
if ppl_lookup is not None:
|
|
fields["PPL"] = ppl
|
|
to_create.append(fields)
|
|
created.append({"name": mc["name"], "id": cid, "status": at_status})
|
|
else:
|
|
changes = {}
|
|
if at_record["fields"].get("Campaign Name") != mc["name"]:
|
|
changes["Campaign Name"] = mc["name"]
|
|
if at_record["fields"].get("Status") != at_status:
|
|
changes["Status"] = at_status
|
|
if at_record["fields"].get("CursoID Text") != cursoid_text:
|
|
changes["CursoID Text"] = cursoid_text
|
|
if ppl_lookup is not None and at_record["fields"].get("PPL") != ppl:
|
|
changes["PPL"] = ppl
|
|
if changes:
|
|
to_update.append((at_record["id"], changes))
|
|
updated.append({"name": mc["name"], "id": cid, "changes": changes})
|
|
|
|
for i in range(0, len(to_create), 10):
|
|
new_records = self.campaigns.batch_create(to_create[i:i + 10])
|
|
for r in new_records:
|
|
cid = str(r["fields"].get("CampaignID", "")).strip()
|
|
if cid:
|
|
at_by_cid[cid] = r
|
|
for i in range(0, len(to_update), 10):
|
|
batch = [{"id": rid, "fields": changes} for rid, changes in to_update[i:i + 10]]
|
|
self.campaigns.batch_update(batch)
|
|
|
|
return {"created": created, "updated": updated, "at_by_cid": at_by_cid}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# MetaCampaignMes (estado mensual) #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def sync_metacampaignmes(
|
|
self,
|
|
meta_campaigns: list[dict],
|
|
monthly_metrics: dict,
|
|
ppl_lookup: dict,
|
|
cap_lookup: dict,
|
|
at_by_cid: dict,
|
|
) -> dict:
|
|
"""
|
|
Crea/actualiza MetaCampaignMes para el mes/año en curso.
|
|
monthly_metrics: {campaign_id: {spend, leads, ...}} mes-a-la-fecha
|
|
(viene de meta.get_campaign_metrics(inicio_mes, ayer)).
|
|
"""
|
|
STATUS_MAP = {"ACTIVE": "Activa"}
|
|
now = datetime.now()
|
|
mes_num, anio_str = str(now.month), str(now.year)
|
|
|
|
formula = f"AND({{Mes}}='{mes_num}',{{Año}}='{anio_str}')"
|
|
existing = self.metacampaignmes.all(formula=formula)
|
|
mcm_by_at_cid = {}
|
|
for r in existing:
|
|
at_cids = r["fields"].get("CampaignID", [])
|
|
if at_cids:
|
|
mcm_by_at_cid[at_cids[0]] = r
|
|
|
|
to_create, to_update = [], []
|
|
|
|
for mc in meta_campaigns:
|
|
cid = mc["id"]
|
|
at_record = at_by_cid.get(cid)
|
|
if not at_record:
|
|
continue
|
|
|
|
at_cid = at_record["id"]
|
|
cursoid_text = extract_cursoid(mc["name"]) or ""
|
|
|
|
metrics = (monthly_metrics or {}).get(cid, {})
|
|
conv = round(metrics.get("leads", 0), 2)
|
|
cost = round(metrics.get("spend", 0), 2)
|
|
ppl = round((ppl_lookup or {}).get(cursoid_text, 0), 2)
|
|
cpa_max = round(ppl * 0.70, 2)
|
|
cap = int((cap_lookup or {}).get(cursoid_text, 0))
|
|
at_status = STATUS_MAP.get(mc["status"], "Pausada")
|
|
|
|
mcm_record = mcm_by_at_cid.get(at_cid)
|
|
|
|
if mcm_record is None:
|
|
to_create.append({
|
|
"CampaignID": [at_cid],
|
|
"Mes": mes_num,
|
|
"Año": anio_str,
|
|
"PPL": ppl,
|
|
"CPAMax": cpa_max,
|
|
"CapTotalMes": cap,
|
|
"CosteMes": cost,
|
|
"ConvMes": conv,
|
|
"Status": at_status,
|
|
})
|
|
else:
|
|
f = mcm_record["fields"]
|
|
changes = {}
|
|
if f.get("CosteMes") != cost:
|
|
changes["CosteMes"] = cost
|
|
if f.get("ConvMes") != conv:
|
|
changes["ConvMes"] = conv
|
|
if f.get("PPL") != ppl:
|
|
changes["PPL"] = ppl
|
|
if f.get("CPAMax") != cpa_max:
|
|
changes["CPAMax"] = cpa_max
|
|
if f.get("CapTotalMes") != cap:
|
|
changes["CapTotalMes"] = cap
|
|
if f.get("Status") != at_status:
|
|
changes["Status"] = at_status
|
|
if changes:
|
|
to_update.append((mcm_record["id"], changes))
|
|
|
|
for i in range(0, len(to_create), 10):
|
|
self.metacampaignmes.batch_create(to_create[i:i + 10], typecast=True)
|
|
for i in range(0, len(to_update), 10):
|
|
batch = [{"id": rid, "fields": f} for rid, f in to_update[i:i + 10]]
|
|
self.metacampaignmes.batch_update(batch, typecast=True)
|
|
|
|
return {"created": len(to_create), "updated": len(to_update)}
|
|
|
|
def get_active_metacampaignmes(self) -> list[dict]:
|
|
"""Lee MetaCampaignMes del mes/año en curso, resolviendo el Meta campaign ID."""
|
|
now = datetime.now()
|
|
mes_num, anio_str = str(now.month), str(now.year)
|
|
|
|
campaigns_records = self.campaigns.all(fields=["CampaignID"])
|
|
at_id_to_cid, cid_to_at_id = {}, {}
|
|
for r in campaigns_records:
|
|
cid = str(r["fields"].get("CampaignID", "")).strip()
|
|
if cid:
|
|
at_id_to_cid[r["id"]] = cid
|
|
cid_to_at_id[cid] = r["id"]
|
|
|
|
formula = f"AND({{Mes}}='{mes_num}',{{Año}}='{anio_str}')"
|
|
records = self.metacampaignmes.all(formula=formula)
|
|
|
|
result = []
|
|
for r in records:
|
|
f = r["fields"]
|
|
at_cids = f.get("CampaignID", [])
|
|
cid = at_id_to_cid.get(at_cids[0], "") if at_cids else ""
|
|
if not cid:
|
|
continue
|
|
result.append({
|
|
"airtable_id": r["id"],
|
|
"campaign_at_id": cid_to_at_id.get(cid, ""),
|
|
"meta_campaign_id": cid,
|
|
"ppl": float(f.get("PPL") or 0),
|
|
"capping_mensual": int(f.get("CapTotalMes") or 0),
|
|
"cpa_maximo": float(f.get("CPAMax") or 0),
|
|
"conv_mes": float(f.get("ConvMes") or 0),
|
|
})
|
|
return result
|
|
|
|
def batch_update_status(self, updates: list[tuple[str, str, str]]) -> None:
|
|
"""updates: [(mcm_record_id, campaign_at_id, meta_status)], meta_status: 'ACTIVE'|'PAUSED'."""
|
|
STATUS_MAP = {"ACTIVE": "Activa"}
|
|
mcm_batch, cat_batch = [], []
|
|
for mcm_id, cat_id, meta_status in updates:
|
|
at_status = STATUS_MAP.get(meta_status, "Pausada")
|
|
if mcm_id:
|
|
mcm_batch.append({"id": mcm_id, "fields": {"Status": at_status}})
|
|
if cat_id:
|
|
cat_batch.append({"id": cat_id, "fields": {"Status": at_status}})
|
|
for i in range(0, len(mcm_batch), 10):
|
|
self.metacampaignmes.batch_update(mcm_batch[i:i + 10])
|
|
for i in range(0, len(cat_batch), 10):
|
|
self.campaigns.batch_update(cat_batch[i:i + 10])
|
|
|
|
def batch_update_metacampaignmes_advice(self, updates: list[tuple[str, str, str, str]]) -> None:
|
|
"""updates: [(mcm_record_id, consejo, criticidad, log_text)]."""
|
|
batch = [
|
|
{"id": rid, "fields": {"Consejo": consejo, "Criticidad": criticidad, "Log": log_text}}
|
|
for rid, consejo, criticidad, log_text in updates
|
|
]
|
|
for i in range(0, len(batch), 10):
|
|
self.metacampaignmes.batch_update(batch[i:i + 10], typecast=True)
|
|
|
|
def batch_update_metacampaignmes_final_leads(self, updates: list[tuple[str, int]]) -> None:
|
|
"""updates: [(mcm_record_id, leads_lake_count)] -> ConvLeadsLakeMesFinal."""
|
|
batch = [{"id": rid, "fields": {"ConvLeadsLakeMesFinal": leads}} for rid, leads in updates]
|
|
for i in range(0, len(batch), 10):
|
|
self.metacampaignmes.batch_update(batch[i:i + 10], typecast=True)
|