from pyairtable import Api from datetime import datetime, timedelta import json import re 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), "conv_leads_lake_mes": int(f.get("ConvLeadsLakeMes") or 0), "metricas_diarias": f.get("MetricasDiarias") or "{}", }) 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, campaign_name: str = "") -> tuple[int, list[str]]: """ Leads del mes actual para una campaña de Google Ads. Cubre tres vías de atribución: 1. GACampaignID / GoogleCampaignID (leads web normales) 2. gad_campaignid en attr_referer (UTM web) 3. attr_referer = campaign_name con UserAgent Google-Ads-Notifications (Lead Form) """ now = datetime.now() mes_inicio = f"{now.year}-{now.month:02d}-01" leadform_clause = ( f"AND({{attr_referer}}='{campaign_name}',{{UserAgent del visitante}}='Google-Ads-Notifications')" if campaign_name else "FALSE()" ) formula = ( f"AND(" f"OR(" f"{{GACampaignID}}='{campaign_id}'," f"FIND(',{campaign_id},',',' & {{GoogleCampaignID}} & ',')," f"FIND('gad_campaignid={campaign_id}',{{attr_referer}})," f"{leadform_clause}" f")," 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 get_leads_by_campaign_on_date(self, date_str: str) -> dict: """ Devuelve {google_campaign_id: count} para todos los leads de un día concreto. Una sola llamada bulk para todos los campañas — más eficiente que una por campaña. date_str: 'YYYY-MM-DD' """ next_day = (datetime.strptime(date_str, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d") formula = f"AND({{creado}}>='{date_str}',{{creado}}<'{next_day}')" records = self.leads.all( formula=formula, fields=["GACampaignID", "GoogleCampaignID", "attr_referer"], ) counts: dict[str, int] = {} for r in records: f = r["fields"] cid = str(f.get("GACampaignID") or "").strip() if not cid: cid = str(f.get("GoogleCampaignID") or "").strip() if not cid: m = re.search(r"gad_campaignid=(\d+)", f.get("attr_referer", "")) if m: cid = m.group(1) if cid: counts[cid] = counts.get(cid, 0) + 1 return counts 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_final_leads(self, gcm_records: list[dict]) -> None: """ Actualiza ConvLeadsLakeMesFinal y ConvLeadsLakeMesGrupo en GACampaignMes. gcm_records: lista de dicts con: - 'airtable_id' - 'conv_leads_lake_mes' → valor para ConvLeadsLakeMesFinal - 'conv_leads_lake_mes_grupo' → valor para ConvLeadsLakeMesGrupo (igual al anterior salvo en PMX con Search companion: usa leads del Search) """ batch = [ {"id": r["airtable_id"], "fields": { "ConvLeadsLakeMesFinal": r["conv_leads_lake_mes"], "ConvLeadsLakeMesGrupo": r["conv_leads_lake_mes_grupo"], }} for r in gcm_records ] for i in range(0, len(batch), 10): self.gacampaignmes.batch_update(batch[i:i+10]) def get_metricas_diarias_prev_month(self) -> dict: """ Devuelve {google_campaign_id: {metricas, coste_mes, conv_mes}} del mes anterior. Usado en el reporter de Slack cuando hay cambio de mes. """ now = datetime.now() prev_month = now.month - 1 if now.month > 1 else 12 prev_year = now.year if now.month > 1 else now.year - 1 formula = f"AND({{Mes}}='{prev_month}',{{Año}}='{prev_year}')" records = self.gacampaignmes.all( formula=formula, fields=["CampaignID", "MetricasDiarias", "CosteMes", "ConvMes", "Campaign Name (from CampaignID)"], ) campaigns_records = self.campaigns.all(fields=["CampaignID"]) at_id_to_gid = {r["id"]: str(r["fields"].get("CampaignID", "")).strip() for r in campaigns_records} result = {} for r in records: at_cids = r["fields"].get("CampaignID", []) metricas = r["fields"].get("MetricasDiarias") or "{}" coste = float(r["fields"].get("CosteMes") or 0) conv = float(r["fields"].get("ConvMes") or 0) name_list = r["fields"].get("Campaign Name (from CampaignID)", []) name = name_list[0] if name_list else "" if at_cids: gid = at_id_to_gid.get(at_cids[0], "") if gid: result[gid] = { "metricas": metricas, "coste_mes": coste, "conv_mes": conv, "nombre": name, } return result def get_gcm_id_map_for_month(self, mes: int, anio: int) -> dict: """ Devuelve {google_campaign_id: gcm_record_id} para el mes/año indicado. Útil para escribir MetricasDiarias en el mes correcto cuando hay cambio de mes. """ formula = f"AND({{Mes}}='{mes}',{{Año}}='{anio}')" records = self.gacampaignmes.all(formula=formula, fields=["CampaignID"]) campaigns_records = self.campaigns.all(fields=["CampaignID"]) at_id_to_gid = {r["id"]: str(r["fields"].get("CampaignID", "")).strip() for r in campaigns_records} result = {} for r in records: at_cids = r["fields"].get("CampaignID", []) if at_cids: gid = at_id_to_gid.get(at_cids[0], "") if gid: result[gid] = r["id"] return result def batch_update_metricas_diarias(self, updates: list[dict]) -> None: """ Actualiza MetricasDiarias en GACampaignMes. updates: lista de {airtable_id, metricas_json (str)} """ batch = [ {"id": u["airtable_id"], "fields": {"MetricasDiarias": u["metricas_json"]}} for u in updates ] for i in range(0, len(batch), 10): self.gacampaignmes.batch_update(batch[i:i+10]) def batch_update_gacampaignmes_advice(self, updates: list[tuple]) -> None: """ Actualiza en lote los campos 'Consejo', 'Criticidad' y 'Log' de GACampaignMes. updates: lista de (gcm_record_id, consejo_text, criticidad, log_text) criticidad: 'Crítico' | 'Peligro' | 'Mantener' """ for i in range(0, len(updates), 10): batch = [ {"id": rid, "fields": {"Consejo": consejo, "Criticidad": criticidad, "Log": log}} for rid, consejo, criticidad, log 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