"""Google Ads connector.
Syncs ad spend, leads, ROAS, and campaign data from the Google Ads API (GAQL).
Uses stream_tokens for pagination and upserts records to AdCampaign and AdMetric tables.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, date, timezone, timedelta
from typing import Any, Dict, List, Optional
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata -------------------------------------------------
register_connector(
"google_ads",
{
"service": "google_ads",
"name": "Google Ads",
"category": "advertising",
"description": "Ad spend, leads, ROAS, and campaign performance.",
"auth_type": "oauth2",
"auth_fields": [
"developer_token",
"client_id",
"client_secret",
"refresh_token",
"customer_id",
],
"capabilities": ["campaigns", "ad_group_ads", "keywords", "keyword_research", "metrics", "conversions"],
"rate_limit": "Quota-based; 500-1000 requests/min typical",
"docs_url": "https://developers.google.com/google-ads/api/docs/start",
},
)
class GoogleAdsConnector(OAuthConnector):
"""Google Ads integration — full sync with GAQL pagination and storage."""
_SERVICE = "google_ads"
_BASE_URL = "https://googleads.googleapis.com/v17"
# -- OAuth 2.0 ------------------------------------------------------------
OAUTH_AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/v2/auth"
OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token"
OAUTH_SCOPES = ["https://www.googleapis.com/auth/adwords"]
@property
def _oauth_client_id(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_GOOGLE_ADS_CLIENT_ID", "")
@property
def _oauth_client_secret(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_GOOGLE_ADS_CLIENT_SECRET", "")
def oauth_authorize_url(self, state: str) -> str:
"""Build the Google OAuth authorization URL."""
from urllib.parse import urlencode
params = {
"client_id": self._oauth_client_id,
"response_type": "code",
"scope": " ".join(self.OAUTH_SCOPES),
"redirect_uri": self._get_redirect_uri(),
"state": state,
"access_type": "offline",
"prompt": "consent",
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for access/refresh tokens via Google."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"Google Ads OAuth credentials not configured. "
"Set OAUTH_GOOGLE_ADS_CLIENT_ID and OAUTH_GOOGLE_ADS_CLIENT_SECRET."
)
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self._get_redirect_uri(),
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh expired access token via Google token endpoint."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("Google Ads OAuth credentials not configured.")
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", refresh_token),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
# -- connect / disconnect -----------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate Google Ads credentials by querying account info."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
raise ValueError("Google Ads customer_id and access_token are required")
response = self._retry(
_gaql_query,
query=(
"SELECT customer_descriptor.customer_id, "
"customer_descriptor.currency_code, "
"customer_descriptor.time_zone "
"FROM customer"
),
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"customer_id": customer_id},
)
return {
"status": "connected",
"service": "google_ads",
"customer_id": customer_id,
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
self._connected = False
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "google_ads"}
# -- sync ----------------------------------------------------------------
def sync(self, sync_keywords: bool = True) -> Dict[str, Any]:
"""Sync campaigns, metrics, and conversions from Google Ads."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
# Sync campaigns
campaigns = self._sync_campaigns()
total_records += campaigns
results["campaigns"] = campaigns
# Sync daily metrics
metrics = self._sync_metrics()
total_records += metrics
results["metrics"] = metrics
# Sync keywords
if sync_keywords:
keywords = self._sync_keywords()
total_records += keywords
results["keywords"] = keywords
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- Campaign sync with pagination ---------------------------------------
def _sync_campaigns(self) -> int:
"""Pull campaigns via GAQL with stream_token pagination, upsert to AdCampaign."""
from app.models import db, AdCampaign
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
logger.warning("Google Ads sync: missing credentials")
return 0
query = (
"SELECT "
"campaign.id, campaign.name, campaign.status, "
"campaign.advertising_channel_type, campaign.campaign_budget_amount_micros, "
"campaign.budget_campaign_period_view, "
"campaign.start_date, campaign.end_date, "
"campaign.status "
"FROM campaign"
)
all_rows: List[Dict[str, Any]] = self._gaql_paginate(
query=query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
count = 0
for row in all_rows:
fields = row.get("fields", {})
campaign_id = str(fields.get("campaign", {}).get("id", ""))
if not campaign_id:
continue
name = fields.get("campaign", {}).get("name", "")
status = fields.get("campaign", {}).get("status", "")
channel_type = fields.get("campaign", {}).get("advertising_channel_type", "")
# Parse budget (micros -> dollars)
budget_micros = fields.get("campaign", {}).get("campaign_budget_amount_micros", 0)
budget = None
if budget_micros:
try:
budget = float(budget_micros) / 1_000_000
except (ValueError, TypeError):
pass
# Parse dates (YYYYMMDD -> datetime)
start_date = fields.get("campaign", {}).get("start_date")
end_date = fields.get("campaign", {}).get("end_date")
start_dt = _parse_ga_date(start_date)
end_dt = _parse_ga_date(end_date)
count += self._merge(
AdCampaign,
self.company_id,
campaign_id,
{
"source_service": "google_ads",
"name": name,
"status": status,
"channel_type": channel_type,
"budget": budget,
"start_date": start_dt,
"end_date": end_dt,
"metadata_json": fields,
},
)
db.session.commit()
logger.info("Google Ads campaigns synced: %d records merged", count)
return count
# -- Metrics sync with pagination ----------------------------------------
def _sync_metrics(self) -> int:
"""Pull daily campaign metrics via GAQL, upsert to AdMetric."""
from app.models import db, AdMetric
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
logger.warning("Google Ads sync: missing credentials")
return 0
# Pull last 90 days of data
end_date = date.today()
start_date = end_date - timedelta(days=90)
start_str = start_date.strftime("%Y-%m-%d")
end_str = end_date.strftime("%Y-%m-%d")
query = (
f"SELECT "
f"campaign.id, segments.date, "
f"metrics.impressions, metrics.clicks, "
f"metrics.cost_micros, metrics.conversions, "
f"metrics.cost_per_conversion "
f"FROM campaign "
f"WHERE segments.date BETWEEN '{start_str}' AND '{end_str}'"
)
all_rows: List[Dict[str, Any]] = self._gaql_paginate(
query=query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
count = 0
for row in all_rows:
fields = row.get("fields", {})
campaign_id = str(fields.get("campaign", {}).get("id", ""))
metric_date_str = fields.get("segments", {}).get("date", "")
if not metric_date_str:
continue
impressions = int(fields.get("metrics", {}).get("impressions", 0) or 0)
clicks = int(fields.get("metrics", {}).get("clicks", 0) or 0)
cost_micros = fields.get("metrics", {}).get("cost_micros", 0)
conversions = float(fields.get("metrics", {}).get("conversions", 0) or 0)
spend = 0.0
if cost_micros:
try:
spend = float(cost_micros) / 1_000_000
except (ValueError, TypeError):
pass
ctr = (clicks / impressions * 100) if impressions > 0 else None
cpc = (spend / clicks) if clicks > 0 else None
roas = None # Would need conversion value data
# Build composite external_id for dedup
ext_id = f"{campaign_id}_{metric_date_str}"
count += self._merge(
AdMetric,
self.company_id,
ext_id,
{
"external_campaign_id": campaign_id,
"source_service": "google_ads",
"metric_date": _parse_ga_date(metric_date_str),
"spend": spend,
"impressions": impressions,
"clicks": clicks,
"conversions": conversions,
"ctr": ctr,
"cpc": cpc,
"roas": roas,
"metadata_json": fields,
},
)
db.session.commit()
logger.info("Google Ads metrics synced: %d records merged", count)
return count
# -- Keyword sync with pagination ----------------------------------------
def _sync_keywords(self) -> int:
"""Pull keywords with lifetime metrics via GAQL, upsert to AdKeyword.
Pulls keywords at the ad group criterion level with their current bid,
status, and lifetime performance metrics (impressions, clicks, spend,
conversions, avg position, avg ctr).
"""
from app.models import db, AdKeyword
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
logger.warning("Google Ads keywords sync: missing credentials")
return 0
# Get all enabled campaigns for this customer to scope the query
campaign_query = (
"SELECT campaign.id, campaign.name, campaign.status "
"FROM campaign "
"WHERE campaign.advertising_channel_type = 'SEARCH'"
)
campaign_rows = self._gaql_paginate(
query=campaign_query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
# Build a set of campaign resource names to filter keywords
campaign_ids = []
for row in campaign_rows:
fields = row.get("fields", {})
cid = str(fields.get("campaign", {}).get("id", ""))
if cid:
campaign_ids.append(cid)
if not campaign_ids:
logger.info("Google Ads keywords sync: no SEARCH campaigns found")
return 0
# Pull keywords from all ad groups in search campaigns
# Include lifetime metrics and current bid/status
keyword_query = (
"SELECT "
"campaign.id, ad_group.id, ad_group.name, "
"ad_group_criterion.id, "
"ad_group_criterion.keyword.text, "
"ad_group_criterion.keyword.match_type, "
"ad_group_criterion.status, "
"ad_group_criterion.cpc_bid_micros, "
"metrics.impressions, "
"metrics.clicks, "
"metrics.cost_micros, "
"metrics.conversions, "
"metrics.average_position, "
"metrics.ctr "
"FROM ad_group_criterion "
"WHERE ad_group_criterion.keyword IS NOT NULL "
"AND segments.date DURING LAST_90_DAYS"
)
all_rows: List[Dict[str, Any]] = self._gaql_paginate(
query=keyword_query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
count = 0
for row in all_rows:
fields = row.get("fields", {})
campaign_id = str(fields.get("campaign", {}).get("id", ""))
ad_group = fields.get("ad_group", {})
ad_group_id = str(ad_group.get("id", ""))
ad_group_name = ad_group.get("name", "")
criterion = fields.get("ad_group_criterion", {})
keyword_info = criterion.get("keyword", {})
keyword_text = str(keyword_info.get("text", "")).strip()
match_type_raw = keyword_info.get("match_type", "BROAD")
criterion_status = criterion.get("status", "ENABLED")
if not keyword_text:
continue
# Map Google match types to our lowercase convention
match_type = match_type_raw.lower() if match_type_raw else "broad"
# Map status
status_map = {
"ENABLED": "enabled",
"PAUSED": "paused",
"REMOVED": "removed",
"ELIGIBLE": "enabled",
"NOT_ELIGIBLE": "paused",
}
status = status_map.get(criterion_status, "enabled")
# Parse metrics
metrics = fields.get("metrics", {})
impressions = int(metrics.get("impressions", 0) or 0)
clicks = int(metrics.get("clicks", 0) or 0)
cost_micros = metrics.get("cost_micros", 0)
conversions = float(metrics.get("conversions", 0) or 0)
spend = 0.0
if cost_micros:
try:
spend = float(cost_micros) / 1_000_000
except (ValueError, TypeError):
pass
ctr = float(metrics.get("ctr", 0) or 0)
if ctr > 0:
ctr = ctr * 100 # GAQL returns decimal fraction, we store percentage
avg_position = float(metrics.get("average_position", 0) or 0)
if avg_position == 0:
avg_position = None
# Parse CPC bid (micros -> dollars)
cpc_micros = criterion.get("cpc_bid_micros")
max_cpc = None
if cpc_micros:
try:
max_cpc = float(cpc_micros) / 1_000_000
except (ValueError, TypeError):
pass
# Build composite external_id for dedup
criterion_id = str(criterion.get("id", ""))
ext_id = f"kw_{campaign_id}_{ad_group_id}_{keyword_text}_{match_type}"
# Check if keyword already exists
existing = AdKeyword.query.filter_by(
company_id=self.company_id,
source_service="google_ads",
external_campaign_id=str(campaign_id),
external_ad_group_id=str(ad_group_id),
text=keyword_text,
match_type=match_type,
).first()
if existing:
# Update existing keyword with latest metrics
existing.status = status
existing.max_cpc = max_cpc
existing.total_impressions = impressions
existing.total_clicks = clicks
existing.total_spend = spend
existing.total_conversions = conversions
existing.avg_position = avg_position
existing.avg_ctr = ctr
existing.metadata_json = {
"criterion_id": criterion_id,
"campaign_id": str(campaign_id),
"ad_group_id": str(ad_group_id),
"ad_group_name": ad_group_name,
"match_type_raw": match_type_raw,
"raw_fields": fields,
}
else:
new_kw = AdKeyword(
company_id=self.company_id,
source_service="google_ads",
external_campaign_id=str(campaign_id),
external_ad_group_id=str(ad_group_id),
text=keyword_text,
match_type=match_type,
max_cpc=max_cpc,
status=status,
total_impressions=impressions,
total_clicks=clicks,
total_spend=spend,
total_conversions=conversions,
avg_position=avg_position,
avg_ctr=ctr,
metadata_json={
"criterion_id": criterion_id,
"campaign_id": str(campaign_id),
"ad_group_id": str(ad_group_id),
"ad_group_name": ad_group_name,
"match_type_raw": match_type_raw,
"raw_fields": fields,
},
)
db.session.add(new_kw)
count += 1 # count both inserts and updates
db.session.commit()
logger.info("Google Ads keywords synced: %d records merged", count)
return count
# -- Keyword research sync -----------------------------------------------
def _sync_keyword_research(self) -> int:
"""Pull keyword research data (search volume, competition, CPC suggestions)
via KeywordPlanIdeaService and upsert to AdKeyword research fields.
Uses the Google Ads API v17 keywordPlanIdeaService endpoint to fetch
search volume, competition level, and suggested CPC for each keyword
already tracked in AdKeyword.
Returns the number of keywords updated with research data.
"""
from app.models import db, AdKeyword
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
logger.warning("Google Ads keyword research sync: missing credentials")
return 0
# Fetch all tracked keywords for this company
keywords = AdKeyword.query.filter_by(
company_id=self.company_id,
source_service="google_ads",
).all()
if not keywords:
logger.info("Google Ads keyword research: no keywords to process")
return 0
# Group keywords by text for batch lookups
unique_texts = list({kw.text for kw in keywords})
logger.info("Google Ads keyword research: %d unique keywords to research", len(unique_texts))
# Fetch keyword plan ideas in batches (API limit: 100 per request)
batch_size = 100
all_ideas = []
for i in range(0, len(unique_texts), batch_size):
batch = unique_texts[i:i + batch_size]
try:
ideas = self._get_keyword_ideas(
keywords=batch,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
all_ideas.extend(ideas)
except Exception as e:
logger.error("Google Ads keyword research batch failed: %s", e)
continue
time.sleep(0.5) # Rate limit between batches
# Build lookup by keyword text (lowercase)
ideas_by_text = {}
for idea in all_ideas:
text = idea.get("keyword_text", "").lower().strip()
if text:
ideas_by_text[text] = idea
# Upsert research data
count = 0
for kw in keywords:
idea = ideas_by_text.get(kw.text.lower().strip())
if idea:
kw.search_volume = idea.get("avg_monthly_searches")
kw.competition_level = idea.get("competition", "").lower() if idea.get("competition") else None
kw.cpc_suggestion = idea.get("suggested_bid")
kw.trend_data = idea.get("search_volume_by_month")
count += 1
db.session.commit()
logger.info("Google Ads keyword research: %d/%d keywords updated", count, len(keywords))
return count
def _get_keyword_ideas(
self,
*,
keywords: List[str],
customer_id: str,
access_token: str,
developer_token: str,
) -> List[Dict[str, Any]]:
"""Fetch keyword plan ideas from KeywordPlanIdeaService.
Uses the non-GAQL REST endpoint for keyword planning data.
Returns a list of dicts with search volume, competition, and CPC.
"""
url = f"{self._BASE_URL}/customers/{customer_id}:generateKeywordIdeas"
# Build seed keywords request
seed_keywords = []
for text in keywords:
seed_keywords.append({
"keywordText": text,
"languageCode": "en",
})
payload = {
"request": {
"seedKeywords": seed_keywords,
"locationIds": ["2840"], # United States
"languageCode": "en",
}
}
headers = {
"developer-token": developer_token,
"login-customer-id": customer_id,
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
}
try:
response = requests.post(url, json=payload, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
except requests.exceptions.RequestException as e:
logger.error("Google Ads keyword ideas request failed: %s", e)
return []
# Parse response
ideas = []
for idea in data.get("keywordIdeas", []):
keyword_text = idea.get("keyword", {}).get("text", "")
search_volume = idea.get("searchVolume", {})
avg_monthly_searches = search_volume.get("avgMonthlySearches")
competition = idea.get("competition", "")
suggested_bid = idea.get("suggestedBid", {})
suggested_bid_micros = suggested_bid.get("microAmount")
# Parse trend data (search volume by month)
trend = search_volume.get("searchVolumeByMonth")
trend_data = None
if trend:
trend_data = []
for month_data in trend:
trend_data.append({
"year": month_data.get("year"),
"month": month_data.get("month"),
"searchVolume": month_data.get("searchVolume"),
})
cpc_suggestion = None
if suggested_bid_micros:
try:
cpc_suggestion = float(suggested_bid_micros) / 1_000_000
except (ValueError, TypeError):
pass
# Map competition enum to human-readable
competition_map = {
"COMPETITION_LOW": "low",
"COMPETITION_MEDIUM": "medium",
"COMPETITION_HIGH": "high",
}
ideas.append({
"keyword_text": keyword_text,
"avg_monthly_searches": avg_monthly_searches,
"competition": competition_map.get(competition, competition.lower() if competition else None),
"suggested_bid": cpc_suggestion,
"search_volume_by_month": trend_data,
})
return ideas
# -- GAQL pagination helper ---------------------------------------------
def _gaql_paginate(
self,
*,
query: str,
customer_id: str,
access_token: str,
developer_token: str,
) -> List[Dict[str, Any]]:
"""Execute GAQL query with stream_token pagination."""
all_rows: List[Dict[str, Any]] = []
stream_token = None
while True:
self._check_rate_limit()
response = self._retry(
_gaql_query,
query=query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
page_token=stream_token,
)
results = response.get("results", [])
if not results:
break
all_rows.extend(results)
logger.info(
"Google Ads: fetched %d rows (total %d)",
len(results), len(all_rows),
)
# Check for stream token (pagination)
stream_token = response.get("nextPageToken")
if not stream_token:
break
time.sleep(0.1) # Stay within rate limits
logger.info("Google Ads: total %d rows", len(all_rows))
return all_rows
# -- Merge (upsert) helper -----------------------------------------------
@staticmethod
def _merge(model, company_id: str, external_id: str, defaults: Dict[str, Any]) -> int:
"""Find or create a record by (company_id, external_id) and update fields."""
from app.models import db
record = model.query.filter_by(
company_id=company_id, external_id=external_id
).first()
if record:
for key, value in defaults.items():
if hasattr(record, key):
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**{k: v for k, v in defaults.items() if hasattr(model, k)}
)
db.session.add(record)
return 1
# -- status --------------------------------------------------------------
def status(self) -> Dict[str, Any]:
if not self._connected:
return {
"service": "google_ads",
"connected": False,
"config_present": bool(self.config.get("access_token")) and bool(self.config.get("customer_id")),
}
try:
response = self._retry(
_gaql_query,
query="SELECT customer_descriptor.customer_id FROM customer",
customer_id=self.config["customer_id"],
access_token=self.config["access_token"],
developer_token=self.config.get("developer_token", ""),
)
return {
"service": "google_ads",
"connected": True,
"customer_id": self.config.get("customer_id"),
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {"service": "google_ads", "connected": False, "error": str(exc)}
# -- Write operations (Lead Gen Engine) ---------------------------------
def create_campaign(self, name: str, budget: float, bidding_strategy: str = "TARGET_CPA", target_cpa: Optional[float] = None, status: str = "ENABLED") -> Dict[str, Any]:
"""Create a new campaign via Google Ads API mutation.
Returns the campaign resource name and ID.
"""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
if not customer_id or not access_token:
return {"error": "Missing credentials"}
# Build campaign mutation
budget_micros = int(budget * 1_000_000)
operations = [{
"create": {
"name": name,
"advertisingChannelType": "SEARCH",
"status": status,
"billingSeason": "STANDARD",
"campaignBudget": {
"resourceName": f"customers/{customer_id}/campaignBudgets/{budget_micros}",
},
"campaignBroadcastSearchSetting": {
"status": "CAMPAIGN_BROADCAST_SEARCH_SETTING_DISABLED",
},
}
}]
if bidding_strategy == "TARGET_CPA" and target_cpa:
operations[0]["create"]["cpcBid"] = {
"cpcBid": {"microAmount": str(int(target_cpa * 1_000_000))}
}
# Create campaign budget first (idempotent)
budget_op = [{
"create": {
"name": f"{name} Budget",
"amountMicros": str(budget_micros),
"deliveryMethod": "STANDARD",
"explicitlyShared": True,
}
}]
self._check_rate_limit()
budget_resp = self._retry(
self._mutate_resource,
resource="campaignBudgets",
operations=budget_op,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
budget_resource = budget_resp.get("results", [{}])[0].get("resourceName", "")
# Update campaign op with actual budget resource
operations[0]["create"]["campaignBudget"] = budget_resource
self._check_rate_limit()
resp = self._retry(
self._mutate_resource,
resource="campaigns",
operations=operations,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
result = resp.get("results", [{}])[0]
resource_name = result.get("resourceName", "")
# Extract campaign ID from resource name: customers/{cid}/campaigns/{campaign_id}
campaign_id = resource_name.split("/")[-1] if resource_name else ""
return {
"resource_name": resource_name,
"campaign_id": campaign_id,
"budget_resource": budget_resource,
"name": name,
"status": status,
}
def create_ad_group(self, campaign_resource_name: str, name: str, bidding_strategy: str = "BID_STRATEGY_UNSPECIFIED", max_cpc: Optional[float] = None) -> Dict[str, Any]:
"""Create an ad group within a campaign."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
operation = {
"create": {
"name": name,
"campaign": campaign_resource_name,
"status": "ENABLED",
"adGroupType": "SEARCH",
}
}
if max_cpc:
operation["create"]["cpcBid"] = {
"microAmount": str(int(max_cpc * 1_000_000))
}
self._check_rate_limit()
resp = self._retry(
self._mutate_resource,
resource="adGroups",
operations=[operation],
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
result = resp.get("results", [{}])[0]
resource_name = result.get("resourceName", "")
ad_group_id = resource_name.split("/")[-1] if resource_name else ""
return {
"resource_name": resource_name,
"ad_group_id": ad_group_id,
"name": name,
}
def add_keyword(self, ad_group_resource_name: str, keyword: str, match_type: str = "BROAD") -> Dict[str, Any]:
"""Add a keyword to an ad group."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
# Map match type strings
match_map = {
"BROAD": "BROAD",
"PHRASE": "PHRASE",
"EXACT": "EXACT",
"NEGATIVE_BROAD": "NEGATIVE_BROAD",
"NEGATIVE_PHRASE": "NEGATIVE_PHRASE",
}
ga_match_type = match_map.get(match_type.upper(), "BROAD")
# Format keyword with match type prefix
formatted_keyword = keyword
if ga_match_type == "PHRASE":
formatted_keyword = f'"{keyword}"'
elif ga_match_type == "EXACT":
formatted_keyword = f"[{keyword}]"
# BROAD and negative types don't need prefix
operation = {
"create": {
"adGroup": ad_group_resource_name,
"adGroupCriterion": {
"keyword": {
"text": keyword,
"matchType": ga_match_type,
},
},
}
}
self._check_rate_limit()
resp = self._retry(
self._mutate_resource,
resource="adGroupCriteria",
operations=[operation],
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
result = resp.get("results", [{}])[0]
resource_name = result.get("resourceName", "")
criterion_id = resource_name.split("/")[-1] if resource_name else ""
return {
"resource_name": resource_name,
"criterion_id": criterion_id,
"keyword": keyword,
"match_type": ga_match_type,
}
def create_ad(self, ad_group_resource_name: str, headlines: List[str], descriptions: List[str], final_url: str, path1: str = "", path2: str = "") -> Dict[str, Any]:
"""Convenience wrapper — create a Responsive Search Ad from lists of headlines/descriptions.
Maps to create_text_ad but accepts headline/description lists and a single URL string.
"""
headline1 = headlines[0] if headlines else ""
headline2 = headlines[1] if len(headlines) > 1 else headline1
description = descriptions[0] if descriptions else ""
final_urls = [final_url] if final_url else []
return self.create_text_ad(ad_group_resource_name, headline1, headline2, description, final_urls, path1, path2)
def create_text_ad(self, ad_group_resource_name: str, headline1: str, headline2: str, description: str, final_urls: List[str], path1: str = "", path2: str = "") -> Dict[str, Any]:
"""Create a Responsive Search Ad."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
ad = {
"type": "ResponsiveSearchAd",
"responsiveSearchAd": {
"headlines": [{"text": headline1}, {"text": headline2}],
"descriptions": [{"text": description}],
"path1": path1,
"path2": path2,
"finalUrls": final_urls,
},
}
operation = {
"create": {
"adGroup": ad_group_resource_name,
"ad": ad,
"status": "ENABLED",
}
}
self._check_rate_limit()
resp = self._retry(
self._mutate_resource,
resource="ads",
operations=[operation],
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
result = resp.get("results", [{}])[0]
resource_name = result.get("resourceName", "")
ad_id = resource_name.split("/")[-1] if resource_name else ""
return {
"resource_name": resource_name,
"ad_id": ad_id,
"headline1": headline1,
"headline2": headline2,
}
def pause_campaign(self, campaign_resource_name: str) -> Dict[str, Any]:
"""Pause (disable) a campaign."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
operation = {
"update": {
"resourceName": campaign_resource_name,
"status": "PAUSED",
"updateMask": "status",
}
}
self._check_rate_limit()
resp = self._retry(
self._mutate_resource,
resource="campaigns",
operations=[operation],
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
return {
"status": "paused",
"resource_name": campaign_resource_name,
}
def update_campaign_budget(self, campaign_resource_name: str, new_budget: float) -> Dict[str, Any]:
"""Update campaign daily budget."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
budget_micros = int(new_budget * 1_000_000)
# First get the campaign's budget resource name
query = f"SELECT campaign.campaign_budget FROM campaign WHERE campaign.resource_name = '{campaign_resource_name}'"
self._check_rate_limit()
resp = self._retry(
_gaql_query,
query=query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
budget_resource = resp.get("results", [{}])[0].get("fields", {}).get("campaign", {}).get("campaignBudget", "")
if not budget_resource:
return {"error": "Could not retrieve campaign budget resource"}
operation = [{
"update": {
"resourceName": budget_resource,
"amountMicros": str(budget_micros),
"updateMask": "amountMicros",
}
}]
self._check_rate_limit()
self._retry(
self._mutate_resource,
resource="campaignBudgets",
operations=operation,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
return {
"status": "updated",
"new_budget": new_budget,
"budget_resource": budget_resource,
}
def pause_keyword(self, ad_group_resource_name: str, keyword: str, match_type: str = "BROAD") -> Dict[str, Any]:
"""Pause a keyword in an ad group via adGroupCriteria mutation.
Uses UPDATE to set status to PAUSED on the matching criterion.
"""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
match_map = {
"BROAD": "BROAD", "PHRASE": "PHRASE", "EXACT": "EXACT",
"NEGATIVE_BROAD": "NEGATIVE_BROAD", "NEGATIVE_PHRASE": "NEGATIVE_PHRASE",
}
ga_match_type = match_map.get(match_type.upper(), "BROAD")
# Find the criterion resource name first
find_query = (
f"SELECT ad_group_criterion.id "
f"FROM ad_group_criterion "
f"WHERE ad_group.id = '{ad_group_resource_name.split('/')[-1]}' "
f"AND ad_group_criterion.keyword.text = '{keyword}' "
f"AND ad_group_criterion.keyword.match_type = '{ga_match_type}'"
)
self._check_rate_limit()
find_resp = self._retry(
_gaql_query,
query=find_query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
results = find_resp.get("results", [])
if not results:
return {"status": "error", "error": f"Keyword not found: '{keyword}' ({ga_match_type})"}
criterion_resource = results[0].get("fields", {}).get("ad_group_criterion", {}).get("id", "")
if not criterion_resource:
return {"status": "error", "error": "Could not resolve keyword criterion resource name"}
# Pause the criterion
operation = [{
"update": {
"resourceName": criterion_resource,
"status": "PAUSED",
"updateMask": "status",
}
}]
self._check_rate_limit()
self._retry(
self._mutate_resource,
resource="adGroupCriteria",
operations=operation,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
return {
"status": "paused",
"keyword": keyword,
"match_type": ga_match_type,
"criterion_resource": criterion_resource,
}
def enable_keyword(self, ad_group_resource_name: str, keyword: str, match_type: str = "BROAD") -> Dict[str, Any]:
"""Resume an enabled keyword in an ad group."""
customer_id = self.config.get("customer_id", "")
access_token = self.config.get("access_token", "")
developer_token = self.config.get("developer_token", "")
match_map = {
"BROAD": "BROAD", "PHRASE": "PHRASE", "EXACT": "EXACT",
"NEGATIVE_BROAD": "NEGATIVE_BROAD", "NEGATIVE_PHRASE": "NEGATIVE_PHRASE",
}
ga_match_type = match_map.get(match_type.upper(), "BROAD")
find_query = (
f"SELECT ad_group_criterion.id "
f"FROM ad_group_criterion "
f"WHERE ad_group.id = '{ad_group_resource_name.split('/')[-1]}' "
f"AND ad_group_criterion.keyword.text = '{keyword}' "
f"AND ad_group_criterion.keyword.match_type = '{ga_match_type}'"
)
self._check_rate_limit()
find_resp = self._retry(
_gaql_query,
query=find_query,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
results = find_resp.get("results", [])
if not results:
return {"status": "error", "error": f"Keyword not found: '{keyword}' ({ga_match_type})"}
criterion_resource = results[0].get("fields", {}).get("ad_group_criterion", {}).get("id", "")
if not criterion_resource:
return {"status": "error", "error": "Could not resolve keyword criterion resource name"}
operation = [{
"update": {
"resourceName": criterion_resource,
"status": "ENABLED",
"updateMask": "status",
}
}]
self._check_rate_limit()
self._retry(
self._mutate_resource,
resource="adGroupCriteria",
operations=operation,
customer_id=customer_id,
access_token=access_token,
developer_token=developer_token,
)
return {
"status": "enabled",
"keyword": keyword,
"match_type": ga_match_type,
"criterion_resource": criterion_resource,
}
def _mutate_resource(
self,
resource: str,
operations: List[Dict[str, Any]],
customer_id: str,
access_token: str,
developer_token: str,
) -> Dict[str, Any]:
"""Execute a mutation request on a Google Ads resource."""
url = f"{self._BASE_URL}/customers/{customer_id}/{resource}/mutate"
payload = {
"customerId": customer_id,
"operations": operations,
}
return requests.post(
url,
json=payload,
headers={
"developer-token": developer_token,
"login-customer-id": customer_id,
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
},
timeout=30,
).json()
# -- End write operations -----------------------------------------------
# -- Register in the framework registry -------------------------------------
_REGISTRY["google_ads"] = GoogleAdsConnector
# -- Internal API helpers ---------------------------------------------------
def _gaql_query(
*,
query: str,
customer_id: str,
access_token: str,
developer_token: str = "",
page_token: Optional[str] = None,
) -> Dict[str, Any]:
"""Execute a GAQL query against the Google Ads REST API."""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"Developer-Token": developer_token,
"Login-Customer-Id": customer_id.replace("-", ""),
}
payload = {"query": query}
if page_token:
payload["pageToken"] = page_token
url = f"https://googleads.googleapis.com/v17/customers/{customer_id.replace('-', '')}:search"
resp = requests.post(url, headers=headers, json=payload, timeout=30)
resp.raise_for_status()
return resp.json()
def _parse_ga_date(date_str: Optional[str]) -> Optional[datetime]:
"""Parse Google Ads YYYYMMDD date string into datetime."""
if not date_str or not isinstance(date_str, str):
return None
try:
d = datetime.strptime(date_str, "%Y%m%d")
return d.replace(tzinfo=timezone.utc)
except ValueError:
return None