"""Creative tools for Google Ad Manager."""
import base64
import logging
import re
from pathlib import Path
from typing import Optional, List
from ..client import get_gam_client
from ..utils import safe_get
logger = logging.getLogger(__name__)
def extract_size_from_filename(filename: str) -> tuple[Optional[int], Optional[int]]:
"""Extract size (e.g., '300x250') from filename.
Args:
filename: The filename to parse
Returns:
Tuple of (width, height) or (None, None) if not found
"""
match = re.search(r'(\d+)x(\d+)', filename)
if match:
return int(match.group(1)), int(match.group(2))
return None, None
def upload_creative(
file_path: str,
advertiser_id: int,
click_through_url: str,
creative_name: Optional[str] = None,
override_size_width: Optional[int] = None,
override_size_height: Optional[int] = None
) -> dict:
"""Upload an image creative to Ad Manager.
Args:
file_path: Path to the image file
advertiser_id: ID of the advertiser
click_through_url: Destination URL when creative is clicked
creative_name: Optional name for the creative (defaults to auto-generated)
override_size_width: Optional width to override the creative size (for serving into different slot)
override_size_height: Optional height to override the creative size (for serving into different slot)
Returns:
dict with created creative details
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
path = Path(file_path)
filename = path.name
if not path.exists():
return {"error": f"File not found: {file_path}"}
# Read and encode image
with open(path, 'rb') as f:
image_data = f.read()
image_data_base64 = base64.b64encode(image_data).decode('utf-8')
# Extract size from filename
width, height = extract_size_from_filename(filename)
if not width or not height:
return {"error": f"Could not extract size from filename: {filename}. Expected format like '300x250'"}
# Determine creative size (use override if provided)
creative_width = override_size_width if override_size_width else width
creative_height = override_size_height if override_size_height else height
use_override = override_size_width is not None and override_size_height is not None
# Generate name if not provided
if creative_name is None:
creative_name = f"Creative - {creative_width}x{creative_height} - {path.stem}"
creative = {
'xsi_type': 'ImageCreative',
'name': creative_name,
'advertiserId': advertiser_id,
'destinationUrl': click_through_url,
'size': {
'width': creative_width,
'height': creative_height,
'isAspectRatio': False
},
'primaryImageAsset': {
'assetByteArray': image_data_base64,
'fileName': filename
}
}
# Set overrideSize when using different dimensions than the actual image
if use_override:
creative['overrideSize'] = True
created_creatives = creative_service.createCreatives([creative])
if not created_creatives:
return {"error": "Failed to create creative"}
created = created_creatives[0]
result = {
"id": safe_get(created, 'id'),
"name": safe_get(created, 'name'),
"advertiser_id": advertiser_id,
"size": f"{creative_width}x{creative_height}",
"click_through_url": click_through_url,
"message": f"Creative '{creative_name}' uploaded successfully"
}
if use_override:
result["original_size"] = f"{width}x{height}"
result["override_size"] = True
return result
def upload_creative_from_base64(
image_base64: str,
filename: str,
advertiser_id: int,
click_through_url: str,
width: int,
height: int,
creative_name: Optional[str] = None
) -> dict:
"""Upload an image creative from base64 data.
Args:
image_base64: Base64 encoded image data
filename: Original filename
advertiser_id: ID of the advertiser
click_through_url: Destination URL when creative is clicked
width: Image width
height: Image height
creative_name: Optional name for the creative
Returns:
dict with created creative details
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
if creative_name is None:
creative_name = f"Creative - {width}x{height}"
creative = {
'xsi_type': 'ImageCreative',
'name': creative_name,
'advertiserId': advertiser_id,
'destinationUrl': click_through_url,
'size': {
'width': width,
'height': height,
'isAspectRatio': False
},
'primaryImageAsset': {
'assetByteArray': image_base64,
'fileName': filename
}
}
created_creatives = creative_service.createCreatives([creative])
if not created_creatives:
return {"error": "Failed to create creative"}
created = created_creatives[0]
return {
"id": created['id'],
"name": created['name'],
"advertiser_id": advertiser_id,
"size": f"{width}x{height}",
"message": f"Creative uploaded successfully"
}
def associate_creative_with_line_item(
creative_id: int,
line_item_id: int,
size_override_width: Optional[int] = None,
size_override_height: Optional[int] = None
) -> dict:
"""Associate a creative with a line item.
Args:
creative_id: The creative ID
line_item_id: The line item ID
size_override_width: Optional width for size override
size_override_height: Optional height for size override
Returns:
dict with association details
"""
client = get_gam_client()
lica_service = client.get_service('LineItemCreativeAssociationService')
lica = {
'creativeId': creative_id,
'lineItemId': line_item_id,
}
if size_override_width and size_override_height:
lica['sizes'] = [{
'width': size_override_width,
'height': size_override_height,
'isAspectRatio': False
}]
created_licas = lica_service.createLineItemCreativeAssociations([lica])
if not created_licas:
return {"error": "Failed to create creative association"}
return {
"creative_id": creative_id,
"line_item_id": line_item_id,
"size_override": f"{size_override_width}x{size_override_height}" if size_override_width else None,
"message": f"Creative {creative_id} associated with line item {line_item_id}"
}
def upload_and_associate_creative(
file_path: str,
advertiser_id: int,
line_item_id: int,
click_through_url: str,
creative_name: Optional[str] = None
) -> dict:
"""Upload a creative and associate it with a line item in one operation.
Args:
file_path: Path to the image file
advertiser_id: ID of the advertiser
line_item_id: ID of the line item
click_through_url: Destination URL
creative_name: Optional name for the creative
Returns:
dict with both upload and association results
"""
# First upload
upload_result = upload_creative(
file_path=file_path,
advertiser_id=advertiser_id,
click_through_url=click_through_url,
creative_name=creative_name
)
if "error" in upload_result:
return upload_result
creative_id = upload_result["id"]
# Then associate
assoc_result = associate_creative_with_line_item(
creative_id=creative_id,
line_item_id=line_item_id
)
if "error" in assoc_result:
return {
"creative": upload_result,
"association_error": assoc_result["error"]
}
return {
"creative_id": creative_id,
"creative_name": upload_result["name"],
"size": upload_result["size"],
"line_item_id": line_item_id,
"click_through_url": click_through_url,
"message": f"Creative uploaded and associated with line item {line_item_id}"
}
def bulk_upload_creatives(
folder_path: str,
advertiser_id: int,
line_item_id: int,
click_through_url: str,
name_prefix: Optional[str] = None
) -> dict:
"""Upload all creatives from a folder and associate with a line item.
Args:
folder_path: Path to folder containing image files
advertiser_id: ID of the advertiser
line_item_id: ID of the line item
click_through_url: Destination URL
name_prefix: Optional prefix for creative names
Returns:
dict with upload results
"""
folder = Path(folder_path)
if not folder.exists():
return {"error": f"Folder not found: {folder_path}"}
# Find all image files
extensions = ['*.jpg', '*.jpeg', '*.png', '*.gif', '*.JPG', '*.JPEG', '*.PNG', '*.GIF']
files = []
for ext in extensions:
files.extend(folder.glob(ext))
if not files:
return {"error": f"No image files found in {folder_path}"}
results = {
"folder": folder_path,
"line_item_id": line_item_id,
"advertiser_id": advertiser_id,
"uploaded": [],
"failed": [],
"total_files": len(files)
}
for file_path in sorted(files):
creative_name = None
if name_prefix:
width, height = extract_size_from_filename(file_path.name)
if width and height:
creative_name = f"{name_prefix} - {width}x{height}"
result = upload_and_associate_creative(
file_path=str(file_path),
advertiser_id=advertiser_id,
line_item_id=line_item_id,
click_through_url=click_through_url,
creative_name=creative_name
)
if "error" in result:
results["failed"].append({
"file": file_path.name,
"error": result["error"]
})
else:
results["uploaded"].append({
"file": file_path.name,
"creative_id": result["creative_id"],
"size": result["size"]
})
results["success_count"] = len(results["uploaded"])
results["fail_count"] = len(results["failed"])
results["message"] = f"Uploaded {results['success_count']} of {results['total_files']} creatives"
return results
def get_creative(creative_id: int) -> dict:
"""Get creative details by ID.
Args:
creative_id: The creative ID
Returns:
dict with creative details
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
statement = client.create_statement()
statement = statement.Where("id = :id").WithBindVariable('id', creative_id)
response = creative_service.getCreativesByStatement(statement.ToStatement())
if 'results' not in response or len(response['results']) == 0:
return {"error": f"Creative {creative_id} not found"}
creative = response['results'][0]
size = safe_get(creative, 'size')
return {
"id": safe_get(creative, 'id'),
"name": safe_get(creative, 'name'),
"advertiser_id": safe_get(creative, 'advertiserId'),
"size": f"{safe_get(size, 'width')}x{safe_get(size, 'height')}" if size else None,
"type": safe_get(creative, 'Creative.Type'),
"destination_url": safe_get(creative, 'destinationUrl')
}
def list_creatives_by_advertiser(advertiser_id: int, limit: int = 100) -> dict:
"""List creatives for an advertiser.
Args:
advertiser_id: The advertiser ID
limit: Maximum number of creatives to return
Returns:
dict with creatives list
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
statement = client.create_statement()
statement = statement.Where(
"advertiserId = :advertiserId"
).WithBindVariable('advertiserId', advertiser_id).Limit(limit)
response = creative_service.getCreativesByStatement(statement.ToStatement())
if 'results' not in response:
return {"creatives": [], "total": 0}
creatives = []
for c in response['results']:
size = safe_get(c, 'size')
creatives.append({
"id": safe_get(c, 'id'),
"name": safe_get(c, 'name'),
"size": f"{safe_get(size, 'width')}x{safe_get(size, 'height')}" if size else None,
"type": safe_get(c, 'Creative.Type')
})
return {
"advertiser_id": advertiser_id,
"creatives": creatives,
"total": len(creatives)
}
def update_creative(
creative_id: int,
destination_url: Optional[str] = None,
name: Optional[str] = None
) -> dict:
"""Update an existing creative's properties.
Args:
creative_id: The creative ID to update
destination_url: New destination URL (click-through URL)
name: New name for the creative
Returns:
dict with updated creative details
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
# First, get the existing creative
statement = client.create_statement()
statement = statement.Where("id = :id").WithBindVariable('id', creative_id)
response = creative_service.getCreativesByStatement(statement.ToStatement())
if 'results' not in response or len(response['results']) == 0:
return {"error": f"Creative {creative_id} not found"}
creative = response['results'][0]
# Update the fields
if destination_url is not None:
creative['destinationUrl'] = destination_url
if name is not None:
creative['name'] = name
# Update the creative
updated_creatives = creative_service.updateCreatives([creative])
if not updated_creatives:
return {"error": "Failed to update creative"}
updated = updated_creatives[0]
size = safe_get(updated, 'size')
return {
"id": safe_get(updated, 'id'),
"name": safe_get(updated, 'name'),
"advertiser_id": safe_get(updated, 'advertiserId'),
"size": f"{safe_get(size, 'width')}x{safe_get(size, 'height')}" if size else None,
"type": safe_get(updated, 'Creative.Type'),
"destination_url": safe_get(updated, 'destinationUrl'),
"message": f"Creative {creative_id} updated successfully"
}
def create_third_party_creative(
advertiser_id: int,
name: str,
width: int,
height: int,
snippet: str,
expanded_snippet: Optional[str] = None,
is_safe_frame_compatible: bool = True
) -> dict:
"""Create a third-party creative (HTML/JavaScript ad tag).
Use this for DCM/Campaign Manager tags, custom HTML ads, or any third-party
ad server tags that need to be served through Google Ad Manager.
Args:
advertiser_id: ID of the advertiser
name: Name for the creative
width: Creative width in pixels
height: Creative height in pixels
snippet: The HTML/JavaScript code snippet (the ad tag)
expanded_snippet: Optional expanded snippet for expandable creatives
is_safe_frame_compatible: Whether the creative works in SafeFrame (default: True)
Returns:
dict with created creative details
"""
client = get_gam_client()
creative_service = client.get_service('CreativeService')
creative = {
'xsi_type': 'ThirdPartyCreative',
'name': name,
'advertiserId': advertiser_id,
'size': {
'width': width,
'height': height,
'isAspectRatio': False
},
'snippet': snippet,
'isSafeFrameCompatible': is_safe_frame_compatible
}
if expanded_snippet:
creative['expandedSnippet'] = expanded_snippet
created_creatives = creative_service.createCreatives([creative])
if not created_creatives:
return {"error": "Failed to create third-party creative"}
created = created_creatives[0]
size = safe_get(created, 'size')
return {
"id": safe_get(created, 'id'),
"name": safe_get(created, 'name'),
"advertiser_id": advertiser_id,
"size": f"{safe_get(size, 'width')}x{safe_get(size, 'height')}" if size else None,
"type": "ThirdPartyCreative",
"is_safe_frame_compatible": is_safe_frame_compatible,
"message": f"Third-party creative '{name}' created successfully"
}
def get_creative_preview_url(
line_item_id: int,
creative_id: int,
site_url: str
) -> dict:
"""Get a preview URL for a creative associated with a line item.
This generates a preview URL that shows how the creative will appear
on the specified site URL.
Args:
line_item_id: The line item ID
creative_id: The creative ID
site_url: The URL of the site where you want to preview the creative
Returns:
dict with preview URL
"""
client = get_gam_client()
lica_service = client.get_service('LineItemCreativeAssociationService')
try:
# Use positional arguments as the SOAP API expects them in order:
# lineItemId, creativeId, siteUrl
preview_url = lica_service.getPreviewUrl(
line_item_id,
creative_id,
site_url
)
return {
"line_item_id": line_item_id,
"creative_id": creative_id,
"site_url": site_url,
"preview_url": preview_url,
"message": "Preview URL generated successfully"
}
except Exception as e:
return {
"error": str(e),
"line_item_id": line_item_id,
"creative_id": creative_id,
"site_url": site_url
}
def list_creatives_by_line_item(line_item_id: int, limit: int = 100) -> dict:
"""List creatives associated with a line item.
Args:
line_item_id: The line item ID
limit: Maximum number of creatives to return
Returns:
dict with creatives list and association details
"""
client = get_gam_client()
lica_service = client.get_service('LineItemCreativeAssociationService')
creative_service = client.get_service('CreativeService')
# Get line item creative associations
statement = client.create_statement()
statement = statement.Where(
"lineItemId = :lineItemId"
).WithBindVariable('lineItemId', line_item_id).Limit(limit)
response = lica_service.getLineItemCreativeAssociationsByStatement(
statement.ToStatement()
)
if 'results' not in response or len(response['results']) == 0:
return {
"line_item_id": line_item_id,
"creatives": [],
"total": 0,
"message": "No creatives associated with this line item"
}
# Get creative IDs from associations
creative_ids = [safe_get(lica, 'creativeId') for lica in response['results']]
associations_map = {
safe_get(lica, 'creativeId'): {
"status": safe_get(lica, 'status'),
"sizes": safe_get(lica, 'sizes')
}
for lica in response['results']
}
# Fetch creative details
creative_ids_str = ', '.join(str(cid) for cid in creative_ids if cid)
creative_statement = client.create_statement()
creative_statement = creative_statement.Where(f"id IN ({creative_ids_str})")
creative_response = creative_service.getCreativesByStatement(
creative_statement.ToStatement()
)
creatives = []
if 'results' in creative_response:
for c in creative_response['results']:
cid = safe_get(c, 'id')
size = safe_get(c, 'size')
assoc = associations_map.get(cid, {})
creatives.append({
"id": cid,
"name": safe_get(c, 'name'),
"size": f"{safe_get(size, 'width')}x{safe_get(size, 'height')}" if size else None,
"type": safe_get(c, 'Creative.Type'),
"destination_url": safe_get(c, 'destinationUrl'),
"association_status": assoc.get("status")
})
return {
"line_item_id": line_item_id,
"creatives": creatives,
"total": len(creatives)
}
