from google.ads.googleads.client import GoogleAdsClient as GAdsClient from google.ads.googleads.errors import GoogleAdsException from datetime import datetime import config class GoogleAdsClient: def __init__(self): self.client = GAdsClient.load_from_dict({ "developer_token": config.GOOGLE_ADS_DEVELOPER_TOKEN, "client_id": config.GOOGLE_ADS_CLIENT_ID, "client_secret": config.GOOGLE_ADS_CLIENT_SECRET, "refresh_token": config.GOOGLE_ADS_REFRESH_TOKEN, "login_customer_id": config.GOOGLE_ADS_LOGIN_CUSTOMER_ID, "use_proto_plus": True, }) self.customer_id = config.GOOGLE_ADS_LOGIN_CUSTOMER_ID def get_all_campaigns(self) -> list[dict]: """Obtiene todas las campañas no eliminadas de la cuenta.""" ga_service = self.client.get_service("GoogleAdsService") query = """ SELECT campaign.id, campaign.name, campaign.status FROM campaign WHERE campaign.status != 'REMOVED' ORDER BY campaign.name """ try: response = ga_service.search(customer_id=self.customer_id, query=query) return [ { "id": str(row.campaign.id), "name": row.campaign.name, "status": row.campaign.status.name, } for row in response ] except GoogleAdsException as e: print(f" ❌ Error obteniendo campañas de Google Ads: {e}") return [] def get_monthly_metrics_all(self) -> dict: """ Devuelve métricas del mes actual para TODAS las campañas en una sola query. Retorna dict {campaign_id: {conversions, cost}}. """ ga_service = self.client.get_service("GoogleAdsService") query = """ SELECT campaign.id, metrics.conversions, metrics.cost_micros FROM campaign WHERE campaign.status != 'REMOVED' AND segments.date DURING THIS_MONTH """ result = {} try: response = ga_service.search(customer_id=self.customer_id, query=query) for row in response: cid = str(row.campaign.id) result[cid] = { "conversions": row.metrics.conversions, "cost": row.metrics.cost_micros / 1_000_000, } except GoogleAdsException as e: print(f" ❌ Error obteniendo métricas mensuales: {e}") return result def get_yesterday_metrics_all(self) -> dict: """ Devuelve métricas del día anterior para TODAS las campañas en una sola query. Retorna dict {campaign_id: {conversions, cost}}. """ from datetime import timedelta yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") ga_service = self.client.get_service("GoogleAdsService") query = f""" SELECT campaign.id, metrics.conversions, metrics.cost_micros FROM campaign WHERE campaign.status != 'REMOVED' AND segments.date = '{yesterday}' """ result = {} try: response = ga_service.search(customer_id=self.customer_id, query=query) for row in response: cid = str(row.campaign.id) result[cid] = { "conversions": row.metrics.conversions, "cost": row.metrics.cost_micros / 1_000_000, } except GoogleAdsException as e: print(f" ❌ Error obteniendo métricas de hoy: {e}") return result def get_daily_conversions_for_month(self, year: int, month: int) -> dict: """ Devuelve conversiones diarias de todas las campañas para un mes completo. Retorna dict {date_str ('YYYY-MM-DD'): {campaign_id: conversions}}. """ start = f"{year}-{month:02d}-01" # último día del mes if month == 12: end = f"{year + 1}-01-01" else: end = f"{year}-{month + 1:02d}-01" ga_service = self.client.get_service("GoogleAdsService") query = f""" SELECT campaign.id, segments.date, metrics.conversions FROM campaign WHERE campaign.status != 'REMOVED' AND segments.date >= '{start}' AND segments.date < '{end}' """ result: dict = {} try: response = ga_service.search(customer_id=self.customer_id, query=query) for row in response: date_str = row.segments.date cid = str(row.campaign.id) result.setdefault(date_str, {})[cid] = row.metrics.conversions except GoogleAdsException as e: print(f" ❌ Error obteniendo métricas mensuales: {e}") return result def get_campaign_metrics(self, campaign_id: str) -> dict: """Métricas del mes en curso para una campaña concreta (acumulado mensual).""" ga_service = self.client.get_service("GoogleAdsService") query = f""" SELECT campaign.id, campaign.name, campaign.status, campaign_budget.amount_micros, campaign_budget.resource_name, metrics.cost_micros, metrics.conversions, metrics.clicks, metrics.impressions FROM campaign WHERE campaign.id = {campaign_id} AND segments.date DURING THIS_MONTH """ try: response = ga_service.search(customer_id=self.customer_id, query=query) cost = conversions = clicks = impressions = 0 meta = {} for row in response: m = row.metrics cost += m.cost_micros / 1_000_000 conversions += m.conversions clicks += m.clicks impressions += m.impressions if not meta: meta = { "name": row.campaign.name, "status": row.campaign.status.name, "budget_daily": row.campaign_budget.amount_micros / 1_000_000, "budget_resource_name": row.campaign_budget.resource_name, } if not meta: return {} ctr = round(clicks / impressions * 100, 2) if impressions > 0 else 0.0 return { "campaign_id": campaign_id, "name": meta["name"], "status": meta["status"], "budget_daily": meta["budget_daily"], "budget_resource_name": meta["budget_resource_name"], "cost": round(cost, 2), "conversions": conversions, "clicks": clicks, "impressions": impressions, "ctr": ctr, } except GoogleAdsException as e: print(f" ❌ Error Google Ads para campaña {campaign_id}: {e}") return {} return {} def set_campaign_budget(self, budget_resource_name: str, new_daily_budget: float): """Ajusta el presupuesto diario de una campaña.""" if config.DRY_RUN: print(f" [DRY RUN] Nuevo presupuesto diario → {new_daily_budget:.2f}€") return True try: budget_service = self.client.get_service("CampaignBudgetService") campaign_budget = self.client.get_type("CampaignBudget") campaign_budget.resource_name = budget_resource_name campaign_budget.amount_micros = int(new_daily_budget * 1_000_000) operation = self.client.get_type("CampaignBudgetOperation") operation.update = campaign_budget operation.update_mask.paths.append("amount_micros") budget_service.mutate_campaign_budgets( customer_id=self.customer_id, operations=[operation] ) return True except GoogleAdsException as e: print(f" ❌ Error ajustando presupuesto: {e}") return False def pause_campaign(self, campaign_id: str): """Pausa una campaña.""" if config.DRY_RUN: print(f" [DRY RUN] Pausar campaña {campaign_id}") return True try: campaign_service = self.client.get_service("CampaignService") campaign = self.client.get_type("Campaign") campaign.resource_name = campaign_service.campaign_path( self.customer_id, campaign_id ) campaign.status = self.client.enums.CampaignStatusEnum.PAUSED operation = self.client.get_type("CampaignOperation") operation.update = campaign operation.update_mask.paths.append("status") campaign_service.mutate_campaigns( customer_id=self.customer_id, operations=[operation] ) return True except GoogleAdsException as e: print(f" ❌ Error pausando campaña: {e}") return False