44 lines
1.6 KiB
Python
44 lines
1.6 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)
|
|
|
|
def get_active_campaigns(self) -> list[dict]:
|
|
"""Lee todas las campañas activas desde 'Google Ads Campaigns'."""
|
|
records = self.campaigns.all(formula="{Activa}=TRUE()")
|
|
result = []
|
|
for r in records:
|
|
f = r["fields"]
|
|
result.append({
|
|
"airtable_id": r["id"],
|
|
"curso": f.get("Campaign Name", "Sin nombre"),
|
|
"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("CPAMaximo", 0)),
|
|
})
|
|
return [c for c in result if c["google_campaign_id"]] # descartar sin ID
|
|
|
|
def get_leads_this_month(self, google_campaign_id: str) -> int:
|
|
"""
|
|
Cuenta todos los leads del mes actual para una campaña.
|
|
Todos los leads en Airtable están validados, sin filtro de estado.
|
|
"""
|
|
now = datetime.now()
|
|
mes_inicio = f"{now.year}-{now.month:02d}-01"
|
|
|
|
formula = (
|
|
f"AND("
|
|
f"{{GoogleCampaignID}}='{google_campaign_id}',"
|
|
f"{{Fecha}}>='{mes_inicio}'"
|
|
f")"
|
|
)
|
|
records = self.leads.all(formula=formula)
|
|
return len(records)
|