Asset Management

def mutate_assets(
    self,
    request: Optional[MutateAssetsRequest] = None,
    customer_id: Optional[str] = None,
    operations: Optional[Sequence[AssetOperation]] = None,
    partial_failure: bool = False,
    validate_only: bool = False,
    response_content_type: Optional[ResponseContentTypeEnum.ResponseContentType] = None
) -> MutateAssetsResponse:
    """
    Create, update, or remove assets.

    Args:
        request: The request object containing all parameters
        customer_id: Required customer ID for the account
        operations: List of asset operations to perform
        partial_failure: Continue processing on individual failures
        validate_only: Validate operations without executing
        response_content_type: Type of response content to return

    Returns:
        MutateAssetsResponse with operation results
    """

def get_asset(
    self,
    request: Optional[GetAssetRequest] = None,
    resource_name: Optional[str] = None
) -> Asset:
    """
    Retrieve an asset by resource name.

    Args:
        request: The request object
        resource_name: Asset resource name (customers/{customer_id}/assets/{asset_id})

    Returns:
        Asset resource object
    """

def mutate_asset_groups(
    self,
    request: Optional[MutateAssetGroupsRequest] = None,
    customer_id: Optional[str] = None,
    operations: Optional[Sequence[AssetGroupOperation]] = None,
    partial_failure: bool = False,
    validate_only: bool = False,
    response_content_type: Optional[ResponseContentTypeEnum.ResponseContentType] = None
) -> MutateAssetGroupsResponse:
    """
    Create, update, or remove asset groups.

    Args:
        request: The request object
        customer_id: Required customer ID
        operations: List of asset group operations
        partial_failure: Continue on individual failures
        validate_only: Validate without executing
        response_content_type: Response content type

    Returns:
        MutateAssetGroupsResponse with results
    """

def get_asset_group(
    self,
    request: Optional[GetAssetGroupRequest] = None,
    resource_name: Optional[str] = None
) -> AssetGroup:
    """
    Retrieve an asset group by resource name.

    Args:
        request: The request object
        resource_name: Asset group resource name

    Returns:
        AssetGroup resource object
    """

def mutate_asset_group_assets(
    self,
    request: Optional[MutateAssetGroupAssetsRequest] = None,
    customer_id: Optional[str] = None,
    operations: Optional[Sequence[AssetGroupAssetOperation]] = None,
    partial_failure: bool = False,
    validate_only: bool = False,
    response_content_type: Optional[ResponseContentTypeEnum.ResponseContentType] = None
) -> MutateAssetGroupAssetsResponse:
    """
    Create, update, or remove asset group assets.

    Args:
        request: The request object
        customer_id: Required customer ID
        operations: List of asset group asset operations
        partial_failure: Continue on individual failures
        validate_only: Validate without executing
        response_content_type: Response content type

    Returns:
        MutateAssetGroupAssetsResponse with results
    """

def mutate_asset_sets(
    self,
    request: Optional[MutateAssetSetsRequest] = None,
    customer_id: Optional[str] = None,
    operations: Optional[Sequence[AssetSetOperation]] = None,
    partial_failure: bool = False,
    validate_only: bool = False,
    response_content_type: Optional[ResponseContentTypeEnum.ResponseContentType] = None
) -> MutateAssetSetsResponse:
    """
    Create, update, or remove asset sets.

    Args:
        request: The request object
        customer_id: Required customer ID
        operations: List of asset set operations
        partial_failure: Continue on individual failures
        validate_only: Validate without executing
        response_content_type: Response content type

    Returns:
        MutateAssetSetsResponse with results
    """

def mutate_asset_set_assets(
    self,
    request: Optional[MutateAssetSetAssetsRequest] = None,
    customer_id: Optional[str] = None,
    operations: Optional[Sequence[AssetSetAssetOperation]] = None,
    partial_failure: bool = False,
    validate_only: bool = False,
    response_content_type: Optional[ResponseContentTypeEnum.ResponseContentType] = None
) -> MutateAssetSetAssetsResponse:
    """
    Create, update, or remove asset set assets.

    Args:
        request: The request object
        customer_id: Required customer ID
        operations: List of asset set asset operations
        partial_failure: Continue on individual failures
        validate_only: Validate without executing
        response_content_type: Response content type

    Returns:
        MutateAssetSetAssetsResponse with results
    """

