From a5ce85ed328a1c0416efe9a7a2303ca9d5c6ff34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Manuel=20G=C3=B3mez?= Date: Tue, 21 Apr 2026 13:00:34 +0200 Subject: [PATCH] Add GACampaignMes sync and use it as optimization source - New GACampaignMes table tracks monthly state of campaigns with activity (ENABLED status or cost > 0 this month) - sync_gacampaignmes(): creates/updates records with PPL, CPAMax, CosteMes, ConvMes, CapTotalMes, Status for current month (numeric Mes: 1-12) - get_active_gacampaignmes(): reads GACampaignMes as source for optimization loop - get_leads_this_month_gads(): filters leads by GoogleCampaignID + utm_source (pmx/google) + utm_medium (paid_search) instead of attr_cursoid - update_gacampaignmes_leads_lake(): links leads to GACampaignMes Leads Lake field - batch_update_gacampaignmes_consejos(): saves agent advice to Consejo field - Google Ads Campaigns table simplified to catalog only (name, status, PPL) Co-Authored-By: Claude Sonnet 4.6 --- airtable_client.py | 286 +++++++++++++++++++++++++++++++++++---------- run.py | 26 ++++- 2 files changed, 249 insertions(+), 63 deletions(-) diff --git a/airtable_client.py b/airtable_client.py index cdbad4b..22e4d0e 100644 --- a/airtable_client.py +++ b/airtable_client.py @@ -11,6 +11,7 @@ class AirtableClient: 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", @@ -22,7 +23,7 @@ class AirtableClient: """ Construye en 3 llamadas bulk: ppl_lookup: {cursoid_text: ppl_ponderado} - PPL = Σ (PPL_centrocurso × % invalidación_centro / 100) + 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 @@ -58,8 +59,7 @@ class AirtableClient: 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%) + # 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( @@ -84,30 +84,21 @@ class AirtableClient: 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 + # ------------------------------------------------------------------ # + # 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: + 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. + 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"} @@ -118,36 +109,24 @@ class AirtableClient: if r["fields"].get("CampaignID") } - created = [] # lista de dicts {name, id, status} - updated = [] # lista de dicts {name, id, changes} + created = [] + updated = [] to_create = [] - to_update = [] # lista de (record_id, changes, name, gid, at_status) + 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) - 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) + # "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) - 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: @@ -165,43 +144,232 @@ class AirtableClient: "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} + return {"created": created, "updated": updated, "at_by_gid": at_by_gid} - def get_leads_this_month(self, cursoid_text: str) -> tuple[int, list[str]]: + # ------------------------------------------------------------------ # + # 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: """ - 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). + 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 + campaigns_records = self.campaigns.all(fields=["CampaignID"]) + at_id_to_gid = { + r["id"]: str(r["fields"].get("CampaignID", "")).strip() + for r in campaigns_records + if r["fields"].get("CampaignID") + } + + 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"], + "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 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"OR({{attr_utm_source}}='pmx',{{attr_utm_source}}='google')," + 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_consejos(self, updates: list[tuple[str, str]]) -> None: + """ + Actualiza en lote el campo 'Consejo' de varios registros GACampaignMes. + updates: lista de (gcm_record_id, consejo_text) + """ + for i in range(0, len(updates), 10): + batch = [ + {"id": rid, "fields": {"Consejo": consejo}} + for rid, consejo in updates[i:i+10] + if consejo + ] + if batch: + self.gacampaignmes.batch_update(batch) + + # ------------------------------------------------------------------ # + # 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}'," @@ -213,5 +381,5 @@ class AirtableClient: 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.""" + """Actualiza el campo 'Leads Lake' de una campaña en Google Ads Campaigns.""" self.campaigns.update(campaign_airtable_id, {"Leads Lake": lead_ids}) diff --git a/run.py b/run.py index 9be30a0..7cbb07f 100644 --- a/run.py +++ b/run.py @@ -104,10 +104,18 @@ def run(): print(f" ~ [{c['id']}] {c['name']} | {field}: '{val['antes']}' → '{val['ahora']}'") if not sync_result["created"] and not sync_result["updated"]: print(" ✓ Sin cambios en el catálogo.") + + # Sincronizar GACampaignMes (campañas con actividad este mes) + print(" Sincronizando GACampaignMes...") + gcm_result = at.sync_gacampaignmes( + google_campaigns, monthly_metrics, ppl_lookup, cap_lookup, + sync_result["at_by_gid"] + ) + print(f" ✓ GACampaignMes: {gcm_result['created']} nuevas, {gcm_result['updated']} actualizadas.") print() - campaigns = at.get_active_campaigns() - print(f"→ {len(campaigns)} campañas activas encontradas") + campaigns = at.get_active_gacampaignmes() + print(f"→ {len(campaigns)} campañas con actividad este mes") print("→ Analizando...\n") # === PRIMERA PASADA: recopilar datos de todas las campañas === @@ -117,8 +125,8 @@ def run(): for campaign in campaigns: cid = campaign["google_campaign_id"] - leads, lead_ids = at.get_leads_this_month(campaign["cursoid_text"]) - at.update_campaign_leads_lake(campaign["airtable_id"], lead_ids) + leads, lead_ids = at.get_leads_this_month_gads(cid) + at.update_gacampaignmes_leads_lake(campaign["airtable_id"], lead_ids) metrics = gads.get_campaign_metrics(cid) if not metrics: @@ -147,6 +155,7 @@ def run(): # === SEGUNDA PASADA: imprimir en orden + aplicar decisiones === resumen = [] + consejo_updates = [] # (gcm_record_id, consejo) para batch update final last_priority = -1 for item in collected: @@ -196,6 +205,9 @@ def run(): apply_decision(campaign, decision, metrics, gads) + if decision.get("consejo"): + consejo_updates.append((campaign["airtable_id"], decision["consejo"])) + resumen.append({ "curso": campaign["curso"], "urgencia": analysis["urgencia"], @@ -210,6 +222,12 @@ def run(): print() + # Guardar consejos en GACampaignMes + if consejo_updates: + print(f"→ Guardando {len(consejo_updates)} consejos en GACampaignMes...") + at.batch_update_gacampaignmes_consejos(consejo_updates) + print(" ✓ Consejos guardados.") + # Resumen final ordenado por prioridad (ya está ordenado) print(f"{'='*55}") print("RESUMEN FINAL")