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") def build_ppl_lookup(self) -> dict: """ Construye un dict {cursoid_text: ppl_sum} calculando el PPL de cada curso sumando los PPL de los registros CENTROCURSO con Estado ROI = 'ABIERTO'. Cadena: CursoID Text → Cursos.CursoID → Cursos.CentroCurso → CENTROCURSO.PPL Solo se usan en lotes (2 llamadas bulk), sin llamadas por campaña. """ # 1. Cargar CENTROCURSO ABIERTO: {record_id: ppl} cc_records = self.centrocurso.all( formula="{Estado ROI}='ABIERTO'", fields=["PPL"], ) cc_ppl = {r["id"]: float(r["fields"].get("PPL") or 0) for r in cc_records} # 2. Cargar Cursos: {cursoid_number: [cc_record_ids]} cursos_records = self.cursos.all(fields=["CursoID", "CentroCurso"]) curso_to_cc = {} for r in cursos_records: cid = r["fields"].get("CursoID") cc_ids = r["fields"].get("CentroCurso", []) if cid is not None: curso_to_cc[str(int(cid))] = cc_ids # 3. Para cada CursoID Text, sumar PPL de CENTROCURSO ABIERTO ppl_lookup = {} for cursoid_text, cc_ids in curso_to_cc.items(): total_ppl = sum(cc_ppl[cc_id] for cc_id in cc_ids if cc_id in cc_ppl) ppl_lookup[cursoid_text] = round(total_ppl, 2) return ppl_lookup 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"]] # descartar sin ID def sync_campaigns_from_google_ads(self, google_campaigns: list[dict], monthly_metrics: dict = None, ppl_lookup: dict = None) -> dict: """ Sincroniza campañas de Google Ads → Airtable. - Crea las campañas nuevas con el status real de Google Ads. - Actualiza nombre y/o status si han cambiado. Devuelve un resumen con listas de campañas creadas y actualizadas. """ 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 = [] # lista de dicts {name, id, status} updated = [] # lista de dicts {name, id, changes} to_create = [] to_update = [] # lista de (record_id, changes, name, gid, at_status) for gc in google_campaigns: gid = gc["id"] at_status = STATUS_MAP.get(gc["status"], "Pausada") at_record = at_by_gid.get(gid) metrics = (monthly_metrics or {}).get(gid, {}) conv = round(metrics.get("conversions", 0), 2) cost = round(metrics.get("cost", 0), 2) # PPL calculado desde CENTROCURSO (CursoID Text viene de Airtable) 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) cpa_max = round(ppl * 0.70, 2) if at_record is None: fields = {"Campaign Name": gc["name"], "CampaignID": gid, "Status": at_status} if monthly_metrics is not None: fields["ConvMes"] = conv fields["CosteMes"] = cost if ppl_lookup is not None: fields["PPL"] = ppl fields["CPAMax"] = cpa_max 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 monthly_metrics is not None: if at_record["fields"].get("ConvMes") != conv: changes["ConvMes"] = conv if at_record["fields"].get("CosteMes") != cost: changes["CosteMes"] = cost if ppl_lookup is not None: if at_record["fields"].get("PPL") != ppl: changes["PPL"] = ppl if at_record["fields"].get("CPAMax") != cpa_max: changes["CPAMax"] = cpa_max if changes: to_update.append((at_record["id"], changes, gc["name"])) if change_detail: updated.append({"name": gc["name"], "id": gid, "changes": change_detail}) # Crear en lotes de 10 for i in range(0, len(to_create), 10): self.campaigns.batch_create(to_create[i:i+10]) # Actualizar en lotes de 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} def get_leads_this_month(self, cursoid_text: str) -> tuple[int, list[str]]: """ Devuelve (count, [record_ids]) de leads del mes actual para un curso. Filtra por attr_cursoid (que coincide con CursoID Text de la campaña). """ 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 con los IDs de leads del mes.""" self.campaigns.update(campaign_airtable_id, {"Leads Lake": lead_ids})