import base64
from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException

def upload_image_asset(client, customer_id, image_path, asset_name):
    """Upload an image asset from a local file."""
    
    asset_service = client.get_service("AssetService")
    
    # Read and encode image file
    with open(image_path, 'rb') as image_file:
        image_data = image_file.read()
    
    # Create asset operation
    operation = client.get_type("AssetOperation")
    asset = operation.create
    asset.name = asset_name
    asset.type_ = client.enums.AssetTypeEnum.IMAGE
    
    # Set image asset details
    image_asset = asset.image_asset
    image_asset.data = image_data
    image_asset.mime_type = client.enums.MimeTypeEnum.IMAGE_JPEG  # or IMAGE_PNG
    
    try:
        response = asset_service.mutate_assets(
            customer_id=customer_id,
            operations=[operation]
        )
        
        asset_resource_name = response.results[0].resource_name
        print(f"Uploaded image asset: {asset_resource_name}")
        return asset_resource_name
        
    except GoogleAdsException as ex:
        print(f"Request failed with status {ex.error.code().name}")
        for error in ex.failure.errors:
            print(f"Error: {error.message}")
        return None

def upload_video_asset(client, customer_id, youtube_video_id, asset_name):
    """Create a video asset from a YouTube video."""
    
    asset_service = client.get_service("AssetService")
    
    operation = client.get_type("AssetOperation")
    asset = operation.create
    asset.name = asset_name
    asset.type_ = client.enums.AssetTypeEnum.YOUTUBE_VIDEO
    
    # Set YouTube video asset details
    youtube_video_asset = asset.youtube_video_asset
    youtube_video_asset.youtube_video_id = youtube_video_id
    youtube_video_asset.youtube_video_title = asset_name
    
    try:
        response = asset_service.mutate_assets(
            customer_id=customer_id,
            operations=[operation]
        )
        
        return response.results[0].resource_name
        
    except GoogleAdsException as ex:
        print(f"Video asset creation failed: {ex.error.code().name}")
        for error in ex.failure.errors:
            print(f"Error: {error.message}")
        return None

def create_text_asset(client, customer_id, text_content, asset_name):
    """Create a text asset for headlines, descriptions, etc."""
    
    asset_service = client.get_service("AssetService")
    
    operation = client.get_type("AssetOperation")
    asset = operation.create
    asset.name = asset_name
    asset.type_ = client.enums.AssetTypeEnum.TEXT
    
    # Set text asset details
    text_asset = asset.text_asset
    text_asset.text = text_content
    
    response = asset_service.mutate_assets(
        customer_id=customer_id,
        operations=[operation]
    )
    
    return response.results[0].resource_name

# Example usage
text_assets = [
    "Premium Quality Products",
    "Free Shipping Worldwide",
    "Shop Now and Save Big",
    "Discover amazing deals on our latest collection of premium products"
]

headline_assets = []
for i, text in enumerate(text_assets[:3]):  # First 3 for headlines
    asset_resource = create_text_asset(
        client, 
        customer_id, 
        text, 
        f"Headline {i+1}"
    )
    headline_assets.append(asset_resource)

description_asset = create_text_asset(
    client,
    customer_id,
    text_assets[3],  # Last one for description
    "Description 1"
)

def upload_logo_asset(client, customer_id, logo_path, asset_name):
    """Upload a logo asset for display campaigns."""
    
    asset_service = client.get_service("AssetService")
    
    with open(logo_path, 'rb') as logo_file:
        logo_data = logo_file.read()
    
    operation = client.get_type("AssetOperation")
    asset = operation.create
    asset.name = asset_name
    asset.type_ = client.enums.AssetTypeEnum.IMAGE
    
    # Set image asset with logo specifications
    image_asset = asset.image_asset
    image_asset.data = logo_data
    image_asset.mime_type = client.enums.MimeTypeEnum.IMAGE_PNG
    
    response = asset_service.mutate_assets(
        customer_id=customer_id,
        operations=[operation]
    )
    
    return response.results[0].resource_name

