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") 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 / 100) 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 # El campo % invalidación está en formato decimal (ej. 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 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, cap_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 y CapTotalMes calculados desde 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) cap = int((cap_lookup or {}).get(cursoid_text, 0)) 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 if cap_lookup is not None: fields["CapTotalMes"] = cap 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 cap_lookup is not None: if at_record["fields"].get("CapTotalMes") != cap: changes["CapTotalMes"] = cap 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})