leads-optimizer/google_ads_client.py

from google.ads.googleads.client import GoogleAdsClient as GAdsClient
from google.ads.googleads.errors import GoogleAdsException
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_campaign_metrics(self, campaign_id: str) -> dict:
        """Métricas del mes en curso para una campaña concreta."""
        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,
                metrics.ctr
            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)
            for row in response:
                m = row.metrics
                return {
                    "campaign_id": campaign_id,
                    "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,
                    "cost": m.cost_micros / 1_000_000,
                    "conversions": m.conversions,
                    "clicks": m.clicks,
                    "impressions": m.impressions,
                    "ctr": round(m.ctr * 100, 2),
                }
        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