def create_performance_max_asset_group(client, customer_id, campaign_resource_name, asset_group_name, final_urls):
    """Create an asset group for Performance Max campaign."""
    
    asset_group_service = client.get_service("AssetGroupService")
    
    operation = client.get_type("AssetGroupOperation")
    asset_group = operation.create
    asset_group.name = asset_group_name
    asset_group.campaign = campaign_resource_name
    asset_group.final_urls.extend(final_urls)
    asset_group.final_mobile_urls.extend(final_urls)
    asset_group.status = client.enums.AssetGroupStatusEnum.ENABLED
    
    # Set path fields for display URLs
    asset_group.path1 = "shop"
    asset_group.path2 = "deals"
    
    try:
        response = asset_group_service.mutate_asset_groups(
            customer_id=customer_id,
            operations=[operation]
        )
        
        asset_group_resource_name = response.results[0].resource_name
        print(f"Created asset group: {asset_group_resource_name}")
        return asset_group_resource_name
        
    except GoogleAdsException as ex:
        print(f"Asset group creation failed: {ex.error.code().name}")
        for error in ex.failure.errors:
            print(f"Error: {error.message}")
        return None

def add_assets_to_asset_group(client, customer_id, asset_group_resource_name, asset_assignments):
    """Add assets to an asset group with field type assignments."""
    
    asset_group_asset_service = client.get_service("AssetGroupAssetService")
    
    operations = []
    
    for assignment in asset_assignments:
        operation = client.get_type("AssetGroupAssetOperation")
        asset_group_asset = operation.create
        asset_group_asset.asset_group = asset_group_resource_name
        asset_group_asset.asset = assignment['asset_resource_name']
        asset_group_asset.field_type = assignment['field_type']
        
        operations.append(operation)
    
    try:
        response = asset_group_asset_service.mutate_asset_group_assets(
            customer_id=customer_id,
            operations=operations
        )
        
        print(f"Added {len(response.results)} assets to asset group")
        return [result.resource_name for result in response.results]
        
    except GoogleAdsException as ex:
        print(f"Asset assignment failed: {ex.error.code().name}")
        for error in ex.failure.errors:
            print(f"Error: {error.message}")
        return []

# Example usage - Complete asset group setup
def setup_complete_asset_group(client, customer_id, campaign_resource_name):
    """Set up a complete asset group with all required assets."""
    
    # Create assets first
    headline_assets = []
    for i in range(5):  # Need 3-15 headlines
        headline_asset = create_text_asset(
            client, customer_id, f"Headline {i+1}", f"Headline {i+1}"
        )
        headline_assets.append(headline_asset)
    
    description_assets = []
    for i in range(3):  # Need 2-5 descriptions
        description_asset = create_text_asset(
            client, customer_id, f"Description {i+1}", f"Description {i+1}"
        )
        description_assets.append(description_asset)
    
    # Upload required images (would need actual image files)
    # logo_asset = upload_logo_asset(client, customer_id, "logo.png", "Company Logo")
    # marketing_image = upload_image_asset(client, customer_id, "marketing.jpg", "Marketing Image")
    # square_marketing_image = upload_image_asset(client, customer_id, "square.jpg", "Square Image")
    
    # Create asset group
    asset_group_name = "Performance Max Asset Group"
    final_urls = ["https://www.example.com"]
    
    asset_group_resource_name = create_performance_max_asset_group(
        client, customer_id, campaign_resource_name, asset_group_name, final_urls
    )
    
    # Prepare asset assignments
    asset_assignments = []
    
    # Add headlines
    for headline_asset in headline_assets:
        asset_assignments.append({
            'asset_resource_name': headline_asset,
            'field_type': client.enums.AssetFieldTypeEnum.HEADLINE
        })
    
    # Add descriptions
    for description_asset in description_assets:
        asset_assignments.append({
            'asset_resource_name': description_asset,
            'field_type': client.enums.AssetFieldTypeEnum.DESCRIPTION
        })
    
    # Add images (commented out as we need actual image files)
    # asset_assignments.extend([
    #     {
    #         'asset_resource_name': logo_asset,
    #         'field_type': client.enums.AssetFieldTypeEnum.LOGO
    #     },
    #     {
    #         'asset_resource_name': marketing_image,
    #         'field_type': client.enums.AssetFieldTypeEnum.MARKETING_IMAGE
    #     },
    #     {
    #         'asset_resource_name': square_marketing_image,
    #         'field_type': client.enums.AssetFieldTypeEnum.SQUARE_MARKETING_IMAGE
    #     }
    # ])
    
    # Add assets to asset group
    add_assets_to_asset_group(client, customer_id, asset_group_resource_name, asset_assignments)
    
    return asset_group_resource_name

