410 lines
17 KiB
Python
410 lines
17 KiB
Python
from pyairtable import Api
|
||
from datetime import datetime
|
||
import config
|
||
|
||
|
||
class AirtableClient:
|
||
def __init__(self):
|
||
self.api = Api(config.AIRTABLE_TOKEN)
|
||
self.leads = self.api.table(config.AIRTABLE_BASE_ID, config.LEADS_TABLE)
|
||
self.campaigns = self.api.table(config.AIRTABLE_BASE_ID, config.CAMPAIGNS_TABLE)
|
||
self.cursos = self.api.table(config.AIRTABLE_BASE_ID, "Cursos")
|
||
self.centrocurso = self.api.table(config.AIRTABLE_BASE_ID, "CentroCurso")
|
||
self.cursomes = self.api.table(config.AIRTABLE_BASE_ID, "CursoMes")
|
||
self.gacampaignmes = self.api.table(config.AIRTABLE_BASE_ID, "GACampaignMes")
|
||
|
||
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 build_campaign_lookups(self) -> tuple[dict, dict]:
|
||
"""
|
||
Construye en 3 llamadas bulk:
|
||
ppl_lookup: {cursoid_text: ppl_ponderado}
|
||
PPL = Σ (PPL_centrocurso × % invalidación_centro)
|
||
solo registros CentroCurso con Estado ROI = 'ABIERTO'
|
||
cap_lookup: {cursoid_text: caping_admitido}
|
||
desde CursoMes para el mes/año en curso; 0 si no existe
|
||
"""
|
||
now = datetime.now()
|
||
mes_nombre = self.MESES_ES[now.month]
|
||
anio_str = str(now.year)
|
||
|
||
# 1. CentroCurso ABIERTO: {record_id: {ppl, pct_invalidacion}}
|
||
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)
|
||
if isinstance(pct_raw, list):
|
||
pct = float(pct_raw[0]) if pct_raw else 0.0
|
||
else:
|
||
pct = float(pct_raw or 0)
|
||
cc_data[r["id"]] = {"ppl": ppl_val, "pct": pct}
|
||
|
||
# 2. Cursos: {record_id: cursoid_text} + {cursoid_text: [cc_ids]}
|
||
cursos_records = self.cursos.all(fields=["CursoID", "CentroCurso"])
|
||
curso_by_recordid = {}
|
||
curso_to_cc = {}
|
||
for r in cursos_records:
|
||
cid = r["fields"].get("CursoID")
|
||
cc_ids = r["fields"].get("CentroCurso", [])
|
||
if cid is not None:
|
||
cursoid_text = str(int(cid))
|
||
curso_by_recordid[r["id"]] = cursoid_text
|
||
curso_to_cc[cursoid_text] = cc_ids
|
||
|
||
# 3. PPL ponderado por % invalidación del centro (decimal: 0.85 = 85%)
|
||
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)
|
||
|
||
# 4. CursoMes del mes/año actual → cap_lookup
|
||
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
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Google Ads Campaigns (catálogo) #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def sync_campaigns_from_google_ads(
|
||
self,
|
||
google_campaigns: list[dict],
|
||
monthly_metrics: dict = None,
|
||
ppl_lookup: dict = None,
|
||
cap_lookup: dict = None,
|
||
) -> dict:
|
||
"""
|
||
Sincroniza el catálogo de campañas Google Ads → Airtable.
|
||
Devuelve {created, updated, at_by_gid} donde at_by_gid es el
|
||
mapping {google_campaign_id: airtable_record} para reutilizar.
|
||
"""
|
||
STATUS_MAP = {"ENABLED": "Activa", "PAUSED": "Pausada"}
|
||
|
||
all_records = self.campaigns.all()
|
||
at_by_gid = {
|
||
str(r["fields"].get("CampaignID", "")).strip(): r
|
||
for r in all_records
|
||
if r["fields"].get("CampaignID")
|
||
}
|
||
|
||
created = []
|
||
updated = []
|
||
to_create = []
|
||
to_update = []
|
||
|
||
for gc in google_campaigns:
|
||
gid = gc["id"]
|
||
at_status = STATUS_MAP.get(gc["status"], "Pausada")
|
||
at_record = at_by_gid.get(gid)
|
||
|
||
# "Google Ads Campaigns" es catálogo: solo nombre, status y PPL
|
||
cursoid_text = str((at_record or {}).get("fields", {}).get("CursoID Text", "")).strip() if at_record else ""
|
||
ppl = round((ppl_lookup or {}).get(cursoid_text, 0), 2)
|
||
|
||
if at_record is None:
|
||
fields = {"Campaign Name": gc["name"], "CampaignID": gid, "Status": at_status}
|
||
if ppl_lookup is not None:
|
||
fields["PPL"] = ppl
|
||
to_create.append(fields)
|
||
created.append({"name": gc["name"], "id": gid, "status": at_status})
|
||
else:
|
||
changes = {}
|
||
change_detail = {}
|
||
if at_record["fields"].get("Campaign Name") != gc["name"]:
|
||
changes["Campaign Name"] = gc["name"]
|
||
change_detail["nombre"] = {
|
||
"antes": at_record["fields"].get("Campaign Name"),
|
||
"ahora": gc["name"],
|
||
}
|
||
if at_record["fields"].get("Status") != at_status:
|
||
changes["Status"] = at_status
|
||
change_detail["status"] = {
|
||
"antes": at_record["fields"].get("Status"),
|
||
"ahora": at_status,
|
||
}
|
||
if ppl_lookup is not None:
|
||
if at_record["fields"].get("PPL") != ppl:
|
||
changes["PPL"] = ppl
|
||
if changes:
|
||
to_update.append((at_record["id"], changes, gc["name"]))
|
||
if change_detail:
|
||
updated.append({"name": gc["name"], "id": gid, "changes": change_detail})
|
||
|
||
for i in range(0, len(to_create), 10):
|
||
self.campaigns.batch_create(to_create[i:i+10])
|
||
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_gid": at_by_gid}
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# GACampaignMes (estado mensual de campañas con actividad) #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def sync_gacampaignmes(
|
||
self,
|
||
google_campaigns: list[dict],
|
||
monthly_metrics: dict,
|
||
ppl_lookup: dict,
|
||
cap_lookup: dict,
|
||
at_by_gid: dict,
|
||
) -> dict:
|
||
"""
|
||
Crea/actualiza registros en GACampaignMes para el mes/año en curso.
|
||
Se incluyen campañas con actividad durante el mes:
|
||
- Estado ENABLED en Google Ads, o
|
||
- Coste > 0 en monthly_metrics (activas en algún momento del mes)
|
||
"""
|
||
STATUS_MAP = {"ENABLED": "Activa", "PAUSED": "Pausada"}
|
||
now = datetime.now()
|
||
mes_num = str(now.month) # valor numérico "1"…"12"
|
||
anio_str = str(now.year)
|
||
|
||
# GIDs con actividad este mes
|
||
active_gids = {
|
||
gc["id"]
|
||
for gc in google_campaigns
|
||
if gc["status"] == "ENABLED"
|
||
or (monthly_metrics or {}).get(gc["id"], {}).get("cost", 0) > 0
|
||
}
|
||
|
||
# Registros GACampaignMes ya existentes este mes: {at_campaign_id: gcm_record}
|
||
formula = f"AND({{Mes}}='{mes_num}',{{Año}}='{anio_str}')"
|
||
existing = self.gacampaignmes.all(formula=formula)
|
||
gcm_by_at_cid = {}
|
||
for r in existing:
|
||
at_cids = r["fields"].get("CampaignID", [])
|
||
if at_cids:
|
||
gcm_by_at_cid[at_cids[0]] = r
|
||
|
||
to_create = []
|
||
to_update = []
|
||
|
||
for gc in google_campaigns:
|
||
gid = gc["id"]
|
||
if gid not in active_gids:
|
||
continue
|
||
at_record = at_by_gid.get(gid)
|
||
if not at_record:
|
||
continue
|
||
|
||
at_cid = at_record["id"]
|
||
cursoid_text = str(at_record["fields"].get("CursoID Text", "")).strip()
|
||
|
||
metrics = (monthly_metrics or {}).get(gid, {})
|
||
conv = round(metrics.get("conversions", 0), 2)
|
||
cost = round(metrics.get("cost", 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(gc["status"], "Pausada")
|
||
|
||
gcm_record = gcm_by_at_cid.get(at_cid)
|
||
|
||
if gcm_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 = gcm_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((gcm_record["id"], changes))
|
||
|
||
for i in range(0, len(to_create), 10):
|
||
self.gacampaignmes.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.gacampaignmes.batch_update(batch, typecast=True)
|
||
|
||
return {"created": len(to_create), "updated": len(to_update)}
|
||
|
||
def get_active_gacampaignmes(self) -> list[dict]:
|
||
"""
|
||
Lee los registros GACampaignMes del mes/año en curso.
|
||
Resuelve el Google Ads campaign ID a partir del registro vinculado
|
||
en 'Google Ads Campaigns'.
|
||
"""
|
||
now = datetime.now()
|
||
mes_num = str(now.month)
|
||
anio_str = str(now.year)
|
||
|
||
# Mapping at_record_id → google_campaign_id y gid → at_record_id
|
||
campaigns_records = self.campaigns.all(fields=["CampaignID"])
|
||
at_id_to_gid = {}
|
||
gid_to_at_id = {}
|
||
for r in campaigns_records:
|
||
gid = str(r["fields"].get("CampaignID", "")).strip()
|
||
if gid:
|
||
at_id_to_gid[r["id"]] = gid
|
||
gid_to_at_id[gid] = r["id"]
|
||
|
||
formula = f"AND({{Mes}}='{mes_num}',{{Año}}='{anio_str}')"
|
||
records = self.gacampaignmes.all(formula=formula)
|
||
|
||
result = []
|
||
for r in records:
|
||
f = r["fields"]
|
||
at_cids = f.get("CampaignID", [])
|
||
gid = at_id_to_gid.get(at_cids[0], "") if at_cids else ""
|
||
if not gid:
|
||
continue
|
||
|
||
name_list = f.get("Campaign Name (from CampaignID)", [])
|
||
name = name_list[0] if name_list else "Sin nombre"
|
||
|
||
result.append({
|
||
"airtable_id": r["id"], # GACampaignMes record ID
|
||
"campaign_at_id": gid_to_at_id.get(gid, ""), # Google Ads Campaigns record ID
|
||
"curso": name,
|
||
"cursoid_text": "",
|
||
"google_campaign_id": gid,
|
||
"ppl": float(f.get("PPL") or 0),
|
||
"capping_mensual": int(f.get("CapTotalMes") or 0),
|
||
"cpa_maximo": float(f.get("CPAMax") or 0),
|
||
"margen_objetivo": float(f.get("MargenObjetivo") or 0),
|
||
})
|
||
return result
|
||
|
||
def batch_update_status(self, updates: list[tuple[str, str, str]]) -> None:
|
||
"""
|
||
Actualiza en lote el campo Status en GACampaignMes y Google Ads Campaigns.
|
||
updates: lista de (gcm_record_id, campaign_at_id, google_status)
|
||
google_status: 'ENABLED' | 'PAUSED'
|
||
"""
|
||
STATUS_MAP = {"ENABLED": "Activa", "PAUSED": "Pausada"}
|
||
|
||
gcm_batch = []
|
||
cat_batch = []
|
||
for gcm_id, cat_id, google_status in updates:
|
||
at_status = STATUS_MAP.get(google_status, "Pausada")
|
||
if gcm_id:
|
||
gcm_batch.append({"id": gcm_id, "fields": {"Status": at_status}})
|
||
if cat_id:
|
||
cat_batch.append({"id": cat_id, "fields": {"Status": at_status}})
|
||
|
||
for i in range(0, len(gcm_batch), 10):
|
||
self.gacampaignmes.batch_update(gcm_batch[i:i+10])
|
||
for i in range(0, len(cat_batch), 10):
|
||
self.campaigns.batch_update(cat_batch[i:i+10])
|
||
|
||
def get_leads_this_month_gads(self, campaign_id: str) -> tuple[int, list[str]]:
|
||
"""
|
||
Leads del mes actual para una campaña de Google Ads.
|
||
Filtra por GoogleCampaignID (extraído de attr_referer),
|
||
attr_utm_source pmx/google y attr_utm_medium paid_search.
|
||
"""
|
||
now = datetime.now()
|
||
mes_inicio = f"{now.year}-{now.month:02d}-01"
|
||
formula = (
|
||
f"AND("
|
||
f"{{GoogleCampaignID}}='{campaign_id}',"
|
||
f"{{attr_utm_medium}}='paid_search',"
|
||
f"{{creado}}>='{mes_inicio}'"
|
||
f")"
|
||
)
|
||
records = self.leads.all(formula=formula, fields=["airtable id"])
|
||
ids = [r["id"] for r in records]
|
||
return len(ids), ids
|
||
|
||
def update_gacampaignmes_leads_lake(self, gcm_record_id: str, lead_ids: list[str]) -> None:
|
||
"""Actualiza el campo 'Leads Lake' de un registro GACampaignMes."""
|
||
self.gacampaignmes.update(gcm_record_id, {"Leads Lake": lead_ids})
|
||
|
||
def batch_update_gacampaignmes_advice(self, updates: list[tuple[str, str, str]]) -> None:
|
||
"""
|
||
Actualiza en lote los campos 'Consejo' y 'Criticidad' de GACampaignMes.
|
||
updates: lista de (gcm_record_id, consejo_text, criticidad)
|
||
criticidad: 'Crítico' | 'Peligro' | 'Mantener'
|
||
"""
|
||
for i in range(0, len(updates), 10):
|
||
batch = [
|
||
{"id": rid, "fields": {"Consejo": consejo, "Criticidad": criticidad}}
|
||
for rid, consejo, criticidad in updates[i:i+10]
|
||
]
|
||
if batch:
|
||
self.gacampaignmes.batch_update(batch, typecast=True)
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Métodos legacy (usados internamente / compatibilidad) #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def get_active_campaigns(self) -> list[dict]:
|
||
"""Lee todas las campañas activas desde 'Google Ads Campaigns'."""
|
||
records = self.campaigns.all(formula="{Status}='Activa'")
|
||
result = []
|
||
for r in records:
|
||
f = r["fields"]
|
||
result.append({
|
||
"airtable_id": r["id"],
|
||
"curso": f.get("Campaign Name", "Sin nombre"),
|
||
"cursoid_text": str(f.get("CursoID Text", "")).strip(),
|
||
"google_campaign_id": str(f.get("CampaignID", "")).strip(),
|
||
"ppl": float(f.get("PPL", 0)),
|
||
"capping_mensual": int(f.get("CapTotalMes", 0)),
|
||
"cpa_maximo": float(f.get("CPAMax", 0)),
|
||
"margen_objetivo": float(f.get("MargenObjetivo", 0)),
|
||
})
|
||
return [c for c in result if c["google_campaign_id"]]
|
||
|
||
def get_leads_this_month(self, cursoid_text: str) -> tuple[int, list[str]]:
|
||
"""Leads del mes actual para un curso (por attr_cursoid)."""
|
||
now = datetime.now()
|
||
mes_inicio = f"{now.year}-{now.month:02d}-01"
|
||
formula = (
|
||
f"AND("
|
||
f"{{attr_cursoid}}='{cursoid_text}',"
|
||
f"{{creado}}>='{mes_inicio}'"
|
||
f")"
|
||
)
|
||
records = self.leads.all(formula=formula, fields=["airtable id"])
|
||
ids = [r["id"] for r in records]
|
||
return len(ids), ids
|
||
|
||
def update_campaign_leads_lake(self, campaign_airtable_id: str, lead_ids: list[str]) -> None:
|
||
"""Actualiza el campo 'Leads Lake' de una campaña en Google Ads Campaigns."""
|
||
self.campaigns.update(campaign_airtable_id, {"Leads Lake": lead_ids})
|