def create_merchant_center_asset_set(client, customer_id, asset_set_name, merchant_id, feed_label=None):
    """Create an asset set linked to Merchant Center for Shopping campaigns."""
    
    asset_set_service = client.get_service("AssetSetService")
    
    operation = client.get_type("AssetSetOperation")
    asset_set = operation.create
    asset_set.name = asset_set_name
    asset_set.type_ = client.enums.AssetSetTypeEnum.MERCHANT_CENTER_FEED
    
    # Configure Merchant Center settings
    merchant_center_feed = asset_set.merchant_center_feed
    merchant_center_feed.merchant_id = merchant_id
    if feed_label:
        merchant_center_feed.feed_label = feed_label
    
    try:
        response = asset_set_service.mutate_asset_sets(
            customer_id=customer_id,
            operations=[operation]
        )
        
        return response.results[0].resource_name
        
    except GoogleAdsException as ex:
        print(f"Asset set creation failed: {ex.error.code().name}")
        for error in ex.failure.errors:
            print(f"Error: {error.message}")
        return None

def create_dynamic_education_asset_set(client, customer_id, asset_set_name, program_ids):
    """Create an asset set for dynamic education ads."""
    
    asset_set_service = client.get_service("AssetSetService")
    
    operation = client.get_type("AssetSetOperation")
    asset_set = operation.create
    asset_set.name = asset_set_name
    asset_set.type_ = client.enums.AssetSetTypeEnum.DYNAMIC_EDUCATION
    
    # Configure dynamic education settings
    dynamic_education_feed = asset_set.dynamic_education_feed
    dynamic_education_feed.program_ids.extend(program_ids)
    dynamic_education_feed.language_code = "en"
    
    response = asset_set_service.mutate_asset_sets(
        customer_id=customer_id,
        operations=[operation]
    )
    
    return response.results[0].resource_name

def link_asset_set_to_campaign(client, customer_id, campaign_resource_name, asset_set_resource_name):
    """Link an asset set to a campaign."""
    
    campaign_asset_set_service = client.get_service("CampaignAssetSetService")
    
    operation = client.get_type("CampaignAssetSetOperation")
    campaign_asset_set = operation.create
    campaign_asset_set.campaign = campaign_resource_name
    campaign_asset_set.asset_set = asset_set_resource_name
    
    response = campaign_asset_set_service.mutate_campaign_asset_sets(
        customer_id=customer_id,
        operations=[operation]
    )
    
    return response.results[0].resource_name

def get_asset_performance(client, customer_id, start_date, end_date):
    """Get performance metrics for assets."""
    
    googleads_service = client.get_service("GoogleAdsService")
    
    query = f'''
        SELECT 
            asset.id,
            asset.name,
            asset.type,
            asset.source,
            segments.asset_interaction_target.asset,
            segments.asset_interaction_target.interaction_on_this_asset,
            metrics.impressions,
            metrics.clicks,
            metrics.ctr,
            metrics.conversions,
            metrics.cost_micros
        FROM asset
        WHERE segments.date BETWEEN '{start_date}' AND '{end_date}'
            AND metrics.impressions > 0
        ORDER BY metrics.impressions DESC
    '''
    
    response = googleads_service.search(
        customer_id=customer_id,
        query=query
    )
    
    asset_performance = []
    for row in response:
        asset = row.asset
        metrics = row.metrics
        
        asset_data = {
            'asset_id': asset.id,
            'asset_name': asset.name,
            'asset_type': asset.type_.name,
            'source': asset.source.name,
            'impressions': metrics.impressions,
            'clicks': metrics.clicks,
            'ctr': metrics.ctr,
            'conversions': metrics.conversions,
            'cost': metrics.cost_micros / 1_000_000
        }
        asset_performance.append(asset_data)
    
    return asset_performance

def get_asset_group_performance(client, customer_id, campaign_resource_name):
    """Get performance report for asset groups in a campaign."""
    
    googleads_service = client.get_service("GoogleAdsService")
    
    query = f'''
        SELECT 
            asset_group.id,
            asset_group.name,
            asset_group.status,
            asset_group.path1,
            asset_group.path2,
            metrics.impressions,
            metrics.clicks,
            metrics.ctr,
            metrics.conversions,
            metrics.conversion_rate,
            metrics.cost_micros,
            metrics.cost_per_conversion
        FROM asset_group
        WHERE asset_group.campaign = '{campaign_resource_name}'
            AND segments.date DURING LAST_30_DAYS
        ORDER BY metrics.cost_micros DESC
    '''
    
    response = googleads_service.search(
        customer_id=customer_id,
        query=query
    )
    
    asset_group_performance = []
    for row in response:
        asset_group = row.asset_group
        metrics = row.metrics
        
        group_data = {
            'asset_group_id': asset_group.id,
            'asset_group_name': asset_group.name,
            'status': asset_group.status.name,
            'path1': asset_group.path1,
            'path2': asset_group.path2,
            'impressions': metrics.impressions,
            'clicks': metrics.clicks,
            'ctr': metrics.ctr,
            'conversions': metrics.conversions,
            'conversion_rate': metrics.conversion_rate,
            'cost': metrics.cost_micros / 1_000_000,
            'cost_per_conversion': metrics.cost_per_conversion / 1_000_000 if metrics.cost_per_conversion else 0
        }
        asset_group_performance.append(group_data)
    
    return asset_group_performance

# Asset types
class Asset:
    resource_name: str
    id: Optional[int]
    name: Optional[str]
    type_: AssetTypeEnum.AssetType
    final_urls: List[str]
    final_mobile_urls: List[str]
    tracking_url_template: Optional[str]
    url_custom_parameters: List[CustomParameter]
    final_url_suffix: Optional[str]
    source: AssetSourceEnum.AssetSource
    policy_summary: AssetPolicySummary
    field_type_policy_summaries: List[AssetFieldTypePolicySummary]
    
    # Asset types (one of the following)
    youtube_video_asset: YoutubeVideoAsset
    media_bundle_asset: MediaBundleAsset
    image_asset: ImageAsset
    text_asset: TextAsset
    lead_form_asset: LeadFormAsset
    book_on_google_asset: BookOnGoogleAsset
    promotion_asset: PromotionAsset
    callout_asset: CalloutAsset
    structured_snippet_asset: StructuredSnippetAsset
    sitelink_asset: SitelinkAsset
    page_feed_asset: PageFeedAsset
    dynamic_education_asset: DynamicEducationAsset
    mobile_app_asset: MobileAppAsset
    hotel_callout_asset: HotelCalloutAsset
    call_asset: CallAsset
    price_asset: PriceAsset
    call_to_action_asset: CallToActionAsset
    dynamic_real_estate_asset: DynamicRealEstateAsset
    dynamic_custom_asset: DynamicCustomAsset
    dynamic_hotels_and_rentals_asset: DynamicHotelsAndRentalsAsset
    dynamic_flights_asset: DynamicFlightsAsset
    discovery_carousel_card_asset: DiscoveryCarouselCardAsset
    dynamic_travel_asset: DynamicTravelAsset
    dynamic_local_asset: DynamicLocalAsset
    dynamic_jobs_asset: DynamicJobsAsset

class ImageAsset:
    data: bytes
    file_size: Optional[int]
    mime_type: MimeTypeEnum.MimeType
    full_size: ImageDimension

class YoutubeVideoAsset:
    youtube_video_id: Optional[str]
    youtube_video_title: Optional[str]

class TextAsset:
    text: Optional[str]

class MediaBundleAsset:
    data: Optional[bytes]

# Asset group types
class AssetGroup:
    resource_name: str
    id: Optional[int]
    campaign: Optional[str]
    name: Optional[str]
    final_urls: List[str]
    final_mobile_urls: List[str]
    status: AssetGroupStatusEnum.AssetGroupStatus
    primary_status: AssetGroupPrimaryStatusEnum.AssetGroupPrimaryStatus
    primary_status_reasons: List[AssetGroupPrimaryStatusReasonEnum.AssetGroupPrimaryStatusReason]
    path1: Optional[str]
    path2: Optional[str]
    ad_strength: AdStrengthEnum.AdStrength

class AssetGroupAsset:
    resource_name: str
    asset_group: Optional[str]
    asset: Optional[str]
    field_type: AssetFieldTypeEnum.AssetFieldType
    status: AssetLinkStatusEnum.AssetLinkStatus
    primary_status: AssetLinkPrimaryStatusEnum.AssetLinkPrimaryStatus
    primary_status_reasons: List[AssetLinkPrimaryStatusReasonEnum.AssetLinkPrimaryStatusReason]
    primary_status_details: List[AssetLinkPrimaryStatusDetails]
    performance_label: AssetPerformanceLabelEnum.AssetPerformanceLabel
    policy_summary: AssetGroupAssetPolicySummary

# Asset set types
class AssetSet:
    resource_name: str
    id: Optional[int]
    name: Optional[str]
    type_: AssetSetTypeEnum.AssetSetType
    status: AssetSetStatusEnum.AssetSetStatus
    
    # Asset set types (one of the following)
    merchant_center_feed: MerchantCenterFeed
    location_group_parent_asset_set_id: Optional[int]
    hotel_property_data: HotelPropertyAsset
    dynamic_education_feed: DynamicEducationFeed
    dynamic_real_estate_feed: DynamicRealEstateFeed
    dynamic_custom_feed: DynamicCustomFeed
    dynamic_hotels_and_rentals_feed: DynamicHotelsAndRentalsFeed
    dynamic_flights_feed: DynamicFlightsFeed
    dynamic_travel_feed: DynamicTravelFeed
    dynamic_local_feed: DynamicLocalFeed
    dynamic_jobs_feed: DynamicJobsFeed

class AssetSetAsset:
    resource_name: str
    asset_set: Optional[str]
    asset: Optional[str]
    status: AssetSetAssetStatusEnum.AssetSetAssetStatus

class MerchantCenterFeed:
    merchant_id: Optional[int]
    feed_label: Optional[str]

# Campaign asset associations
class CampaignAsset:
    resource_name: str
    campaign: Optional[str]
    asset: Optional[str]
    field_type: AssetFieldTypeEnum.AssetFieldType
    source: AssetSourceEnum.AssetSource
    status: AssetLinkStatusEnum.AssetLinkStatus
    primary_status: AssetLinkPrimaryStatusEnum.AssetLinkPrimaryStatus
    primary_status_details: List[AssetLinkPrimaryStatusDetails]
    primary_status_reasons: List[AssetLinkPrimaryStatusReasonEnum.AssetLinkPrimaryStatusReason]

class CampaignAssetSet:
    resource_name: str
    campaign: Optional[str]
    asset_set: Optional[str]
    status: AssetSetLinkStatusEnum.AssetSetLinkStatus

# Operation types
class AssetOperation:
    update_mask: FieldMask
    create: Asset
    update: Asset
    remove: str

class AssetGroupOperation:
    update_mask: FieldMask
    create: AssetGroup
    update: AssetGroup
    remove: str

class AssetGroupAssetOperation:
    update_mask: FieldMask
    create: AssetGroupAsset
    update: AssetGroupAsset
    remove: str

class AssetSetOperation:
    update_mask: FieldMask
    create: AssetSet
    update: AssetSet
    remove: str

class AssetSetAssetOperation:
    update_mask: FieldMask
    create: AssetSetAsset
    update: AssetSetAsset
    remove: str

# Response types
class MutateAssetsResponse:
    partial_failure_error: Status
    results: List[MutateAssetResult]

class MutateAssetResult:
    resource_name: str
    asset: Asset

class MutateAssetGroupsResponse:
    partial_failure_error: Status
    results: List[MutateAssetGroupResult]

class MutateAssetGroupResult:
    resource_name: str
    asset_group: AssetGroup

class MutateAssetGroupAssetsResponse:
    partial_failure_error: Status
    results: List[MutateAssetGroupAssetResult]

class MutateAssetGroupAssetResult:
    resource_name: str
    asset_group_asset: AssetGroupAsset

class MutateAssetSetsResponse:
    partial_failure_error: Status
    results: List[MutateAssetSetResult]

class MutateAssetSetResult:
    resource_name: str
    asset_set: AssetSet