diff --git a/src/adloop/ads/write.py b/src/adloop/ads/write.py index ed25b96..ac2a40d 100644 --- a/src/adloop/ads/write.py +++ b/src/adloop/ads/write.py @@ -1518,14 +1518,43 @@ def update_campaign( return preview +def _asset_scope( + ad_group_id: str, campaign_id: str, customer_id: str +) -> tuple[str, str, str]: + """Resolve asset linkage scope using most-specific-wins precedence. + + Returns ``(scope, entity_type, entity_id)`` for ChangePlan. Precedence: + ad_group > campaign > customer. CALLOUT, STRUCTURED_SNIPPET, PROMOTION, + PRICE, and CALL all support all three linkage levels per the Google Ads + API. + """ + if ad_group_id: + return "ad_group", "ad_group_asset", ad_group_id + if campaign_id: + return "campaign", "campaign_asset", campaign_id + return "customer", "customer_asset", customer_id + + def draft_callouts( config: AdLoopConfig, *, customer_id: str = "", campaign_id: str = "", + ad_group_id: str = "", callouts: list[str] | None = None, ) -> dict: - """Draft campaign callout assets.""" + """Draft callout assets — returns a PREVIEW. + + Scope (most-specific wins): + - If ``ad_group_id`` is provided, the callouts link to that ad group + via ``AdGroupAsset``. Use this for per-service callouts inside a + multi-ad-group campaign so they don't bleed onto sibling ad groups. + - Else if ``campaign_id`` is provided, the callouts link at the + campaign level via ``CampaignAsset``. + - Else, the callouts link at the customer/account level via + ``CustomerAsset`` and become available to all eligible campaigns + automatically. + """ from adloop.safety.guards import SafetyViolation, check_blocked_operation from adloop.safety.preview import ChangePlan, store_plan @@ -1534,17 +1563,22 @@ def draft_callouts( except SafetyViolation as e: return {"error": str(e)} - validated_callouts, errors = _validate_callouts(campaign_id, callouts or []) + validated_callouts, errors = _validate_callouts(callouts or []) if errors: return {"error": "Validation failed", "details": errors} + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) plan = ChangePlan( operation="create_callouts", - entity_type="campaign_asset", - entity_id=campaign_id, + entity_type=entity_type, + entity_id=entity_id, customer_id=customer_id, changes={ + "scope": scope, "campaign_id": campaign_id, + "ad_group_id": ad_group_id, "callouts": validated_callouts, }, ) @@ -1557,9 +1591,19 @@ def draft_structured_snippets( *, customer_id: str = "", campaign_id: str = "", + ad_group_id: str = "", snippets: list[dict] | None = None, ) -> dict: - """Draft campaign structured snippet assets.""" + """Draft structured snippet assets — returns a PREVIEW. + + Scope (most-specific wins): + - If ``ad_group_id`` is provided, snippets link to that ad group via + ``AdGroupAsset``. + - Else if ``campaign_id`` is provided, snippets attach at the campaign + level via ``CampaignAsset``. + - Else, snippets attach at the customer/account level and apply to all + eligible campaigns by default. + """ from adloop.safety.guards import SafetyViolation, check_blocked_operation from adloop.safety.preview import ChangePlan, store_plan @@ -1568,19 +1612,22 @@ def draft_structured_snippets( except SafetyViolation as e: return {"error": str(e)} - validated_snippets, errors = _validate_structured_snippets( - campaign_id, snippets or [] - ) + validated_snippets, errors = _validate_structured_snippets(snippets or []) if errors: return {"error": "Validation failed", "details": errors} + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) plan = ChangePlan( operation="create_structured_snippets", - entity_type="campaign_asset", - entity_id=campaign_id, + entity_type=entity_type, + entity_id=entity_id, customer_id=customer_id, changes={ + "scope": scope, "campaign_id": campaign_id, + "ad_group_id": ad_group_id, "snippets": validated_snippets, }, ) @@ -1713,29 +1760,1366 @@ def draft_sitelinks( ], } - if len(validated) < 2: - warnings.append( - "Google recommends at least 4 sitelinks per campaign. " - "Fewer than 2 may not show at all." - ) - elif len(validated) < 4: - warnings.append( - f"Only {len(validated)} sitelinks — Google recommends at least 4 for " - f"maximum ad real estate." - ) + if len(validated) < 2: + warnings.append( + "Google recommends at least 4 sitelinks per campaign. " + "Fewer than 2 may not show at all." + ) + elif len(validated) < 4: + warnings.append( + f"Only {len(validated)} sitelinks — Google recommends at least 4 for " + f"maximum ad real estate." + ) + + plan = ChangePlan( + operation="create_sitelinks", + entity_type="campaign_asset", + entity_id=campaign_id, + customer_id=customer_id, + changes={"campaign_id": campaign_id, "sitelinks": validated}, + ) + store_plan(plan) + preview = plan.to_preview() + if warnings: + preview["warnings"] = warnings + return preview + + +# --------------------------------------------------------------------------- +# Call assets + ad schedule +# --------------------------------------------------------------------------- + +# Pulled dynamically from the google-ads SDK so we don't drift when Google +# adds new enum members in a future API version. +from adloop.ads.enums import enum_names as _enum_names + +_COUNTRY_DIAL_CODES = { + "US": "+1", "CA": "+1", "GB": "+44", "DE": "+49", "FR": "+33", + "IT": "+39", "ES": "+34", "NL": "+31", "BE": "+32", "AT": "+43", + "CH": "+41", "AU": "+61", "NZ": "+64", "IE": "+353", "PT": "+351", +} + + +def _normalize_phone_e164(phone: str, country_code: str) -> tuple[str, str | None]: + """Return (normalized, error_or_None). Strips formatting, ensures + prefix. + + Handles trunk-prefix patterns: + - Already-E.164 ("+..."): returned unchanged. + - International access prefix "00": the digits after it already carry the + country code, so "0044 20..." -> "+44 20..." (country_code arg ignored). + - North America (US/CA): leading "1" before a 10-digit number is the + country code; strip it before re-adding "+1". + - European trunk "0": GB/DE/FR/ES/NL/BE/AT/CH/IE/PT/AU/NZ drop exactly ONE + domestic trunk "0" when adding the international prefix. Italy (IT) is + the exception — it KEEPS its leading 0 in E.164 form, so it is not + stripped. Only a single trunk zero is removed, never every leading zero. + """ + raw = "".join(ch for ch in phone if ch.isdigit() or ch == "+") + if not raw: + return "", "phone_number is empty after stripping formatting" + if raw.startswith("+"): + return raw, None + # "00" is the international access prefix in most of the world; the digits + # that follow already include the country code, so convert straight to "+". + if raw.startswith("00"): + return "+" + raw[2:], None + dial = _COUNTRY_DIAL_CODES.get(country_code.upper()) + if not dial: + return "", ( + f"country_code '{country_code}' is not in the dial-code map; " + f"pass phone in E.164 form (with leading '+')" + ) + cc_upper = country_code.upper() + if cc_upper in ("US", "CA") and len(raw) == 11 and raw.startswith("1"): + raw = raw[1:] + elif cc_upper not in ("US", "CA", "IT") and raw.startswith("0"): + # Strip exactly ONE domestic trunk "0" — not every leading zero. + # Italy keeps its leading 0 internationally, hence the IT exception. + raw = raw[1:] + return f"{dial}{raw}", None + + +_VALID_DAYS_OF_WEEK = { + "MONDAY", "TUESDAY", "WEDNESDAY", "THURSDAY", "FRIDAY", + "SATURDAY", "SUNDAY", +} +_VALID_MINUTES = {0, 15, 30, 45} + + +def _validate_ad_schedule( + schedule: list[dict], +) -> tuple[list[dict], list[str]]: + """Validate ad schedule entries. Returns (validated, errors). + + Each entry: {day_of_week, start_hour, end_hour, start_minute=0, end_minute=0}. + Google Ads only accepts minutes in {0, 15, 30, 45}, hours in 0-24, and + requires end > start. Day-of-week strings are normalized to upper-case. + """ + errors = [] + validated = [] + for i, entry in enumerate(schedule or []): + if not isinstance(entry, dict): + errors.append(f"ad_schedule[{i}]: must be a dict") + continue + day = str(entry.get("day_of_week", "")).strip().upper() + if day not in _VALID_DAYS_OF_WEEK: + errors.append( + f"ad_schedule[{i}]: day_of_week must be one of {sorted(_VALID_DAYS_OF_WEEK)}" + ) + continue + try: + start_hour = int(entry.get("start_hour", -1)) + end_hour = int(entry.get("end_hour", -1)) + start_minute = int(entry.get("start_minute", 0)) + end_minute = int(entry.get("end_minute", 0)) + except (TypeError, ValueError): + errors.append(f"ad_schedule[{i}]: hour/minute values must be integers") + continue + if not (0 <= start_hour <= 23): + errors.append(f"ad_schedule[{i}]: start_hour must be in 0..23") + if not (0 <= end_hour <= 24): + errors.append(f"ad_schedule[{i}]: end_hour must be in 0..24") + if start_minute not in _VALID_MINUTES: + errors.append( + f"ad_schedule[{i}]: start_minute must be one of {sorted(_VALID_MINUTES)}" + ) + if end_minute not in _VALID_MINUTES: + errors.append( + f"ad_schedule[{i}]: end_minute must be one of {sorted(_VALID_MINUTES)}" + ) + if (end_hour, end_minute) <= (start_hour, start_minute): + errors.append( + f"ad_schedule[{i}]: end ({end_hour}:{end_minute:02d}) must be after " + f"start ({start_hour}:{start_minute:02d})" + ) + validated.append({ + "day_of_week": day, + "start_hour": start_hour, + "start_minute": start_minute, + "end_hour": end_hour, + "end_minute": end_minute, + }) + return validated, errors + + +def draft_call_asset( + config: AdLoopConfig, + *, + customer_id: str = "", + phone_number: str = "", + country_code: str = "US", + campaign_id: str = "", + ad_group_id: str = "", + call_conversion_action_id: str = "", + ad_schedule: list[dict] | None = None, +) -> dict: + """Draft a call asset (phone extension) — returns a PREVIEW. + + Scope (most-specific wins): + - If ``ad_group_id`` is provided, the call asset attaches to that + ad group via ``AdGroupAsset``. Use this for per-service call + tracking inside a multi-ad-group campaign. + - Else if ``campaign_id`` is provided, the call asset attaches to + that campaign via ``CampaignAsset``. + - Else, the call asset attaches at the customer/account level via + ``CustomerAsset``. + + phone_number: human or E.164 (e.g. "+15555550142" or "(555) 555-0142"). + country_code: 2-letter ISO country code used to canonicalize a national + number to E.164. Ignored when phone_number already starts with '+'. + call_conversion_action_id: optional Google Ads conversion action ID to + count calls of qualifying duration (typically >=60 sec). When omitted, + the call asset uses the default account-level call-conversion settings. + ad_schedule: optional schedule dict list — see add_ad_schedule for shape + (day_of_week, start_hour/minute, end_hour/minute). Used to limit the + hours when the call extension shows. + + Important: Google Ads requires manual phone-number verification before + the call asset can serve. The asset is created in the account but won't + show until verification completes in the Ads UI. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("create_call_asset", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not phone_number: + return {"error": "phone_number is required"} + + normalized_phone, phone_err = _normalize_phone_e164(phone_number, country_code) + if phone_err: + return {"error": phone_err} + + schedule_validated, schedule_errors = _validate_ad_schedule(ad_schedule or []) + if schedule_errors: + return {"error": "Ad schedule validation failed", "details": schedule_errors} + + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) + + warnings = [ + "Google Ads requires phone-number verification before call assets serve. " + "Complete verification in Ads UI -> Tools -> Assets -> Calls." + ] + + plan = ChangePlan( + operation="create_call_asset", + entity_type=entity_type, + entity_id=entity_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "ad_group_id": ad_group_id, + "phone_number": normalized_phone, + "country_code": country_code.upper(), + "call_conversion_action_id": call_conversion_action_id, + "ad_schedule": schedule_validated, + }, + ) + store_plan(plan) + preview = plan.to_preview() + preview["warnings"] = warnings + return preview + + +def add_ad_schedule( + config: AdLoopConfig, + *, + customer_id: str = "", + campaign_id: str = "", + schedule: list[dict] | None = None, +) -> dict: + """Draft ad schedule additions for a campaign — returns a PREVIEW. + + Creates ``CampaignCriterion`` records of type AD_SCHEDULE so the + campaign only serves during the specified hours/days. + + schedule: list of dicts with keys: + - day_of_week: MONDAY..SUNDAY + - start_hour: 0..23 + - end_hour: 0..24 (must be > start) + - start_minute / end_minute: 0, 15, 30, or 45 (default 0) + + Note: ad-schedule hours follow the account's configured time zone. + Adding a schedule is additive — it does NOT replace existing schedule + criteria. Pause/remove existing schedule entries first if you want a + clean slate. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("add_ad_schedule", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not campaign_id: + return {"error": "campaign_id is required"} + validated, errors = _validate_ad_schedule(schedule or []) + if errors: + return {"error": "Validation failed", "details": errors} + if not validated: + return {"error": "At least one schedule entry is required"} + + plan = ChangePlan( + operation="add_ad_schedule", + entity_type="campaign_criterion", + entity_id=campaign_id, + customer_id=customer_id, + changes={ + "campaign_id": campaign_id, + "schedule": validated, + }, + ) + store_plan(plan) + return plan.to_preview() + + +# --------------------------------------------------------------------------- +# Promotion assets +# --------------------------------------------------------------------------- + +_VALID_PROMOTION_OCCASIONS = _enum_names("PromotionExtensionOccasionEnum") +_VALID_DISCOUNT_MODIFIERS = _enum_names("PromotionExtensionDiscountModifierEnum") + + +def _is_valid_iso_date(value: str) -> bool: + """True if value parses as a YYYY-MM-DD calendar date.""" + from datetime import datetime + + try: + datetime.strptime(value, "%Y-%m-%d") + return True + except (TypeError, ValueError): + return False + + +def _validate_promotion_inputs( + *, + promotion_target: str, + final_url: str, + money_off: float, + percent_off: float, + currency_code: str, + promotion_code: str, + orders_over_amount: float, + occasion: str, + discount_modifier: str, + language_code: str, + start_date: str, + end_date: str, + redemption_start_date: str, + redemption_end_date: str, + ad_schedule: list[dict] | None, +) -> tuple[dict, list[str]]: + """Validate every PromotionAsset field. Returns (normalized, errors).""" + errors: list[str] = [] + target = (promotion_target or "").strip() + url = (final_url or "").strip() + + if not target: + errors.append("promotion_target is required") + elif len(target) > 20: + errors.append( + f"promotion_target '{target}' is {len(target)} chars (max 20)" + ) + + if not url: + errors.append("final_url is required") + + has_money = money_off and money_off > 0 + has_percent = percent_off and percent_off > 0 + if has_money and has_percent: + errors.append( + "Specify exactly one of money_off or percent_off, not both" + ) + elif not has_money and not has_percent: + errors.append( + "One of money_off or percent_off is required (must be > 0)" + ) + + if has_percent and (percent_off <= 0 or percent_off > 100): + errors.append(f"percent_off must be in (0, 100]; got {percent_off}") + + code = (promotion_code or "").strip() + if code and len(code) > 15: + errors.append( + f"promotion_code '{code}' is {len(code)} chars (max 15)" + ) + + has_orders_over = bool(orders_over_amount and orders_over_amount > 0) + if code and has_orders_over: + errors.append( + "promotion_code and orders_over_amount are mutually exclusive " + "(Google Ads PromotionAsset.promotion_trigger is a oneof) — " + "specify exactly one" + ) + + occ = (occasion or "").strip().upper() + if occ and occ not in _VALID_PROMOTION_OCCASIONS: + errors.append( + f"occasion '{occ}' invalid; valid values: " + f"{sorted(_VALID_PROMOTION_OCCASIONS)}" + ) + + modifier = (discount_modifier or "").strip().upper() + if modifier and modifier not in _VALID_DISCOUNT_MODIFIERS: + errors.append( + f"discount_modifier '{modifier}' invalid; valid: " + f"{sorted(_VALID_DISCOUNT_MODIFIERS)} (or empty for none)" + ) + + for label, value in ( + ("start_date", start_date), + ("end_date", end_date), + ("redemption_start_date", redemption_start_date), + ("redemption_end_date", redemption_end_date), + ): + if value and not _is_valid_iso_date(value): + errors.append(f"{label} '{value}' must be YYYY-MM-DD") + + schedule_validated, schedule_errors = _validate_ad_schedule(ad_schedule or []) + errors.extend(schedule_errors) + + if errors: + return {}, errors + + if url: + url_checks = _validate_urls([url]) + url_err = url_checks.get(url) + if url_err: + errors.append(f"final_url '{url}' is not reachable: {url_err}") + return {}, errors + + normalized: dict = { + "promotion_target": target, + "final_url": url, + "currency_code": (currency_code or "USD").upper(), + "promotion_code": code, + "orders_over_amount": float(orders_over_amount or 0), + "occasion": occ, + "discount_modifier": modifier, + "language_code": (language_code or "en").lower(), + "start_date": start_date or "", + "end_date": end_date or "", + "redemption_start_date": redemption_start_date or "", + "redemption_end_date": redemption_end_date or "", + "ad_schedule": schedule_validated, + } + if has_money: + normalized["money_off"] = float(money_off) + normalized["percent_off"] = 0.0 + else: + normalized["money_off"] = 0.0 + normalized["percent_off"] = float(percent_off) + + return normalized, [] + + +def draft_promotion( + config: AdLoopConfig, + *, + customer_id: str = "", + promotion_target: str = "", + final_url: str = "", + money_off: float = 0, + percent_off: float = 0, + currency_code: str = "USD", + promotion_code: str = "", + orders_over_amount: float = 0, + occasion: str = "", + discount_modifier: str = "", + language_code: str = "en", + start_date: str = "", + end_date: str = "", + redemption_start_date: str = "", + redemption_end_date: str = "", + campaign_id: str = "", + ad_group_id: str = "", + ad_schedule: list[dict] | None = None, +) -> dict: + """Draft a promotion extension asset — returns a PREVIEW. + + Creates a PromotionAsset and links it at ad group, campaign, or customer + scope. Exactly one of money_off / percent_off must be provided. + + Scope (most-specific wins): + - ad_group_id provided → AdGroupAsset link (per-service promo inside + a multi-ad-group campaign). + - campaign_id provided → CampaignAsset link. + - both empty → CustomerAsset link (account-level). + + Required: + promotion_target: what the promotion is for, e.g. "Spring Sale" + (max 20 chars; this is the label Google shows in the ad). + final_url: landing page for the promotion (must return 2xx/3xx). + money_off OR percent_off: the discount amount. + + Optional: + currency_code: ISO 4217 (default USD). Used for money_off and + orders_over_amount. + promotion_code: optional coupon code (max 15 chars). + orders_over_amount: minimum order amount that unlocks the promo. + occasion: optional event tag — e.g. BLACK_FRIDAY, SUMMER_SALE. + discount_modifier: optional modifier; "UP_TO" surfaces as + "Up to $X off" instead of "$X off". + language_code: BCP-47 (default "en"). + start_date / end_date: YYYY-MM-DD. Leave blank for always-on. + redemption_start_date / redemption_end_date: YYYY-MM-DD. + ad_schedule: optional list of {day_of_week, start_hour, end_hour, + start_minute, end_minute} entries restricting when the promo shows. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("create_promotion", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + normalized, errors = _validate_promotion_inputs( + promotion_target=promotion_target, + final_url=final_url, + money_off=money_off, + percent_off=percent_off, + currency_code=currency_code, + promotion_code=promotion_code, + orders_over_amount=orders_over_amount, + occasion=occasion, + discount_modifier=discount_modifier, + language_code=language_code, + start_date=start_date, + end_date=end_date, + redemption_start_date=redemption_start_date, + redemption_end_date=redemption_end_date, + ad_schedule=ad_schedule, + ) + if errors: + return {"error": "Validation failed", "details": errors} + + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) + plan = ChangePlan( + operation="create_promotion", + entity_type=entity_type, + entity_id=entity_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "ad_group_id": ad_group_id, + "promotion": normalized, + }, + ) + store_plan(plan) + return plan.to_preview() + + +def update_promotion( + config: AdLoopConfig, + *, + customer_id: str = "", + asset_id: str = "", + campaign_id: str = "", + promotion_target: str = "", + final_url: str = "", + money_off: float = 0, + percent_off: float = 0, + currency_code: str = "USD", + promotion_code: str = "", + orders_over_amount: float = 0, + occasion: str = "", + discount_modifier: str = "", + language_code: str = "en", + start_date: str = "", + end_date: str = "", + redemption_start_date: str = "", + redemption_end_date: str = "", + ad_schedule: list[dict] | None = None, +) -> dict: + """Update a promotion via swap — returns a PREVIEW. + + PromotionAsset fields are immutable once created in the Google Ads API, so + "update" is implemented as a swap: + 1. Create a new PromotionAsset with the updated values. + 2. Link the new asset at the same scope. + 3. Unlink the old asset. + + The old Asset row itself stays in the account (orphaned). The Ads API does + not support hard-deleting Asset rows; Google reclaims orphaned assets in + due course. + + asset_id: numeric ID of the existing PromotionAsset to replace. + campaign_id: pass to scope BOTH the new and old links to that campaign. + Leave empty for customer/account-level scope. + + All other fields: see draft_promotion docstring. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_promotion", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not asset_id: + return {"error": "asset_id is required (the existing PromotionAsset to replace)"} + + normalized, errors = _validate_promotion_inputs( + promotion_target=promotion_target, + final_url=final_url, + money_off=money_off, + percent_off=percent_off, + currency_code=currency_code, + promotion_code=promotion_code, + orders_over_amount=orders_over_amount, + occasion=occasion, + discount_modifier=discount_modifier, + language_code=language_code, + start_date=start_date, + end_date=end_date, + redemption_start_date=redemption_start_date, + redemption_end_date=redemption_end_date, + ad_schedule=ad_schedule, + ) + if errors: + return {"error": "Validation failed", "details": errors} + + scope = "campaign" if campaign_id else "customer" + warnings = [ + "Update is a swap: a new PromotionAsset is created and linked, " + "the old link is unlinked. The old Asset row stays in the account " + "(orphaned) — Google Ads API does not support deleting Asset rows." + ] + + plan = ChangePlan( + operation="update_promotion", + entity_type="campaign_asset" if scope == "campaign" else "customer_asset", + entity_id=campaign_id or customer_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "old_asset_id": asset_id, + "promotion": normalized, + }, + ) + store_plan(plan) + preview = plan.to_preview() + preview["warnings"] = warnings + return preview + + +# --------------------------------------------------------------------------- +# Price assets +# --------------------------------------------------------------------------- + +_VALID_PRICE_TYPES = _enum_names("PriceExtensionTypeEnum") +_VALID_PRICE_QUALIFIERS = _enum_names("PriceExtensionPriceQualifierEnum") +_VALID_PRICE_UNITS = _enum_names("PriceExtensionPriceUnitEnum") + +# Google Ads requires between 3 and 8 price offerings per price asset. +_PRICE_OFFERING_MIN = 3 +_PRICE_OFFERING_MAX = 8 +# Header and description are each capped at 25 characters by Google. +_PRICE_TEXT_MAX = 25 + + +def _validate_price_inputs( + *, + price_type: str, + price_qualifier: str, + language_code: str, + currency_code: str, + offerings: list[dict] | None, +) -> tuple[dict, list[str]]: + """Validate every PriceAsset field. Returns (normalized, errors). + + A price asset needs a type (e.g. SERVICES), an optional FROM/UP_TO/AVERAGE + qualifier, and 3-8 price offerings. Each offering needs a header (<=25), + a description (<=25), a price amount (> 0), and a reachable final_url. + """ + errors: list[str] = [] + + ptype = (price_type or "").strip().upper() + if not ptype: + errors.append("price_type is required (e.g. SERVICES, BRANDS, EVENTS)") + elif ptype not in _VALID_PRICE_TYPES: + errors.append( + f"price_type '{ptype}' invalid; valid values: " + f"{sorted(_VALID_PRICE_TYPES)}" + ) + + qualifier = (price_qualifier or "").strip().upper() + if qualifier and qualifier not in _VALID_PRICE_QUALIFIERS: + errors.append( + f"price_qualifier '{qualifier}' invalid; valid: " + f"{sorted(_VALID_PRICE_QUALIFIERS)} (or empty for none)" + ) + + rows = offerings or [] + if not (_PRICE_OFFERING_MIN <= len(rows) <= _PRICE_OFFERING_MAX): + errors.append( + f"price asset needs between {_PRICE_OFFERING_MIN} and " + f"{_PRICE_OFFERING_MAX} offerings; got {len(rows)}" + ) + + normalized_offerings: list[dict] = [] + seen_headers: set[str] = set() + urls_to_check: list[str] = [] + for i, row in enumerate(rows): + label = f"offering[{i}]" + header = str(row.get("header", "")).strip() + description = str(row.get("description", "")).strip() + url = str(row.get("final_url", "")).strip() + mobile_url = str(row.get("final_mobile_url", "")).strip() + unit = str(row.get("unit", "")).strip().upper() + + if not header: + errors.append(f"{label}: header is required") + elif len(header) > _PRICE_TEXT_MAX: + errors.append( + f"{label}: header '{header}' is {len(header)} chars " + f"(max {_PRICE_TEXT_MAX})" + ) + # Google rejects a price asset with two identical offering headers. + key = header.lower() + if key and key in seen_headers: + errors.append(f"{label}: duplicate header '{header}'") + seen_headers.add(key) + + if not description: + errors.append(f"{label}: description is required") + elif len(description) > _PRICE_TEXT_MAX: + errors.append( + f"{label}: description '{description}' is " + f"{len(description)} chars (max {_PRICE_TEXT_MAX})" + ) + + try: + price = float(row.get("price", 0)) + except (TypeError, ValueError): + price = 0.0 + errors.append(f"{label}: price must be a number") + if price <= 0: + errors.append(f"{label}: price must be > 0") + + if not url: + errors.append(f"{label}: final_url is required") + else: + urls_to_check.append(url) + if mobile_url: + urls_to_check.append(mobile_url) + + if unit and unit not in _VALID_PRICE_UNITS: + errors.append( + f"{label}: unit '{unit}' invalid; valid: " + f"{sorted(_VALID_PRICE_UNITS)} (or empty for none)" + ) + + normalized_offerings.append( + { + "header": header, + "description": description, + "price": price, + "final_url": url, + "final_mobile_url": mobile_url, + "unit": unit, + } + ) + + if errors: + return {}, errors + + # Only hit the network once validation has otherwise passed. + if urls_to_check: + url_checks = _validate_urls(list(dict.fromkeys(urls_to_check))) + for url, err in url_checks.items(): + if err: + errors.append(f"final_url '{url}' is not reachable: {err}") + if errors: + return {}, errors + + normalized = { + "price_type": ptype, + "price_qualifier": qualifier, + "language_code": (language_code or "en").lower(), + "currency_code": (currency_code or "USD").upper(), + "offerings": normalized_offerings, + } + return normalized, [] + + +def draft_price_asset( + config: AdLoopConfig, + *, + customer_id: str = "", + campaign_id: str = "", + ad_group_id: str = "", + price_type: str = "SERVICES", + price_qualifier: str = "FROM", + language_code: str = "en", + currency_code: str = "USD", + offerings: list[dict] | None = None, +) -> dict: + """Draft a price extension asset — returns a PREVIEW. + + Creates a PriceAsset (the price carousel that appears under an ad) and + links it at ad group, campaign, or customer scope. Exactly 3-8 offerings + are required by Google Ads. + + Scope (most-specific wins): + - ad_group_id provided → AdGroupAsset link. + - campaign_id provided → CampaignAsset link. + - both empty → CustomerAsset link (account-level). + + Required: + price_type: what the prices describe — SERVICES, BRANDS, EVENTS, + LOCATIONS, NEIGHBORHOODS, PRODUCT_CATEGORIES, PRODUCT_TIERS, + SERVICE_CATEGORIES, SERVICE_TIERS. + offerings: list of 3-8 dicts, each with: + - header (str, <=25 chars, required, unique) + - description (str, <=25 chars, required) + - price (number, > 0, required) — in ``currency_code`` + - final_url (str, required) — must return 2xx/3xx (validated) + - final_mobile_url (str, optional) + - unit (str, optional) — PER_HOUR, PER_DAY, PER_WEEK, PER_MONTH, + PER_YEAR, PER_NIGHT + + Optional: + price_qualifier: FROM (default), UP_TO, or AVERAGE — leave blank for + none. "FROM" surfaces "From $X". + language_code: BCP-47 (default "en"). + currency_code: ISO 4217 (default USD) — applied to every offering. + + IMPORTANT: every price MUST match the price shown on its final_url landing + page. Google disapproves price assets whose amounts don't match the page. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("create_price_asset", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + normalized, errors = _validate_price_inputs( + price_type=price_type, + price_qualifier=price_qualifier, + language_code=language_code, + currency_code=currency_code, + offerings=offerings, + ) + if errors: + return {"error": "Validation failed", "details": errors} + + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) + plan = ChangePlan( + operation="create_price_asset", + entity_type=entity_type, + entity_id=entity_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "ad_group_id": ad_group_id, + "price": normalized, + }, + ) + store_plan(plan) + return plan.to_preview() + + +def update_structured_snippet( + config: AdLoopConfig, + *, + customer_id: str = "", + asset_id: str = "", + campaign_id: str = "", + ad_group_id: str = "", + header: str = "", + values: list[str] | None = None, +) -> dict: + """Update a structured snippet via swap — returns a PREVIEW. + + StructuredSnippetAsset fields (header + values) are immutable once created, + so "update" is a swap (same pattern as update_promotion): + 1. Create a new StructuredSnippetAsset with the updated header/values. + 2. Link the new asset at the same scope. + 3. Unlink the old asset link. + + The old Asset row stays in the account (orphaned) — Google reclaims it. + + asset_id: numeric ID of the existing StructuredSnippetAsset to replace. + + Scope (most-specific wins) — must match where the OLD asset is linked so + the swap unlinks the right link: + - ad_group_id provided → AdGroupAsset. + - campaign_id provided → CampaignAsset. + - both empty → CustomerAsset (account-level). + + header: official structured-snippet header (e.g. "Brands", "Services"). + values: 3-10 values, each <= 25 chars. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_structured_snippet", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not asset_id: + return { + "error": "asset_id is required (the existing " + "StructuredSnippetAsset to replace)" + } + + validated, errors = _validate_structured_snippets( + [{"header": header, "values": values or []}] + ) + if errors: + return {"error": "Validation failed", "details": errors} + + scope, entity_type, entity_id = _asset_scope( + ad_group_id, campaign_id, customer_id + ) + warnings = [ + "Update is a swap: a new StructuredSnippetAsset is created and " + "linked, the old link is unlinked. The old Asset row stays in the " + "account (orphaned) — Google Ads API does not support deleting " + "Asset rows." + ] + + plan = ChangePlan( + operation="update_structured_snippet", + entity_type=entity_type, + entity_id=entity_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "ad_group_id": ad_group_id, + "old_asset_id": asset_id, + "snippet": validated[0], + }, + ) + store_plan(plan) + preview = plan.to_preview() + preview["warnings"] = warnings + return preview + + +# --------------------------------------------------------------------------- +# In-place asset updates (call asset, sitelink, callout) +# --------------------------------------------------------------------------- + +_VALID_CALL_REPORTING_STATES = _enum_names("CallConversionReportingStateEnum") + + +def update_call_asset( + config: AdLoopConfig, + *, + customer_id: str = "", + asset_id: str = "", + phone_number: str = "", + country_code: str = "", + call_conversion_action_id: str = "", + call_conversion_reporting_state: str = "", + ad_schedule: list[dict] | None = None, +) -> dict: + """Update an existing CallAsset in place — returns a PREVIEW. + + Use this to: + - re-point a CallAsset at a specific conversion action with + USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION + - change the phone number / country code + - replace the ad-schedule windows + + Pass only the fields you want to change. Empty strings/None are treated as + "do not change". + + asset_id: numeric ID of the existing call asset. + call_conversion_reporting_state: one of DISABLED | + USE_ACCOUNT_LEVEL_CALL_CONVERSION_ACTION | + USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_call_asset", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not asset_id: + return {"error": "asset_id is required"} + + errors: list[str] = [] + normalized_phone = "" + if phone_number: + cc = (country_code or "US").upper() + normalized_phone, phone_err = _normalize_phone_e164(phone_number, cc) + if phone_err: + errors.append(phone_err) + + if call_conversion_reporting_state and ( + call_conversion_reporting_state not in _VALID_CALL_REPORTING_STATES + ): + errors.append( + f"call_conversion_reporting_state '{call_conversion_reporting_state}'" + f" invalid; valid: {sorted(_VALID_CALL_REPORTING_STATES)}" + ) + + schedule_validated, schedule_errors = _validate_ad_schedule(ad_schedule or []) + errors.extend(schedule_errors) + + if errors: + return {"error": "Validation failed", "details": errors} + + changes: dict = {"asset_id": str(asset_id)} + if normalized_phone: + changes["phone_number"] = normalized_phone + if country_code: + changes["country_code"] = country_code.upper() + if call_conversion_action_id: + changes["call_conversion_action_id"] = str(call_conversion_action_id) + if call_conversion_reporting_state: + changes["call_conversion_reporting_state"] = call_conversion_reporting_state + if ad_schedule is not None: + changes["ad_schedule"] = schedule_validated + + if len(changes) == 1: + return {"error": "No fields to update"} + + plan = ChangePlan( + operation="update_call_asset", + entity_type="asset", + entity_id=str(asset_id), + customer_id=customer_id, + changes=changes, + ) + store_plan(plan) + return plan.to_preview() + + +def update_sitelink( + config: AdLoopConfig, + *, + customer_id: str = "", + asset_id: str = "", + link_text: str = "", + final_url: str = "", + description1: str = "", + description2: str = "", +) -> dict: + """Update an existing SitelinkAsset in place — returns a PREVIEW. + + Pass only the fields you want to change. Empty string = "do not change". + + asset_id: numeric ID of the existing sitelink asset. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_sitelink", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not asset_id: + return {"error": "asset_id is required"} + + errors: list[str] = [] + if link_text and len(link_text) > 25: + errors.append( + f"link_text '{link_text}' is {len(link_text)} chars (max 25)" + ) + if description1 and len(description1) > 35: + errors.append( + f"description1 is {len(description1)} chars (max 35)" + ) + if description2 and len(description2) > 35: + errors.append( + f"description2 is {len(description2)} chars (max 35)" + ) + if errors: + return {"error": "Validation failed", "details": errors} + + if final_url: + url_checks = _validate_urls([final_url]) + url_err = url_checks.get(final_url) + if url_err: + return { + "error": "URL validation failed", + "details": [f"'{final_url}' is not reachable: {url_err}"], + } + + changes: dict = {"asset_id": str(asset_id)} + if link_text: + changes["link_text"] = link_text + if final_url: + changes["final_url"] = final_url + if description1: + changes["description1"] = description1 + if description2: + changes["description2"] = description2 + + if len(changes) == 1: + return {"error": "No fields to update"} + + plan = ChangePlan( + operation="update_sitelink", + entity_type="asset", + entity_id=str(asset_id), + customer_id=customer_id, + changes=changes, + ) + store_plan(plan) + return plan.to_preview() + + +def update_callout( + config: AdLoopConfig, + *, + customer_id: str = "", + asset_id: str = "", + callout_text: str = "", +) -> dict: + """Update an existing CalloutAsset's text in place — returns a PREVIEW. + + asset_id: numeric ID of the existing callout asset. + callout_text: new callout text (max 25 chars). + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("update_callout", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not asset_id: + return {"error": "asset_id is required"} + text = (callout_text or "").strip() + if not text: + return {"error": "callout_text is required"} + if len(text) > 25: + return { + "error": "Validation failed", + "details": [f"callout_text is {len(text)} chars (max 25)"], + } + + plan = ChangePlan( + operation="update_callout", + entity_type="asset", + entity_id=str(asset_id), + customer_id=customer_id, + changes={"asset_id": str(asset_id), "callout_text": text}, + ) + store_plan(plan) + return plan.to_preview() + + +# --------------------------------------------------------------------------- +# Location + business-name assets +# --------------------------------------------------------------------------- + + +def draft_location_asset( + config: AdLoopConfig, + *, + customer_id: str = "", + business_profile_account_id: str = "", + asset_set_name: str = "", + campaign_id: str = "", + label_filters: list[str] | None = None, + listing_id_filters: list[str] | None = None, +) -> dict: + """Draft a Google Business Profile-backed location AssetSet — PREVIEW. + + Creates an ``AssetSet`` of type LOCATION_SYNC that pulls locations from a + linked Google Business Profile and exposes them as location assets. The + set is attached at the customer level (so all eligible campaigns get it by + default). Optionally also creates a ``CampaignAssetSet`` link to a specific + campaign. + + Required preflight: the Google Business Profile must already be linked in + Google Ads → Tools → Linked accounts → Business Profile. + + business_profile_account_id: numeric Business Profile (LBC) account ID. + asset_set_name: optional name for the AssetSet. Defaults to + "GBP Locations - ". + label_filters: optional list of GBP location labels to limit sync. + listing_id_filters: optional list of GBP listing IDs to limit sync. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("create_location_asset", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not business_profile_account_id: + return { + "error": ( + "business_profile_account_id is required (numeric GBP/LBC account ID). " + "Find it in Google Business Profile admin." + ) + } + + name = asset_set_name or f"GBP Locations - {business_profile_account_id}" + warnings = [ + "The Google Business Profile must already be linked at Tools -> Linked " + "accounts -> Business Profile in Google Ads. If it isn't, this tool " + "will fail at apply time." + ] + + plan = ChangePlan( + operation="create_location_asset", + entity_type="asset_set", + entity_id=customer_id, + customer_id=customer_id, + changes={ + "scope": "campaign" if campaign_id else "customer", + "campaign_id": campaign_id, + "business_profile_account_id": str(business_profile_account_id), + "asset_set_name": name, + "label_filters": list(label_filters or []), + "listing_id_filters": [str(x) for x in (listing_id_filters or [])], + }, + ) + store_plan(plan) + preview = plan.to_preview() + preview["warnings"] = warnings + return preview + + +def draft_business_name_asset( + config: AdLoopConfig, + *, + customer_id: str = "", + campaign_id: str = "", + business_name: str = "", +) -> dict: + """Draft a business-name asset — returns a PREVIEW. + + Creates a TEXT asset and links it as ``BUSINESS_NAME`` at customer or + campaign scope. Google shows the business name alongside ads so users can + recognize the brand at a glance. + + Scope: + - If ``campaign_id`` is empty (default), the asset is linked at the + customer/account level via CustomerAsset. + - If ``campaign_id`` is provided, the asset is scoped to that single + campaign via CampaignAsset. + + business_name: max 25 characters per Google Ads policy. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("create_business_name_asset", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + text = (business_name or "").strip() + if not text: + return {"error": "business_name is required"} + if len(text) > 25: + return { + "error": "Validation failed", + "details": [ + f"business_name '{text}' is {len(text)} chars (max 25)" + ], + } + + scope = "campaign" if campaign_id else "customer" + plan = ChangePlan( + operation="create_business_name_asset", + entity_type="campaign_asset" if scope == "campaign" else "customer_asset", + entity_id=campaign_id or customer_id, + customer_id=customer_id, + changes={ + "scope": scope, + "campaign_id": campaign_id, + "business_name": text, + }, + ) + store_plan(plan) + return plan.to_preview() + + +# --------------------------------------------------------------------------- +# link_asset_to_customer — promote existing assets to customer/account scope +# --------------------------------------------------------------------------- + +# AssetFieldType values that are valid for CustomerAsset (account-level). +_VALID_CUSTOMER_ASSET_FIELD_TYPES = { + "SITELINK", "CALLOUT", "STRUCTURED_SNIPPET", "PROMOTION", "PRICE", + "CALL", "MOBILE_APP", "HOTEL_CALLOUT", "BUSINESS_LOGO", "BUSINESS_NAME", + "AD_IMAGE", "MARKETING_IMAGE", "SQUARE_MARKETING_IMAGE", + "PORTRAIT_MARKETING_IMAGE", "LOGO", "LANDSCAPE_LOGO", + "YOUTUBE_VIDEO", "MEDIA_BUNDLE", "BOOK_ON_GOOGLE", "LEAD_FORM", + "HEADLINE", "DESCRIPTION", "LONG_HEADLINE", +} + + +def link_asset_to_customer( + config: AdLoopConfig, + *, + customer_id: str = "", + links: list[dict] | None = None, +) -> dict: + """Link EXISTING assets to the customer (account) — returns a PREVIEW. + + Use this to "promote" assets that already exist in the account (typically + attached to legacy campaigns) so they apply at the account level and + inherit to every eligible campaign automatically. + + Unlike draft_image_assets / draft_callouts / etc., this tool does NOT + create new Asset rows — it only adds CustomerAsset link rows pointing to + assets you already have. Find candidate asset_ids via: + SELECT asset.id, asset.type, asset.name FROM asset + + links: list of dicts, each with: + - asset_id (str, required) — numeric asset ID + - field_type (str, required) — AssetFieldType, e.g. BUSINESS_LOGO, + AD_IMAGE, MARKETING_IMAGE, BUSINESS_NAME, SITELINK, CALLOUT, CALL, + PROMOTION, etc. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("link_asset_to_customer", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not links: + return {"error": "At least one link is required"} + + errors: list[str] = [] + validated: list[dict] = [] + for i, item in enumerate(links): + if not isinstance(item, dict): + errors.append(f"Link {i + 1}: must be a dict, got {type(item).__name__}") + continue + asset_id = str(item.get("asset_id", "")).strip() + field_type = str(item.get("field_type", "")).strip().upper() + if not asset_id: + errors.append(f"Link {i + 1}: asset_id is required") + continue + if not asset_id.isdigit(): + errors.append( + f"Link {i + 1}: asset_id '{asset_id}' must be numeric" + ) + continue + if not field_type: + errors.append(f"Link {i + 1}: field_type is required") + continue + if field_type not in _VALID_CUSTOMER_ASSET_FIELD_TYPES: + errors.append( + f"Link {i + 1}: field_type '{field_type}' is not valid for " + f"CustomerAsset; valid: " + f"{sorted(_VALID_CUSTOMER_ASSET_FIELD_TYPES)}" + ) + continue + validated.append({"asset_id": asset_id, "field_type": field_type}) + + if errors: + return {"error": "Validation failed", "details": errors} plan = ChangePlan( - operation="create_sitelinks", - entity_type="campaign_asset", - entity_id=campaign_id, + operation="link_asset_to_customer", + entity_type="customer_asset", + entity_id=customer_id, customer_id=customer_id, - changes={"campaign_id": campaign_id, "sitelinks": validated}, + changes={"links": validated}, ) store_plan(plan) - preview = plan.to_preview() - if warnings: - preview["warnings"] = warnings - return preview + return plan.to_preview() # --------------------------------------------------------------------------- @@ -1998,13 +3382,11 @@ def _ad_group_campaign_bidding_strategy( def _validate_callouts( - campaign_id: str, callouts: list[str] + callouts: list[str], ) -> tuple[list[str], list[str]]: errors = [] validated = [] - if not campaign_id: - errors.append("campaign_id is required") if not callouts: errors.append("At least one callout is required") @@ -2023,13 +3405,11 @@ def _validate_callouts( def _validate_structured_snippets( - campaign_id: str, snippets: list[dict] + snippets: list[dict], ) -> tuple[list[dict], list[str]]: errors = [] validated = [] - if not campaign_id: - errors.append("campaign_id is required") if not snippets: errors.append("At least one structured snippet is required") @@ -2554,6 +3934,18 @@ def _execute_plan(config: AdLoopConfig, plan: object) -> dict: "create_structured_snippets": _apply_create_structured_snippets, "create_image_assets": _apply_create_image_assets, "create_sitelinks": _apply_create_sitelinks, + "create_call_asset": _apply_create_call_asset, + "add_ad_schedule": _apply_add_ad_schedule, + "create_promotion": _apply_create_promotion, + "create_price_asset": _apply_create_price_asset, + "update_promotion": _apply_update_promotion, + "update_structured_snippet": _apply_update_structured_snippet, + "update_call_asset": _apply_update_call_asset, + "update_sitelink": _apply_update_sitelink, + "update_callout": _apply_update_callout, + "create_location_asset": _apply_create_location_asset, + "create_business_name_asset": _apply_create_business_name_asset, + "link_asset_to_customer": _apply_link_asset_to_customer, } handler = dispatch.get(plan.operation) @@ -3363,7 +4755,41 @@ def _apply_campaign_assets( field_type: object, populate_asset: object, ) -> dict: - """Create assets and link them to a campaign via CampaignAsset.""" + """Create assets and link them to a campaign via CampaignAsset (legacy alias).""" + return _apply_assets( + client, + cid, + assets, + field_type, + populate_asset, + scope="campaign", + campaign_id=campaign_id, + ) + + +def _apply_assets( + client: object, + cid: str, + assets: list[dict], + field_type: object, + populate_asset: object, + *, + scope: str = "campaign", + campaign_id: str = "", + ad_group_id: str = "", +) -> dict: + """Create Asset rows + link them at ad group, campaign, or customer scope. + + scope: + - "ad_group" → AdGroupAsset (requires ad_group_id). Use for + per-service assets inside a multi-ad-group campaign so they don't + bleed onto sibling ad groups. CALLOUT, STRUCTURED_SNIPPET, + PROMOTION, PRICE, and CALL all support AdGroupAsset linking per the + Google Ads API. + - "campaign" → CampaignAsset (requires campaign_id) + - "customer" → CustomerAsset (account-level, applies to all eligible + campaigns by default) + """ asset_service = client.get_service("AssetService") googleads_service = client.get_service("GoogleAdsService") operations = [] @@ -3375,69 +4801,108 @@ def _apply_campaign_assets( populate_asset(asset, payload) operations.append(op) - for i in range(len(assets)): - op = client.get_type("MutateOperation") - ca = op.campaign_asset_operation.create - ca.asset = asset_service.asset_path(cid, str(-(i + 1))) - ca.campaign = googleads_service.campaign_path(cid, campaign_id) - ca.field_type = field_type - operations.append(op) + if scope == "ad_group": + if not ad_group_id: + raise ValueError("ad_group_id required for ad_group-scope assets") + for i in range(len(assets)): + op = client.get_type("MutateOperation") + aga = op.ad_group_asset_operation.create + aga.asset = asset_service.asset_path(cid, str(-(i + 1))) + aga.ad_group = googleads_service.ad_group_path(cid, ad_group_id) + aga.field_type = field_type + operations.append(op) + elif scope == "campaign": + if not campaign_id: + raise ValueError("campaign_id is required for campaign-scope assets") + for i in range(len(assets)): + op = client.get_type("MutateOperation") + ca = op.campaign_asset_operation.create + ca.asset = asset_service.asset_path(cid, str(-(i + 1))) + ca.campaign = googleads_service.campaign_path(cid, campaign_id) + ca.field_type = field_type + operations.append(op) + elif scope == "customer": + for i in range(len(assets)): + op = client.get_type("MutateOperation") + cust_asset = op.customer_asset_operation.create + cust_asset.asset = asset_service.asset_path(cid, str(-(i + 1))) + cust_asset.field_type = field_type + operations.append(op) + else: + raise ValueError(f"Unknown asset scope: {scope}") response = googleads_service.mutate( customer_id=cid, mutate_operations=operations ) - results = {"assets": [], "campaign_assets": []} + if scope == "ad_group": + results = {"assets": [], "ad_group_assets": []} + link_key = "ad_group_assets" + elif scope == "campaign": + results = {"assets": [], "campaign_assets": []} + link_key = "campaign_assets" + else: + results = {"assets": [], "customer_assets": []} + link_key = "customer_assets" + num_assets = len(assets) for i, resp in enumerate(response.mutate_operation_responses): resource = None if resp.asset_result.resource_name: resource = resp.asset_result.resource_name - elif resp.campaign_asset_result.resource_name: + elif scope == "ad_group" and resp.ad_group_asset_result.resource_name: + resource = resp.ad_group_asset_result.resource_name + elif scope == "campaign" and resp.campaign_asset_result.resource_name: resource = resp.campaign_asset_result.resource_name + elif scope == "customer" and resp.customer_asset_result.resource_name: + resource = resp.customer_asset_result.resource_name if resource: if i < num_assets: results["assets"].append(resource) else: - results["campaign_assets"].append(resource) + results[link_key].append(resource) return results def _apply_create_callouts(client: object, cid: str, changes: dict) -> dict: - """Create callout assets and link them to a campaign.""" + """Create callout assets at ad group, campaign, or customer scope.""" def populate(asset: object, payload: dict) -> None: asset.callout_asset.callout_text = payload["callout_text"] assets = [{"callout_text": text} for text in changes["callouts"]] - return _apply_campaign_assets( + return _apply_assets( client, cid, - changes["campaign_id"], assets, client.enums.AssetFieldTypeEnum.CALLOUT, populate, + scope=changes.get("scope", "campaign"), + campaign_id=changes.get("campaign_id", ""), + ad_group_id=changes.get("ad_group_id", ""), ) def _apply_create_structured_snippets( client: object, cid: str, changes: dict ) -> dict: - """Create structured snippet assets and link them to a campaign.""" + """Create structured snippet assets at ad group, campaign, or customer scope.""" def populate(asset: object, payload: dict) -> None: asset.structured_snippet_asset.header = payload["header"] asset.structured_snippet_asset.values.extend(payload["values"]) - return _apply_campaign_assets( + return _apply_assets( client, cid, - changes["campaign_id"], changes["snippets"], client.enums.AssetFieldTypeEnum.STRUCTURED_SNIPPET, populate, + scope=changes.get("scope", "campaign"), + campaign_id=changes.get("campaign_id", ""), + ad_group_id=changes.get("ad_group_id", ""), ) @@ -3486,6 +4951,713 @@ def populate(asset: object, payload: dict) -> None: ) +_AD_SCHEDULE_DAY_ENUM = { + "MONDAY": "MONDAY", + "TUESDAY": "TUESDAY", + "WEDNESDAY": "WEDNESDAY", + "THURSDAY": "THURSDAY", + "FRIDAY": "FRIDAY", + "SATURDAY": "SATURDAY", + "SUNDAY": "SUNDAY", +} +_MINUTE_TO_ENUM = {0: "ZERO", 15: "FIFTEEN", 30: "THIRTY", 45: "FORTY_FIVE"} + + +def _populate_ad_schedule_info(client: object, info: object, entry: dict) -> None: + """Set fields on an AdScheduleInfo proto from a validated entry.""" + info.day_of_week = getattr( + client.enums.DayOfWeekEnum, _AD_SCHEDULE_DAY_ENUM[entry["day_of_week"]] + ) + info.start_hour = int(entry["start_hour"]) + info.end_hour = int(entry["end_hour"]) + info.start_minute = getattr( + client.enums.MinuteOfHourEnum, _MINUTE_TO_ENUM[int(entry["start_minute"])] + ) + info.end_minute = getattr( + client.enums.MinuteOfHourEnum, _MINUTE_TO_ENUM[int(entry["end_minute"])] + ) + + +def _apply_add_ad_schedule(client: object, cid: str, changes: dict) -> dict: + """Add AdScheduleInfo CampaignCriterion records to a campaign.""" + campaign_service = client.get_service("CampaignService") + crit_service = client.get_service("CampaignCriterionService") + operations = [] + for entry in changes["schedule"]: + op = client.get_type("CampaignCriterionOperation") + crit = op.create + crit.campaign = campaign_service.campaign_path( + cid, changes["campaign_id"] + ) + _populate_ad_schedule_info(client, crit.ad_schedule, entry) + operations.append(op) + response = crit_service.mutate_campaign_criteria( + customer_id=cid, operations=operations + ) + return { + "campaign_criteria": [r.resource_name for r in response.results], + } + + +def _apply_create_call_asset(client: object, cid: str, changes: dict) -> dict: + """Create a CallAsset at ad group, campaign, or customer scope.""" + asset_service = client.get_service("AssetService") + googleads_service = client.get_service("GoogleAdsService") + operations = [] + + op = client.get_type("MutateOperation") + asset = op.asset_operation.create + asset.resource_name = asset_service.asset_path(cid, "-1") + asset.call_asset.country_code = changes["country_code"] + asset.call_asset.phone_number = changes["phone_number"] + if changes.get("call_conversion_action_id"): + ca_service = client.get_service("ConversionActionService") + asset.call_asset.call_conversion_action = ca_service.conversion_action_path( + cid, str(changes["call_conversion_action_id"]) + ) + asset.call_asset.call_conversion_reporting_state = ( + client.enums.CallConversionReportingStateEnum.USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION + ) + for entry in changes.get("ad_schedule") or []: + info = client.get_type("AdScheduleInfo") + _populate_ad_schedule_info(client, info, entry) + asset.call_asset.ad_schedule_targets.append(info) + operations.append(op) + + scope = changes.get("scope", "campaign") + if scope == "ad_group": + if not changes.get("ad_group_id"): + raise ValueError("ad_group_id required for ad_group-scope call asset") + link_op = client.get_type("MutateOperation") + ag_asset = link_op.ad_group_asset_operation.create + ag_asset.asset = asset_service.asset_path(cid, "-1") + ag_asset.ad_group = googleads_service.ad_group_path( + cid, changes["ad_group_id"] + ) + ag_asset.field_type = client.enums.AssetFieldTypeEnum.CALL + operations.append(link_op) + elif scope == "campaign": + if not changes.get("campaign_id"): + raise ValueError("campaign_id required for campaign-scope call asset") + link_op = client.get_type("MutateOperation") + ca = link_op.campaign_asset_operation.create + ca.asset = asset_service.asset_path(cid, "-1") + ca.campaign = googleads_service.campaign_path(cid, changes["campaign_id"]) + ca.field_type = client.enums.AssetFieldTypeEnum.CALL + operations.append(link_op) + elif scope == "customer": + link_op = client.get_type("MutateOperation") + cust_asset = link_op.customer_asset_operation.create + cust_asset.asset = asset_service.asset_path(cid, "-1") + cust_asset.field_type = client.enums.AssetFieldTypeEnum.CALL + operations.append(link_op) + else: + raise ValueError(f"Unknown scope: {scope}") + + response = googleads_service.mutate( + customer_id=cid, mutate_operations=operations + ) + + result = {"asset": "", "link": ""} + for resp in response.mutate_operation_responses: + if resp.asset_result.resource_name and not result["asset"]: + result["asset"] = resp.asset_result.resource_name + elif scope == "ad_group" and resp.ad_group_asset_result.resource_name: + result["link"] = resp.ad_group_asset_result.resource_name + elif scope == "campaign" and resp.campaign_asset_result.resource_name: + result["link"] = resp.campaign_asset_result.resource_name + elif scope == "customer" and resp.customer_asset_result.resource_name: + result["link"] = resp.customer_asset_result.resource_name + return result + + +def _populate_promotion_asset(client: object, asset: object, promo: dict) -> None: + """Fill an Asset proto with PromotionAsset fields from a normalized dict.""" + p = asset.promotion_asset + p.promotion_target = promo["promotion_target"] + if promo.get("money_off"): + p.money_amount_off.amount_micros = int( + float(promo["money_off"]) * 1_000_000 + ) + p.money_amount_off.currency_code = promo["currency_code"] + elif promo.get("percent_off"): + p.percent_off = int(float(promo["percent_off"]) * 1_000_000) + + if promo.get("promotion_code"): + p.promotion_code = promo["promotion_code"] + + if promo.get("orders_over_amount"): + p.orders_over_amount.amount_micros = int( + float(promo["orders_over_amount"]) * 1_000_000 + ) + p.orders_over_amount.currency_code = promo["currency_code"] + + if promo.get("occasion"): + p.occasion = getattr( + client.enums.PromotionExtensionOccasionEnum, promo["occasion"] + ) + + if promo.get("discount_modifier"): + p.discount_modifier = getattr( + client.enums.PromotionExtensionDiscountModifierEnum, + promo["discount_modifier"], + ) + + p.language_code = promo.get("language_code") or "en" + if promo.get("start_date"): + p.start_date = promo["start_date"] + if promo.get("end_date"): + p.end_date = promo["end_date"] + if promo.get("redemption_start_date"): + p.redemption_start_date = promo["redemption_start_date"] + if promo.get("redemption_end_date"): + p.redemption_end_date = promo["redemption_end_date"] + + for entry in promo.get("ad_schedule") or []: + info = client.get_type("AdScheduleInfo") + _populate_ad_schedule_info(client, info, entry) + p.ad_schedule_targets.append(info) + + asset.final_urls.append(promo["final_url"]) + + +def _apply_create_promotion(client: object, cid: str, changes: dict) -> dict: + """Create a PromotionAsset and link it at ad group, campaign, or customer scope.""" + + def populate(asset: object, payload: dict) -> None: + _populate_promotion_asset(client, asset, payload) + + return _apply_assets( + client, + cid, + [changes["promotion"]], + client.enums.AssetFieldTypeEnum.PROMOTION, + populate, + scope=changes.get("scope", "campaign"), + campaign_id=changes.get("campaign_id", ""), + ad_group_id=changes.get("ad_group_id", ""), + ) + + +def _populate_price_asset(client: object, asset: object, price: dict) -> None: + """Fill an Asset proto with PriceAsset fields from a normalized dict.""" + p = asset.price_asset + p.type_ = getattr( + client.enums.PriceExtensionTypeEnum, price["price_type"] + ) + if price.get("price_qualifier"): + p.price_qualifier = getattr( + client.enums.PriceExtensionPriceQualifierEnum, + price["price_qualifier"], + ) + p.language_code = price.get("language_code") or "en" + + currency = price.get("currency_code") or "USD" + for row in price["offerings"]: + offering = client.get_type("PriceOffering") + offering.header = row["header"] + offering.description = row["description"] + offering.price.amount_micros = int(float(row["price"]) * 1_000_000) + offering.price.currency_code = currency + offering.final_url = row["final_url"] + if row.get("final_mobile_url"): + offering.final_mobile_url = row["final_mobile_url"] + if row.get("unit"): + offering.unit = getattr( + client.enums.PriceExtensionPriceUnitEnum, row["unit"] + ) + p.price_offerings.append(offering) + + +def _apply_create_price_asset(client: object, cid: str, changes: dict) -> dict: + """Create a PriceAsset and link it at ad group, campaign, or customer scope.""" + + def populate(asset: object, payload: dict) -> None: + _populate_price_asset(client, asset, payload) + + return _apply_assets( + client, + cid, + [changes["price"]], + client.enums.AssetFieldTypeEnum.PRICE, + populate, + scope=changes.get("scope", "campaign"), + campaign_id=changes.get("campaign_id", ""), + ad_group_id=changes.get("ad_group_id", ""), + ) + + +def _find_asset_link( + client: object, + cid: str, + asset_id: str, + field_type: str, + scope: str, + campaign_id: str, + ad_group_id: str, +) -> str: + """Look up the AdGroupAsset / CampaignAsset / CustomerAsset link resource + name for a given asset_id and field type, across any of the three scopes. + """ + googleads_service = client.get_service("GoogleAdsService") + asset_service = client.get_service("AssetService") + asset_resource = asset_service.asset_path(cid, asset_id) + + if scope == "ad_group": + if not ad_group_id: + return "" + ag_resource = googleads_service.ad_group_path(cid, ad_group_id) + query = ( + "SELECT ad_group_asset.resource_name FROM ad_group_asset " + f"WHERE ad_group_asset.asset = '{asset_resource}' " + f" AND ad_group_asset.ad_group = '{ag_resource}' " + f" AND ad_group_asset.field_type = '{field_type}'" + ) + elif scope == "campaign": + if not campaign_id: + return "" + camp_resource = client.get_service("CampaignService").campaign_path( + cid, campaign_id + ) + query = ( + "SELECT campaign_asset.resource_name FROM campaign_asset " + f"WHERE campaign_asset.asset = '{asset_resource}' " + f" AND campaign_asset.campaign = '{camp_resource}' " + f" AND campaign_asset.field_type = '{field_type}'" + ) + else: + query = ( + "SELECT customer_asset.resource_name FROM customer_asset " + f"WHERE customer_asset.asset = '{asset_resource}' " + f" AND customer_asset.field_type = '{field_type}'" + ) + + response = googleads_service.search(customer_id=cid, query=query) + for row in response: + if scope == "ad_group": + return row.ad_group_asset.resource_name + if scope == "campaign": + return row.campaign_asset.resource_name + return row.customer_asset.resource_name + return "" + + +def _apply_update_promotion(client: object, cid: str, changes: dict) -> dict: + """Swap a PromotionAsset: create new + link, then unlink old. + + Steps (create + link batched into one mutate call, then unlink old link): + 1. Create a new Asset with the new promotion fields. + 2. Link the new Asset (CampaignAsset or CustomerAsset). + 3. Remove the old link (matching the old asset_id at the same scope). + """ + asset_service = client.get_service("AssetService") + googleads_service = client.get_service("GoogleAdsService") + campaign_service = client.get_service("CampaignService") + + scope = changes.get("scope", "campaign") + campaign_id = changes.get("campaign_id", "") + old_asset_id = str(changes["old_asset_id"]) + promo = changes["promotion"] + + operations = [] + + # 1. Create new Asset + create_op = client.get_type("MutateOperation") + new_asset = create_op.asset_operation.create + new_asset.resource_name = asset_service.asset_path(cid, "-1") + _populate_promotion_asset(client, new_asset, promo) + operations.append(create_op) + + # 2. Link the new asset + if scope == "campaign": + if not campaign_id: + raise ValueError("campaign_id required for campaign-scope update") + link_op = client.get_type("MutateOperation") + ca = link_op.campaign_asset_operation.create + ca.asset = asset_service.asset_path(cid, "-1") + ca.campaign = campaign_service.campaign_path(cid, campaign_id) + ca.field_type = client.enums.AssetFieldTypeEnum.PROMOTION + operations.append(link_op) + elif scope == "customer": + link_op = client.get_type("MutateOperation") + cust = link_op.customer_asset_operation.create + cust.asset = asset_service.asset_path(cid, "-1") + cust.field_type = client.enums.AssetFieldTypeEnum.PROMOTION + operations.append(link_op) + else: + raise ValueError(f"Unknown scope: {scope}") + + response = googleads_service.mutate( + customer_id=cid, mutate_operations=operations + ) + + new_asset_resource = "" + new_link_resource = "" + for resp in response.mutate_operation_responses: + if resp.asset_result.resource_name and not new_asset_resource: + new_asset_resource = resp.asset_result.resource_name + elif scope == "campaign" and resp.campaign_asset_result.resource_name: + new_link_resource = resp.campaign_asset_result.resource_name + elif scope == "customer" and resp.customer_asset_result.resource_name: + new_link_resource = resp.customer_asset_result.resource_name + + # 3. Find the old link and remove it + old_link_resource = _find_asset_link( + client, cid, old_asset_id, "PROMOTION", scope, campaign_id, "" + ) + old_link_removed = "" + if old_link_resource: + if scope == "campaign": + ca_service = client.get_service("CampaignAssetService") + rm_op = client.get_type("CampaignAssetOperation") + rm_op.remove = old_link_resource + ca_service.mutate_campaign_assets(customer_id=cid, operations=[rm_op]) + else: + cust_service = client.get_service("CustomerAssetService") + rm_op = client.get_type("CustomerAssetOperation") + rm_op.remove = old_link_resource + cust_service.mutate_customer_assets(customer_id=cid, operations=[rm_op]) + old_link_removed = old_link_resource + + return { + "new_asset": new_asset_resource, + "new_link": new_link_resource, + "old_link_removed": old_link_removed, + } + + +def _apply_update_structured_snippet( + client: object, cid: str, changes: dict +) -> dict: + """Swap a StructuredSnippetAsset: create new + link, then unlink old. + + Mirrors _apply_update_promotion but generalized to ad-group / campaign / + customer scope via _find_asset_link. + """ + asset_service = client.get_service("AssetService") + googleads_service = client.get_service("GoogleAdsService") + campaign_service = client.get_service("CampaignService") + + scope = changes.get("scope", "customer") + campaign_id = changes.get("campaign_id", "") + ad_group_id = changes.get("ad_group_id", "") + old_asset_id = str(changes["old_asset_id"]) + snippet = changes["snippet"] + field_type = client.enums.AssetFieldTypeEnum.STRUCTURED_SNIPPET + + operations = [] + + # 1. Create the new Asset. + create_op = client.get_type("MutateOperation") + new_asset = create_op.asset_operation.create + new_asset.resource_name = asset_service.asset_path(cid, "-1") + new_asset.structured_snippet_asset.header = snippet["header"] + new_asset.structured_snippet_asset.values.extend(snippet["values"]) + operations.append(create_op) + + # 2. Link the new Asset at the same scope as the old one. + link_op = client.get_type("MutateOperation") + if scope == "ad_group": + if not ad_group_id: + raise ValueError("ad_group_id required for ad_group-scope update") + aga = link_op.ad_group_asset_operation.create + aga.asset = asset_service.asset_path(cid, "-1") + aga.ad_group = googleads_service.ad_group_path(cid, ad_group_id) + aga.field_type = field_type + elif scope == "campaign": + if not campaign_id: + raise ValueError("campaign_id required for campaign-scope update") + ca = link_op.campaign_asset_operation.create + ca.asset = asset_service.asset_path(cid, "-1") + ca.campaign = campaign_service.campaign_path(cid, campaign_id) + ca.field_type = field_type + elif scope == "customer": + cust = link_op.customer_asset_operation.create + cust.asset = asset_service.asset_path(cid, "-1") + cust.field_type = field_type + else: + raise ValueError(f"Unknown scope: {scope}") + operations.append(link_op) + + response = googleads_service.mutate( + customer_id=cid, mutate_operations=operations + ) + + new_asset_resource = "" + new_link_resource = "" + for resp in response.mutate_operation_responses: + if resp.asset_result.resource_name and not new_asset_resource: + new_asset_resource = resp.asset_result.resource_name + elif scope == "ad_group" and resp.ad_group_asset_result.resource_name: + new_link_resource = resp.ad_group_asset_result.resource_name + elif scope == "campaign" and resp.campaign_asset_result.resource_name: + new_link_resource = resp.campaign_asset_result.resource_name + elif scope == "customer" and resp.customer_asset_result.resource_name: + new_link_resource = resp.customer_asset_result.resource_name + + # 3. Find and unlink the old link. + old_link_resource = _find_asset_link( + client, + cid, + old_asset_id, + "STRUCTURED_SNIPPET", + scope, + campaign_id, + ad_group_id, + ) + old_link_removed = "" + if old_link_resource: + if scope == "ad_group": + svc = client.get_service("AdGroupAssetService") + rm_op = client.get_type("AdGroupAssetOperation") + rm_op.remove = old_link_resource + svc.mutate_ad_group_assets(customer_id=cid, operations=[rm_op]) + elif scope == "campaign": + svc = client.get_service("CampaignAssetService") + rm_op = client.get_type("CampaignAssetOperation") + rm_op.remove = old_link_resource + svc.mutate_campaign_assets(customer_id=cid, operations=[rm_op]) + else: + svc = client.get_service("CustomerAssetService") + rm_op = client.get_type("CustomerAssetOperation") + rm_op.remove = old_link_resource + svc.mutate_customer_assets(customer_id=cid, operations=[rm_op]) + old_link_removed = old_link_resource + + return { + "new_asset": new_asset_resource, + "new_link": new_link_resource, + "old_link_removed": old_link_removed, + } + + +def _apply_update_call_asset(client: object, cid: str, changes: dict) -> dict: + """In-place update of an existing CallAsset.""" + from google.protobuf import field_mask_pb2 + + asset_service = client.get_service("AssetService") + op = client.get_type("AssetOperation") + asset = op.update + asset.resource_name = asset_service.asset_path(cid, changes["asset_id"]) + + paths: list[str] = [] + if "phone_number" in changes: + asset.call_asset.phone_number = changes["phone_number"] + paths.append("call_asset.phone_number") + if "country_code" in changes: + asset.call_asset.country_code = changes["country_code"] + paths.append("call_asset.country_code") + if "call_conversion_action_id" in changes: + ca_service = client.get_service("ConversionActionService") + asset.call_asset.call_conversion_action = ca_service.conversion_action_path( + cid, changes["call_conversion_action_id"] + ) + paths.append("call_asset.call_conversion_action") + if "call_conversion_reporting_state" in changes: + asset.call_asset.call_conversion_reporting_state = getattr( + client.enums.CallConversionReportingStateEnum, + changes["call_conversion_reporting_state"], + ) + paths.append("call_asset.call_conversion_reporting_state") + if "ad_schedule" in changes: + # Replace the schedule list entirely + for entry in changes["ad_schedule"]: + info = client.get_type("AdScheduleInfo") + _populate_ad_schedule_info(client, info, entry) + asset.call_asset.ad_schedule_targets.append(info) + paths.append("call_asset.ad_schedule_targets") + + op.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=paths)) + response = asset_service.mutate_assets(customer_id=cid, operations=[op]) + return {"resource_name": response.results[0].resource_name} + + +def _apply_update_sitelink(client: object, cid: str, changes: dict) -> dict: + """In-place update of an existing SitelinkAsset.""" + from google.protobuf import field_mask_pb2 + + asset_service = client.get_service("AssetService") + op = client.get_type("AssetOperation") + asset = op.update + asset.resource_name = asset_service.asset_path(cid, changes["asset_id"]) + + paths: list[str] = [] + if "link_text" in changes: + asset.sitelink_asset.link_text = changes["link_text"] + paths.append("sitelink_asset.link_text") + if "description1" in changes: + asset.sitelink_asset.description1 = changes["description1"] + paths.append("sitelink_asset.description1") + if "description2" in changes: + asset.sitelink_asset.description2 = changes["description2"] + paths.append("sitelink_asset.description2") + if "final_url" in changes: + asset.final_urls.append(changes["final_url"]) + paths.append("final_urls") + + op.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=paths)) + response = asset_service.mutate_assets(customer_id=cid, operations=[op]) + return {"resource_name": response.results[0].resource_name} + + +def _apply_update_callout(client: object, cid: str, changes: dict) -> dict: + """In-place update of an existing CalloutAsset's text.""" + from google.protobuf import field_mask_pb2 + + asset_service = client.get_service("AssetService") + op = client.get_type("AssetOperation") + asset = op.update + asset.resource_name = asset_service.asset_path(cid, changes["asset_id"]) + asset.callout_asset.callout_text = changes["callout_text"] + op.update_mask.CopyFrom(field_mask_pb2.FieldMask( + paths=["callout_asset.callout_text"] + )) + response = asset_service.mutate_assets(customer_id=cid, operations=[op]) + return {"resource_name": response.results[0].resource_name} + + +def _apply_create_location_asset( + client: object, cid: str, changes: dict +) -> dict: + """Create a LOCATION_SYNC AssetSet linked to a Google Business Profile. + + Steps: + 1. Create AssetSet (type=LOCATION_SYNC, business_profile_location_set). + 2. Create CustomerAssetSet linking the set to the customer (customer scope). + 3. Or create CampaignAssetSet linking it to one campaign (campaign scope). + """ + asset_set_service = client.get_service("AssetSetService") + + # Step 1 — create AssetSet + set_op = client.get_type("AssetSetOperation") + asset_set = set_op.create + asset_set.name = changes["asset_set_name"] + asset_set.type_ = client.enums.AssetSetTypeEnum.LOCATION_SYNC + bpls = asset_set.location_set.business_profile_location_set + # business_account_id is exposed as STRING by proto-plus even though the + # value is a numeric GBP/LBC account id. + bpls.business_account_id = str(changes["business_profile_account_id"]) + for label in changes.get("label_filters") or []: + bpls.label_filters.append(label) + for listing_id in changes.get("listing_id_filters") or []: + bpls.listing_id_filters.append(int(listing_id)) + + set_response = asset_set_service.mutate_asset_sets( + customer_id=cid, operations=[set_op] + ) + asset_set_resource = set_response.results[0].resource_name + + result = {"asset_set": asset_set_resource} + + # Step 2/3 — link to customer or campaign + scope = changes.get("scope", "customer") + if scope == "customer": + cas_service = client.get_service("CustomerAssetSetService") + cas_op = client.get_type("CustomerAssetSetOperation") + cas_op.create.asset_set = asset_set_resource + cas_response = cas_service.mutate_customer_asset_sets( + customer_id=cid, operations=[cas_op] + ) + result["customer_asset_set"] = cas_response.results[0].resource_name + elif scope == "campaign": + if not changes.get("campaign_id"): + raise ValueError("campaign_id required for campaign-scope location asset") + campaign_service = client.get_service("CampaignService") + cas_service = client.get_service("CampaignAssetSetService") + cas_op = client.get_type("CampaignAssetSetOperation") + cas_op.create.asset_set = asset_set_resource + cas_op.create.campaign = campaign_service.campaign_path( + cid, changes["campaign_id"] + ) + cas_response = cas_service.mutate_campaign_asset_sets( + customer_id=cid, operations=[cas_op] + ) + result["campaign_asset_set"] = cas_response.results[0].resource_name + else: + raise ValueError(f"Unknown scope: {scope}") + + return result + + +def _apply_create_business_name_asset( + client: object, cid: str, changes: dict +) -> dict: + """Create a TEXT asset and link as BUSINESS_NAME at customer or campaign scope.""" + asset_service = client.get_service("AssetService") + googleads_service = client.get_service("GoogleAdsService") + operations: list = [] + + # 1) Create Asset (TEXT) + op = client.get_type("MutateOperation") + asset = op.asset_operation.create + asset.resource_name = asset_service.asset_path(cid, "-1") + asset.type_ = client.enums.AssetTypeEnum.TEXT + asset.text_asset.text = changes["business_name"] + operations.append(op) + + # 2) Link as BUSINESS_NAME + scope = changes.get("scope", "customer") + link_op = client.get_type("MutateOperation") + if scope == "campaign": + if not changes.get("campaign_id"): + raise ValueError("campaign_id required for campaign-scope business_name asset") + link = link_op.campaign_asset_operation.create + link.asset = asset_service.asset_path(cid, "-1") + link.campaign = googleads_service.campaign_path(cid, changes["campaign_id"]) + link.field_type = client.enums.AssetFieldTypeEnum.BUSINESS_NAME + elif scope == "customer": + link = link_op.customer_asset_operation.create + link.asset = asset_service.asset_path(cid, "-1") + link.field_type = client.enums.AssetFieldTypeEnum.BUSINESS_NAME + else: + raise ValueError(f"Unknown scope: {scope}") + operations.append(link_op) + + response = googleads_service.mutate( + customer_id=cid, mutate_operations=operations + ) + + result = {"asset": "", "link": ""} + for resp in response.mutate_operation_responses: + if resp.asset_result.resource_name and not result["asset"]: + result["asset"] = resp.asset_result.resource_name + elif scope == "campaign" and resp.campaign_asset_result.resource_name: + result["link"] = resp.campaign_asset_result.resource_name + elif scope == "customer" and resp.customer_asset_result.resource_name: + result["link"] = resp.customer_asset_result.resource_name + return result + + +def _apply_link_asset_to_customer( + client: object, cid: str, changes: dict +) -> dict: + """Create CustomerAsset link rows pointing to existing Asset rows. + + Does NOT create new Asset rows — only the link. Use this to promote + existing assets (e.g. images, logos on a legacy campaign) to account level. + """ + asset_service = client.get_service("AssetService") + cust_service = client.get_service("CustomerAssetService") + + operations = [] + for link in changes["links"]: + op = client.get_type("CustomerAssetOperation") + ca = op.create + ca.asset = asset_service.asset_path(cid, link["asset_id"]) + ca.field_type = getattr( + client.enums.AssetFieldTypeEnum, link["field_type"] + ) + operations.append(op) + + response = cust_service.mutate_customer_assets( + customer_id=cid, operations=operations + ) + return { + "customer_assets": [r.resource_name for r in response.results], + "linked_count": len(response.results), + } + + def _apply_create_negative_keyword_list( client: object, cid: str, changes: dict ) -> dict: diff --git a/src/adloop/server.py b/src/adloop/server.py index 5ebe0ff..85f14f2 100644 --- a/src/adloop/server.py +++ b/src/adloop/server.py @@ -1837,17 +1837,27 @@ def update_ad_group( @mcp.tool(annotations=_WRITE) @_safe def draft_callouts( - campaign_id: str, callouts: _StrList, + campaign_id: str = "", + ad_group_id: str = "", customer_id: str = "", ) -> dict: - """Draft campaign callout assets — returns a PREVIEW.""" + """Draft callout assets — returns a PREVIEW. + + Scope (most-specific wins): + - ad_group_id set → the callouts link to that ad group only (use for + per-service callouts inside a multi-ad-group campaign). + - campaign_id set → the callouts link at the campaign level. + - both empty → the callouts link at the customer/account level and + apply to all eligible campaigns. + """ from adloop.ads.write import draft_callouts as _impl return _impl( current_config(), customer_id=customer_id or current_config().ads.customer_id, campaign_id=campaign_id, + ad_group_id=ad_group_id, callouts=callouts, ) @@ -1855,17 +1865,25 @@ def draft_callouts( @mcp.tool(annotations=_WRITE) @_safe def draft_structured_snippets( - campaign_id: str, snippets: _DictList, + campaign_id: str = "", + ad_group_id: str = "", customer_id: str = "", ) -> dict: - """Draft campaign structured snippet assets — returns a PREVIEW.""" + """Draft structured snippet assets — returns a PREVIEW. + + Scope (most-specific wins): + - ad_group_id set → snippets link to that ad group only. + - campaign_id set → snippets link at the campaign level. + - both empty → snippets link at the customer/account level. + """ from adloop.ads.write import draft_structured_snippets as _impl return _impl( current_config(), customer_id=customer_id or current_config().ads.customer_id, campaign_id=campaign_id, + ad_group_id=ad_group_id, snippets=snippets, ) @@ -2013,6 +2031,458 @@ def draft_sitelinks( ) +@mcp.tool(annotations=_WRITE) +@_safe +def draft_call_asset( + phone_number: str, + country_code: str = "US", + campaign_id: str = "", + ad_group_id: str = "", + call_conversion_action_id: str = "", + ad_schedule: _DictListOpt = None, + customer_id: str = "", +) -> dict: + """Draft a call asset (phone extension) — returns a PREVIEW. + + Scope (most-specific wins): + - ad_group_id set → the call asset attaches to that ad group only. + - campaign_id set → the call asset attaches at the campaign level. + - both empty → the call asset attaches at the customer/account level. + + phone_number: human or E.164 (e.g. "+15555550142" or "(555) 555-0142"). + country_code: 2-letter ISO code used to canonicalize a national number to + E.164; ignored when phone_number already starts with '+'. + call_conversion_action_id: optional conversion action ID to count + qualifying calls at the resource level. + ad_schedule: optional list of {day_of_week, start_hour, end_hour, + start_minute, end_minute} entries limiting when the extension shows. + + Google Ads requires manual phone-number verification before call assets + serve. Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import draft_call_asset as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + phone_number=phone_number, + country_code=country_code, + campaign_id=campaign_id, + ad_group_id=ad_group_id, + call_conversion_action_id=call_conversion_action_id, + ad_schedule=ad_schedule, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def add_ad_schedule( + campaign_id: str, + schedule: _DictList, + customer_id: str = "", +) -> dict: + """Draft ad schedule additions for a campaign — returns a PREVIEW. + + Creates AD_SCHEDULE CampaignCriterion records so the campaign only serves + during the given hours/days (in the account's time zone). Additive — it + does NOT replace existing schedule criteria. + + schedule: list of dicts with keys: + - day_of_week: MONDAY..SUNDAY + - start_hour: 0..23 + - end_hour: 0..24 (must be > start) + - start_minute / end_minute: 0, 15, 30, or 45 (default 0) + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import add_ad_schedule as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + campaign_id=campaign_id, + schedule=schedule, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_promotion( + promotion_target: str, + final_url: str, + money_off: float = 0, + percent_off: float = 0, + currency_code: str = "USD", + promotion_code: str = "", + orders_over_amount: float = 0, + occasion: str = "", + discount_modifier: str = "", + language_code: str = "en", + start_date: str = "", + end_date: str = "", + redemption_start_date: str = "", + redemption_end_date: str = "", + campaign_id: str = "", + ad_group_id: str = "", + ad_schedule: _DictListOpt = None, + customer_id: str = "", +) -> dict: + """Draft a promotion extension asset — returns a PREVIEW. + + Scope (most-specific wins): ad_group_id → campaign_id → customer/account. + Exactly one of money_off / percent_off is required. + + promotion_target: label Google shows (max 20 chars). + final_url: reachable landing page (validated). + occasion / discount_modifier: optional Google enum tags. + promotion_code and orders_over_amount are mutually exclusive. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import draft_promotion as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + promotion_target=promotion_target, + final_url=final_url, + money_off=money_off, + percent_off=percent_off, + currency_code=currency_code, + promotion_code=promotion_code, + orders_over_amount=orders_over_amount, + occasion=occasion, + discount_modifier=discount_modifier, + language_code=language_code, + start_date=start_date, + end_date=end_date, + redemption_start_date=redemption_start_date, + redemption_end_date=redemption_end_date, + campaign_id=campaign_id, + ad_group_id=ad_group_id, + ad_schedule=ad_schedule, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def update_promotion( + asset_id: str, + promotion_target: str, + final_url: str, + money_off: float = 0, + percent_off: float = 0, + currency_code: str = "USD", + promotion_code: str = "", + orders_over_amount: float = 0, + occasion: str = "", + discount_modifier: str = "", + language_code: str = "en", + start_date: str = "", + end_date: str = "", + redemption_start_date: str = "", + redemption_end_date: str = "", + campaign_id: str = "", + ad_schedule: _DictListOpt = None, + customer_id: str = "", +) -> dict: + """Update a promotion asset via swap — returns a PREVIEW. + + PromotionAsset fields are immutable, so this creates a new asset with the + new values, links it at the same scope, and unlinks the old one. Pass + campaign_id for campaign scope; leave empty for customer/account scope. + The old Asset row stays in the account (orphaned) — Google reclaims it. + + asset_id: numeric ID of the existing PromotionAsset to replace. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import update_promotion as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + asset_id=asset_id, + campaign_id=campaign_id, + promotion_target=promotion_target, + final_url=final_url, + money_off=money_off, + percent_off=percent_off, + currency_code=currency_code, + promotion_code=promotion_code, + orders_over_amount=orders_over_amount, + occasion=occasion, + discount_modifier=discount_modifier, + language_code=language_code, + start_date=start_date, + end_date=end_date, + redemption_start_date=redemption_start_date, + redemption_end_date=redemption_end_date, + ad_schedule=ad_schedule, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_price_asset( + offerings: _DictList, + campaign_id: str = "", + ad_group_id: str = "", + price_type: str = "SERVICES", + price_qualifier: str = "FROM", + language_code: str = "en", + currency_code: str = "USD", + customer_id: str = "", +) -> dict: + """Draft a price extension asset — returns a PREVIEW. + + Scope (most-specific wins): ad_group_id → campaign_id → customer/account. + Google Ads requires 3-8 offerings. + + offerings: list of 3-8 dicts, each with header (<=25), description (<=25), + price (> 0), final_url (reachable), optional final_mobile_url and unit. + price_type: SERVICES (default), BRANDS, EVENTS, etc. + price_qualifier: FROM (default), UP_TO, AVERAGE, or empty. + + Every price MUST match the amount shown on its final_url page. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import draft_price_asset as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + campaign_id=campaign_id, + ad_group_id=ad_group_id, + price_type=price_type, + price_qualifier=price_qualifier, + language_code=language_code, + currency_code=currency_code, + offerings=offerings, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def update_structured_snippet( + asset_id: str, + header: str, + values: _StrList, + campaign_id: str = "", + ad_group_id: str = "", + customer_id: str = "", +) -> dict: + """Update a structured snippet via swap — returns a PREVIEW. + + StructuredSnippetAsset fields are immutable, so this creates a new asset + with the new header/values, links it at the same scope, and unlinks the + old one. Scope (most-specific wins): ad_group_id → campaign_id → + customer/account, and MUST match where the old asset is linked. + + asset_id: numeric ID of the existing StructuredSnippetAsset to replace. + header: official structured-snippet header (e.g. "Brands", "Services"). + values: 3-10 values, each <= 25 chars. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import update_structured_snippet as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + asset_id=asset_id, + campaign_id=campaign_id, + ad_group_id=ad_group_id, + header=header, + values=values, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def update_call_asset( + asset_id: str, + phone_number: str = "", + country_code: str = "", + call_conversion_action_id: str = "", + call_conversion_reporting_state: str = "", + ad_schedule: _DictListOpt = None, + customer_id: str = "", +) -> dict: + """Update an existing CallAsset in place — returns a PREVIEW. + + Pass only the fields to change (empty string/None = leave unchanged). Use + this to re-point a call asset at a resource-level conversion action, change + the number, or replace the ad-schedule windows. + + asset_id: numeric ID of the existing call asset. + call_conversion_reporting_state: DISABLED | + USE_ACCOUNT_LEVEL_CALL_CONVERSION_ACTION | + USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import update_call_asset as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + asset_id=asset_id, + phone_number=phone_number, + country_code=country_code, + call_conversion_action_id=call_conversion_action_id, + call_conversion_reporting_state=call_conversion_reporting_state, + ad_schedule=ad_schedule, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def update_sitelink( + asset_id: str, + link_text: str = "", + final_url: str = "", + description1: str = "", + description2: str = "", + customer_id: str = "", +) -> dict: + """Update an existing SitelinkAsset in place — returns a PREVIEW. + + Pass only the fields to change (empty string = leave unchanged). A changed + final_url is validated for reachability. + + asset_id: numeric ID of the existing sitelink asset. + link_text: max 25 chars. description1/description2: max 35 chars each. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import update_sitelink as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + asset_id=asset_id, + link_text=link_text, + final_url=final_url, + description1=description1, + description2=description2, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def update_callout( + asset_id: str, + callout_text: str, + customer_id: str = "", +) -> dict: + """Update an existing CalloutAsset's text in place — returns a PREVIEW. + + asset_id: numeric ID of the existing callout asset. + callout_text: new callout text (max 25 chars). + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import update_callout as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + asset_id=asset_id, + callout_text=callout_text, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_location_asset( + business_profile_account_id: str, + asset_set_name: str = "", + campaign_id: str = "", + label_filters: _StrListOpt = None, + listing_id_filters: _StrListOpt = None, + customer_id: str = "", +) -> dict: + """Draft a Google Business Profile location AssetSet — returns a PREVIEW. + + Creates a LOCATION_SYNC AssetSet that pulls locations from a linked Google + Business Profile and links it at customer scope (or to one campaign when + campaign_id is set). The GBP must already be linked in Google Ads → Tools + → Linked accounts → Business Profile. + + business_profile_account_id: numeric GBP/LBC account ID. + label_filters / listing_id_filters: optional filters to limit which GBP + locations sync. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import draft_location_asset as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + business_profile_account_id=business_profile_account_id, + asset_set_name=asset_set_name, + campaign_id=campaign_id, + label_filters=label_filters, + listing_id_filters=listing_id_filters, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_business_name_asset( + business_name: str, + campaign_id: str = "", + customer_id: str = "", +) -> dict: + """Draft a business-name asset — returns a PREVIEW. + + Creates a TEXT asset linked as BUSINESS_NAME. Links at customer scope by + default, or to one campaign when campaign_id is set. + + business_name: max 25 characters per Google Ads policy. + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import draft_business_name_asset as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + campaign_id=campaign_id, + business_name=business_name, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def link_asset_to_customer( + links: _DictList, + customer_id: str = "", +) -> dict: + """Link EXISTING assets to the customer/account level — returns a PREVIEW. + + "Promotes" assets that already exist (typically on legacy campaigns) so + they apply account-wide. Does NOT create new Asset rows — only CustomerAsset + link rows. Find candidate asset_ids via: SELECT asset.id, asset.type, + asset.name FROM asset. + + links: list of dicts, each with asset_id (numeric str) and field_type + (AssetFieldType, e.g. BUSINESS_LOGO, MARKETING_IMAGE, SITELINK, CALL). + + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.write import link_asset_to_customer as _impl + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + links=links, + ) + + @mcp.tool(annotations=_DESTRUCTIVE) @_safe def confirm_and_apply( diff --git a/tests/test_ads_extensions.py b/tests/test_ads_extensions.py new file mode 100644 index 0000000..8f4ae34 --- /dev/null +++ b/tests/test_ads_extensions.py @@ -0,0 +1,1220 @@ +"""Tests for asset-extension write tools (call/price/promotion/location/etc.). + +Covers preview validation and the apply-layer mutate operations for the +asset-extension tools, including the campaign/ad-group/customer multi-scope +paths. All data here is synthetic — placeholder phone numbers (555-01xx), +placeholder business names, and example.com URLs. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from google.ads.googleads.client import GoogleAdsClient + +from adloop.ads import write +from adloop.ads.client import GOOGLE_ADS_API_VERSION +from adloop.config import AdLoopConfig, AdsConfig, SafetyConfig +from adloop.safety import preview as preview_store + + +# --------------------------------------------------------------------------- +# Fakes — mirror the lightweight harness in tests/test_ads_write.py +# --------------------------------------------------------------------------- + + +class _FakeResult: + def __init__(self, resource_name: str = ""): + self.resource_name = resource_name + + +class _FakeMutateOperationResponse: + def __init__(self, response_type: str | None = None, resource_name: str = ""): + self.asset_result = _FakeResult() + self.campaign_asset_result = _FakeResult() + self.ad_group_asset_result = _FakeResult() + self.customer_asset_result = _FakeResult() + if response_type: + getattr(self, response_type).resource_name = resource_name + + +class _FakePathService: + def __init__(self, prefix: str): + self.prefix = prefix + + def campaign_path(self, customer_id: str, entity_id: str) -> str: + return f"customers/{customer_id}/campaigns/{entity_id}" + + def ad_group_path(self, customer_id: str, entity_id: str) -> str: + return f"customers/{customer_id}/adGroups/{entity_id}" + + def asset_path(self, customer_id: str, entity_id: str) -> str: + return f"customers/{customer_id}/assets/{entity_id}" + + def conversion_action_path(self, customer_id: str, entity_id: str) -> str: + return f"customers/{customer_id}/conversionActions/{entity_id}" + + +class _FakeGoogleAdsService(_FakePathService): + def __init__(self, responses: list[_FakeMutateOperationResponse] | None = None): + super().__init__("googleAds") + self.operations = None + self._responses = responses or [] + self.search_queries: list[str] = [] + self.search_rows: list[object] = [] + + def mutate(self, customer_id: str, mutate_operations: list[object]) -> object: + self.operations = mutate_operations + return SimpleNamespace(mutate_operation_responses=self._responses) + + def search(self, customer_id: str, query: str) -> list[object]: + self.search_queries.append(query) + return self.search_rows + + +class _FakeCampaignCriterionService: + def __init__(self): + self.operations = None + + def mutate_campaign_criteria( + self, customer_id: str, operations: list[object] + ) -> object: + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/campaignCriteria/{i}" + ) + for i, _ in enumerate(operations) + ] + ) + + +class _FakeAssetService(_FakePathService): + def __init__(self): + super().__init__("assets") + self.operations = None + + def mutate_assets(self, customer_id: str, operations: list[object]) -> object: + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/assets/{i}" + ) + for i, _ in enumerate(operations) + ] + ) + + +class _FakeLinkService: + """Generic fake for CampaignAsset / CustomerAsset / AdGroupAsset services.""" + + def __init__(self, prefix: str): + self.prefix = prefix + self.operations = None + + def _mutate(self, customer_id, operations): + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/{self.prefix}/{i}" + ) + for i, _ in enumerate(operations) + ] + ) + + mutate_campaign_assets = _mutate + mutate_customer_assets = _mutate + mutate_ad_group_assets = _mutate + + +class _FakeCustomerAssetService: + def __init__(self): + self.operations = None + + def mutate_customer_assets( + self, customer_id: str, operations: list[object] + ) -> object: + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/customerAssets/{i}" + ) + for i, _ in enumerate(operations) + ] + ) + + +class _FakeAssetSetService: + def __init__(self): + self.operations = None + + def mutate_asset_sets(self, customer_id: str, operations: list[object]) -> object: + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/assetSets/1" + ) + ] + ) + + +class _FakeAssetSetLinkService: + def __init__(self, prefix: str): + self.prefix = prefix + self.operations = None + + def _mutate(self, customer_id, operations): + self.operations = operations + return SimpleNamespace( + results=[ + SimpleNamespace( + resource_name=f"customers/{customer_id}/{self.prefix}/1" + ) + ] + ) + + mutate_customer_asset_sets = _mutate + mutate_campaign_asset_sets = _mutate + + +class _FakeClient: + def __init__(self, services: dict[str, object]): + self._base = GoogleAdsClient( + credentials=None, + developer_token="test-token", + use_proto_plus=True, + version=GOOGLE_ADS_API_VERSION, + ) + self.enums = self._base.enums + self.get_type = self._base.get_type + self._services = services + + def get_service(self, name: str) -> object: + return self._services[name] + + +@pytest.fixture(autouse=True) +def clear_pending_plans(): + preview_store.set_plan_store(preview_store.InMemoryPlanStore()) + yield + preview_store.set_plan_store(preview_store.InMemoryPlanStore()) + + +@pytest.fixture +def config() -> AdLoopConfig: + return AdLoopConfig( + ads=AdsConfig(customer_id="123-456-7890"), + safety=SafetyConfig(require_dry_run=True), + ) + + +def _asset_link_client(responses): + google_ads_service = _FakeGoogleAdsService(responses) + return google_ads_service, _FakeClient( + { + "GoogleAdsService": google_ads_service, + "AssetService": _FakePathService("assets"), + "CampaignService": _FakePathService("campaigns"), + "ConversionActionService": _FakePathService("conversionActions"), + } + ) + + +# --------------------------------------------------------------------------- +# Phone normalization +# --------------------------------------------------------------------------- + + +def test_normalize_phone_us_national(): + normalized, err = write._normalize_phone_e164("(555) 555-0142", "US") + assert err is None + assert normalized == "+15555550142" + + +def test_normalize_phone_strips_leading_country_code(): + normalized, err = write._normalize_phone_e164("1-555-555-0143", "US") + assert err is None + assert normalized == "+15555550143" + + +def test_normalize_phone_already_e164_passthrough(): + normalized, err = write._normalize_phone_e164("+445555550144", "GB") + assert err is None + assert normalized == "+445555550144" + + +def test_normalize_phone_strips_european_trunk_zero(): + normalized, err = write._normalize_phone_e164("020 5550 0145", "GB") + assert err is None + # exactly one trunk "0" removed (not every leading zero) + assert normalized == "+442055500145" + + +def test_normalize_phone_double_zero_international_prefix(): + # "00" is the international access prefix; the following digits already + # carry the country code — must become "+44...", not "+44 00...". + normalized, err = write._normalize_phone_e164("0044 20 5550 0145", "GB") + assert err is None + assert normalized == "+442055500145" + + +def test_normalize_phone_italy_keeps_leading_zero(): + # Italy is the exception — the trunk "0" is part of the E.164 number and + # must NOT be stripped. + normalized, err = write._normalize_phone_e164("06 5550 0145", "IT") + assert err is None + assert normalized == "+390655500145" + + +def test_normalize_phone_unknown_country_code_errors(): + normalized, err = write._normalize_phone_e164("5555550146", "ZZ") + assert normalized == "" + assert "dial-code map" in err + + +# --------------------------------------------------------------------------- +# draft_call_asset — preview + scope +# --------------------------------------------------------------------------- + + +def test_draft_call_asset_requires_phone(config): + result = write.draft_call_asset(config, customer_id="123-456-7890") + assert result["error"] == "phone_number is required" + + +def test_draft_call_asset_campaign_scope(config): + result = write.draft_call_asset( + config, + customer_id="123-456-7890", + phone_number="(555) 555-0142", + campaign_id="1001", + ) + assert result["operation"] == "create_call_asset" + assert result["changes"]["scope"] == "campaign" + assert result["changes"]["phone_number"] == "+15555550142" + assert result["warnings"] + + +def test_draft_call_asset_ad_group_scope_wins(config): + result = write.draft_call_asset( + config, + customer_id="123-456-7890", + phone_number="+15555550142", + campaign_id="1001", + ad_group_id="2002", + ) + assert result["changes"]["scope"] == "ad_group" + assert result["entity_type"] == "ad_group_asset" + assert result["entity_id"] == "2002" + + +def test_draft_call_asset_customer_scope_default(config): + result = write.draft_call_asset( + config, + customer_id="123-456-7890", + phone_number="+15555550142", + ) + assert result["changes"]["scope"] == "customer" + assert result["entity_type"] == "customer_asset" + + +def test_draft_call_asset_rejects_bad_schedule(config): + result = write.draft_call_asset( + config, + customer_id="123-456-7890", + phone_number="+15555550142", + ad_schedule=[{"day_of_week": "FUNDAY", "start_hour": 9, "end_hour": 17}], + ) + assert result["error"] == "Ad schedule validation failed" + + +def test_apply_create_call_asset_campaign_scope_links_call_field(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/1"), + _FakeMutateOperationResponse( + "campaign_asset_result", "customers/1234567890/campaignAssets/1" + ), + ] + google_ads_service, client = _asset_link_client(responses) + + write._apply_create_call_asset( + client, + "1234567890", + { + "scope": "campaign", + "campaign_id": "1001", + "ad_group_id": "", + "phone_number": "+15555550142", + "country_code": "US", + "call_conversion_action_id": "", + "ad_schedule": [], + }, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.call_asset.phone_number == "+15555550142" + link = google_ads_service.operations[1].campaign_asset_operation.create + assert link.field_type == client.enums.AssetFieldTypeEnum.CALL + + +def test_apply_create_call_asset_ad_group_scope_and_conversion_action(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/1"), + _FakeMutateOperationResponse( + "ad_group_asset_result", "customers/1234567890/adGroupAssets/1" + ), + ] + google_ads_service, client = _asset_link_client(responses) + + write._apply_create_call_asset( + client, + "1234567890", + { + "scope": "ad_group", + "campaign_id": "", + "ad_group_id": "2002", + "phone_number": "+15555550142", + "country_code": "US", + "call_conversion_action_id": "555999", + "ad_schedule": [ + { + "day_of_week": "MONDAY", + "start_hour": 9, + "start_minute": 0, + "end_hour": 17, + "end_minute": 30, + } + ], + }, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.call_asset.call_conversion_action.endswith("conversionActions/555999") + assert ( + create.call_asset.call_conversion_reporting_state + == client.enums.CallConversionReportingStateEnum.USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION + ) + assert len(create.call_asset.ad_schedule_targets) == 1 + link = google_ads_service.operations[1].ad_group_asset_operation.create + assert link.field_type == client.enums.AssetFieldTypeEnum.CALL + + +# --------------------------------------------------------------------------- +# add_ad_schedule +# --------------------------------------------------------------------------- + + +def test_add_ad_schedule_requires_campaign(config): + result = write.add_ad_schedule( + config, + customer_id="123-456-7890", + schedule=[{"day_of_week": "MONDAY", "start_hour": 9, "end_hour": 17}], + ) + assert result["error"] == "campaign_id is required" + + +def test_add_ad_schedule_rejects_bad_minute(config): + result = write.add_ad_schedule( + config, + customer_id="123-456-7890", + campaign_id="1001", + schedule=[ + { + "day_of_week": "MONDAY", + "start_hour": 9, + "start_minute": 10, + "end_hour": 17, + } + ], + ) + assert result["error"] == "Validation failed" + assert any("start_minute" in d for d in result["details"]) + + +def test_add_ad_schedule_rejects_end_before_start(config): + result = write.add_ad_schedule( + config, + customer_id="123-456-7890", + campaign_id="1001", + schedule=[ + {"day_of_week": "MONDAY", "start_hour": 17, "end_hour": 9} + ], + ) + assert result["error"] == "Validation failed" + assert any("must be after" in d for d in result["details"]) + + +def test_add_ad_schedule_preview_ok(config): + result = write.add_ad_schedule( + config, + customer_id="123-456-7890", + campaign_id="1001", + schedule=[ + { + "day_of_week": "monday", + "start_hour": 9, + "start_minute": 15, + "end_hour": 17, + "end_minute": 45, + } + ], + ) + assert result["operation"] == "add_ad_schedule" + assert result["changes"]["schedule"][0]["day_of_week"] == "MONDAY" + + +def test_apply_add_ad_schedule_builds_criteria(): + crit_service = _FakeCampaignCriterionService() + client = _FakeClient( + { + "CampaignService": _FakePathService("campaigns"), + "CampaignCriterionService": crit_service, + } + ) + result = write._apply_add_ad_schedule( + client, + "1234567890", + { + "campaign_id": "1001", + "schedule": [ + { + "day_of_week": "MONDAY", + "start_hour": 9, + "start_minute": 0, + "end_hour": 17, + "end_minute": 0, + } + ], + }, + ) + assert len(result["campaign_criteria"]) == 1 + crit = crit_service.operations[0].create + assert crit.ad_schedule.day_of_week == client.enums.DayOfWeekEnum.MONDAY + assert crit.ad_schedule.start_hour == 9 + + +# --------------------------------------------------------------------------- +# draft_promotion +# --------------------------------------------------------------------------- + + +@pytest.fixture +def stub_urls(monkeypatch): + monkeypatch.setattr( + "adloop.ads.write._validate_urls", + lambda urls, timeout=10: {u: None for u in urls}, + ) + + +def test_draft_promotion_requires_one_discount(config, stub_urls): + result = write.draft_promotion( + config, + customer_id="123-456-7890", + promotion_target="Spring Sale", + final_url="https://example.com/promo", + campaign_id="1001", + ) + assert result["error"] == "Validation failed" + assert any("money_off or percent_off" in d for d in result["details"]) + + +def test_draft_promotion_rejects_both_discounts(config, stub_urls): + result = write.draft_promotion( + config, + customer_id="123-456-7890", + promotion_target="Spring Sale", + final_url="https://example.com/promo", + money_off=10, + percent_off=15, + campaign_id="1001", + ) + assert result["error"] == "Validation failed" + assert any("not both" in d for d in result["details"]) + + +def test_draft_promotion_rejects_long_target(config, stub_urls): + result = write.draft_promotion( + config, + customer_id="123-456-7890", + promotion_target="This target is far too long for a promo", + final_url="https://example.com/promo", + percent_off=20, + campaign_id="1001", + ) + assert result["error"] == "Validation failed" + assert any("max 20" in d for d in result["details"]) + + +def test_draft_promotion_rejects_code_and_orders_over(config, stub_urls): + result = write.draft_promotion( + config, + customer_id="123-456-7890", + promotion_target="Spring Sale", + final_url="https://example.com/promo", + percent_off=20, + promotion_code="SAVE20", + orders_over_amount=50, + campaign_id="1001", + ) + assert result["error"] == "Validation failed" + assert any("mutually exclusive" in d for d in result["details"]) + + +def test_draft_promotion_ad_group_scope(config, stub_urls): + result = write.draft_promotion( + config, + customer_id="123-456-7890", + promotion_target="Spring Sale", + final_url="https://example.com/promo", + percent_off=20, + campaign_id="1001", + ad_group_id="2002", + ) + assert result["operation"] == "create_promotion" + assert result["changes"]["scope"] == "ad_group" + assert result["changes"]["promotion"]["percent_off"] == 20.0 + + +def test_apply_create_promotion_money_off_campaign_scope(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/1"), + _FakeMutateOperationResponse( + "campaign_asset_result", "customers/1234567890/campaignAssets/1" + ), + ] + google_ads_service, client = _asset_link_client(responses) + + write._apply_create_promotion( + client, + "1234567890", + { + "scope": "campaign", + "campaign_id": "1001", + "ad_group_id": "", + "promotion": { + "promotion_target": "Spring Sale", + "final_url": "https://example.com/promo", + "currency_code": "USD", + "money_off": 25.0, + "percent_off": 0.0, + "promotion_code": "", + "orders_over_amount": 0.0, + "occasion": "", + "discount_modifier": "", + "language_code": "en", + "start_date": "", + "end_date": "", + "redemption_start_date": "", + "redemption_end_date": "", + "ad_schedule": [], + }, + }, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.promotion_asset.promotion_target == "Spring Sale" + assert create.promotion_asset.money_amount_off.amount_micros == 25_000_000 + assert create.final_urls[0] == "https://example.com/promo" + link = google_ads_service.operations[1].campaign_asset_operation.create + assert link.field_type == client.enums.AssetFieldTypeEnum.PROMOTION + + +def test_apply_update_promotion_swaps_and_unlinks_old(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/9"), + _FakeMutateOperationResponse( + "campaign_asset_result", "customers/1234567890/campaignAssets/9" + ), + ] + google_ads_service = _FakeGoogleAdsService(responses) + google_ads_service.search_rows = [ + SimpleNamespace( + campaign_asset=SimpleNamespace( + resource_name="customers/1234567890/campaignAssets/OLD" + ) + ) + ] + ca_link_service = _FakeLinkService("campaignAssets") + client = _FakeClient( + { + "GoogleAdsService": google_ads_service, + "AssetService": _FakePathService("assets"), + "CampaignService": _FakePathService("campaigns"), + "CampaignAssetService": ca_link_service, + } + ) + + result = write._apply_update_promotion( + client, + "1234567890", + { + "scope": "campaign", + "campaign_id": "1001", + "old_asset_id": "555", + "promotion": { + "promotion_target": "Summer Sale", + "final_url": "https://example.com/promo", + "currency_code": "USD", + "money_off": 0.0, + "percent_off": 30.0, + "promotion_code": "", + "orders_over_amount": 0.0, + "occasion": "", + "discount_modifier": "", + "language_code": "en", + "start_date": "", + "end_date": "", + "redemption_start_date": "", + "redemption_end_date": "", + "ad_schedule": [], + }, + }, + ) + assert result["new_asset"].endswith("assets/9") + assert result["old_link_removed"].endswith("campaignAssets/OLD") + assert ca_link_service.operations[0].remove.endswith("campaignAssets/OLD") + + +# --------------------------------------------------------------------------- +# draft_price_asset +# --------------------------------------------------------------------------- + + +def _offerings(n=3): + return [ + { + "header": f"Service {i}", + "description": f"Desc {i}", + "price": 100 + i, + "final_url": f"https://example.com/service-{i}", + } + for i in range(n) + ] + + +def test_draft_price_asset_requires_three_offerings(config, stub_urls): + result = write.draft_price_asset( + config, + customer_id="123-456-7890", + campaign_id="1001", + offerings=_offerings(2), + ) + assert result["error"] == "Validation failed" + assert any("between 3 and 8" in d for d in result["details"]) + + +def test_draft_price_asset_rejects_duplicate_headers(config, stub_urls): + rows = _offerings(3) + rows[1]["header"] = rows[0]["header"] + result = write.draft_price_asset( + config, + customer_id="123-456-7890", + campaign_id="1001", + offerings=rows, + ) + assert result["error"] == "Validation failed" + assert any("duplicate header" in d for d in result["details"]) + + +def test_draft_price_asset_rejects_bad_type(config, stub_urls): + result = write.draft_price_asset( + config, + customer_id="123-456-7890", + campaign_id="1001", + price_type="NOPE", + offerings=_offerings(3), + ) + assert result["error"] == "Validation failed" + assert any("price_type" in d for d in result["details"]) + + +def test_draft_price_asset_customer_scope(config, stub_urls): + result = write.draft_price_asset( + config, + customer_id="123-456-7890", + offerings=_offerings(3), + ) + assert result["operation"] == "create_price_asset" + assert result["changes"]["scope"] == "customer" + assert len(result["changes"]["price"]["offerings"]) == 3 + + +def test_apply_create_price_asset_builds_offerings(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/1"), + _FakeMutateOperationResponse( + "ad_group_asset_result", "customers/1234567890/adGroupAssets/1" + ), + ] + google_ads_service, client = _asset_link_client(responses) + + write._apply_create_price_asset( + client, + "1234567890", + { + "scope": "ad_group", + "campaign_id": "", + "ad_group_id": "2002", + "price": { + "price_type": "SERVICES", + "price_qualifier": "FROM", + "language_code": "en", + "currency_code": "USD", + "offerings": [ + { + "header": "Service A", + "description": "Desc A", + "price": 149.0, + "final_url": "https://example.com/a", + "final_mobile_url": "", + "unit": "", + }, + { + "header": "Service B", + "description": "Desc B", + "price": 199.0, + "final_url": "https://example.com/b", + "final_mobile_url": "", + "unit": "PER_HOUR", + }, + { + "header": "Service C", + "description": "Desc C", + "price": 249.0, + "final_url": "https://example.com/c", + "final_mobile_url": "", + "unit": "", + }, + ], + }, + }, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.price_asset.type_ == client.enums.PriceExtensionTypeEnum.SERVICES + assert len(create.price_asset.price_offerings) == 3 + assert create.price_asset.price_offerings[0].price.amount_micros == 149_000_000 + assert ( + create.price_asset.price_offerings[1].unit + == client.enums.PriceExtensionPriceUnitEnum.PER_HOUR + ) + link = google_ads_service.operations[1].ad_group_asset_operation.create + assert link.field_type == client.enums.AssetFieldTypeEnum.PRICE + + +# --------------------------------------------------------------------------- +# In-place update tools: callout / sitelink / call asset +# --------------------------------------------------------------------------- + + +def test_update_callout_requires_asset_id(config): + result = write.update_callout(config, customer_id="123-456-7890", callout_text="Hi") + assert result["error"] == "asset_id is required" + + +def test_update_callout_rejects_long_text(config): + result = write.update_callout( + config, + customer_id="123-456-7890", + asset_id="55", + callout_text="This callout text is definitely far too long", + ) + assert result["error"] == "Validation failed" + + +def test_update_callout_preview(config): + result = write.update_callout( + config, customer_id="123-456-7890", asset_id="55", callout_text="Free Wi-Fi" + ) + assert result["operation"] == "update_callout" + assert result["changes"]["callout_text"] == "Free Wi-Fi" + + +def test_apply_update_callout_sets_field_mask(): + asset_service = _FakeAssetService() + client = _FakeClient({"AssetService": asset_service}) + result = write._apply_update_callout( + client, "1234567890", {"asset_id": "55", "callout_text": "Free Wi-Fi"} + ) + assert result["resource_name"].endswith("assets/0") + op = asset_service.operations[0] + assert op.update.callout_asset.callout_text == "Free Wi-Fi" + assert list(op.update_mask.paths) == ["callout_asset.callout_text"] + + +def test_update_sitelink_rejects_empty_change(config): + result = write.update_sitelink(config, customer_id="123-456-7890", asset_id="55") + assert result["error"] == "No fields to update" + + +def test_update_sitelink_preview(config, stub_urls): + result = write.update_sitelink( + config, + customer_id="123-456-7890", + asset_id="55", + link_text="Pricing", + final_url="https://example.com/pricing", + ) + assert result["operation"] == "update_sitelink" + assert result["changes"]["link_text"] == "Pricing" + + +def test_apply_update_sitelink_partial_mask(): + asset_service = _FakeAssetService() + client = _FakeClient({"AssetService": asset_service}) + write._apply_update_sitelink( + client, + "1234567890", + { + "asset_id": "55", + "link_text": "Pricing", + "final_url": "https://example.com/pricing", + }, + ) + op = asset_service.operations[0] + assert op.update.sitelink_asset.link_text == "Pricing" + assert op.update.final_urls[0] == "https://example.com/pricing" + assert set(op.update_mask.paths) == {"sitelink_asset.link_text", "final_urls"} + + +def test_update_call_asset_rejects_empty_change(config): + result = write.update_call_asset(config, customer_id="123-456-7890", asset_id="55") + assert result["error"] == "No fields to update" + + +def test_update_call_asset_rejects_bad_reporting_state(config): + result = write.update_call_asset( + config, + customer_id="123-456-7890", + asset_id="55", + call_conversion_reporting_state="NONSENSE", + ) + assert result["error"] == "Validation failed" + + +def test_update_call_asset_preview_normalizes_phone(config): + result = write.update_call_asset( + config, + customer_id="123-456-7890", + asset_id="55", + phone_number="(555) 555-0142", + country_code="US", + ) + assert result["changes"]["phone_number"] == "+15555550142" + assert result["changes"]["country_code"] == "US" + + +def test_apply_update_call_asset_builds_field_mask(): + asset_service = _FakeAssetService() + client = _FakeClient( + { + "AssetService": asset_service, + "ConversionActionService": _FakePathService("conversionActions"), + } + ) + write._apply_update_call_asset( + client, + "1234567890", + { + "asset_id": "55", + "phone_number": "+15555550142", + "call_conversion_action_id": "999", + "call_conversion_reporting_state": "USE_RESOURCE_LEVEL_CALL_CONVERSION_ACTION", + }, + ) + op = asset_service.operations[0] + assert op.update.call_asset.phone_number == "+15555550142" + assert op.update.call_asset.call_conversion_action.endswith( + "conversionActions/999" + ) + assert set(op.update_mask.paths) == { + "call_asset.phone_number", + "call_asset.call_conversion_action", + "call_asset.call_conversion_reporting_state", + } + + +# --------------------------------------------------------------------------- +# update_structured_snippet (swap) +# --------------------------------------------------------------------------- + + +def test_update_structured_snippet_requires_asset_id(config): + result = write.update_structured_snippet( + config, + customer_id="123-456-7890", + header="Brands", + values=["A", "B", "C"], + ) + assert "asset_id is required" in result["error"] + + +def test_update_structured_snippet_rejects_bad_header(config): + result = write.update_structured_snippet( + config, + customer_id="123-456-7890", + asset_id="55", + header="Nope", + values=["A", "B", "C"], + ) + assert result["error"] == "Validation failed" + + +def test_update_structured_snippet_ad_group_scope_preview(config): + result = write.update_structured_snippet( + config, + customer_id="123-456-7890", + asset_id="55", + campaign_id="1001", + ad_group_id="2002", + header="Services", + values=["Repair", "Install", "Maintain"], + ) + assert result["operation"] == "update_structured_snippet" + assert result["changes"]["scope"] == "ad_group" + assert result["changes"]["old_asset_id"] == "55" + + +def test_apply_update_structured_snippet_swaps_ad_group_link(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/9"), + _FakeMutateOperationResponse( + "ad_group_asset_result", "customers/1234567890/adGroupAssets/9" + ), + ] + google_ads_service = _FakeGoogleAdsService(responses) + google_ads_service.search_rows = [ + SimpleNamespace( + ad_group_asset=SimpleNamespace( + resource_name="customers/1234567890/adGroupAssets/OLD" + ) + ) + ] + aga_service = _FakeLinkService("adGroupAssets") + client = _FakeClient( + { + "GoogleAdsService": google_ads_service, + "AssetService": _FakePathService("assets"), + "CampaignService": _FakePathService("campaigns"), + "AdGroupAssetService": aga_service, + } + ) + + result = write._apply_update_structured_snippet( + client, + "1234567890", + { + "scope": "ad_group", + "campaign_id": "1001", + "ad_group_id": "2002", + "old_asset_id": "55", + "snippet": {"header": "Services", "values": ["Repair", "Install", "Maintain"]}, + }, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.structured_snippet_asset.header == "Services" + assert list(create.structured_snippet_asset.values) == [ + "Repair", + "Install", + "Maintain", + ] + assert result["old_link_removed"].endswith("adGroupAssets/OLD") + assert aga_service.operations[0].remove.endswith("adGroupAssets/OLD") + + +# --------------------------------------------------------------------------- +# draft_business_name_asset +# --------------------------------------------------------------------------- + + +def test_draft_business_name_requires_name(config): + result = write.draft_business_name_asset(config, customer_id="123-456-7890") + assert result["error"] == "business_name is required" + + +def test_draft_business_name_rejects_long_name(config): + result = write.draft_business_name_asset( + config, + customer_id="123-456-7890", + business_name="A Business Name That Is Way Too Long", + ) + assert result["error"] == "Validation failed" + + +def test_draft_business_name_campaign_scope(config): + result = write.draft_business_name_asset( + config, + customer_id="123-456-7890", + campaign_id="1001", + business_name="Example Plumbing", + ) + assert result["operation"] == "create_business_name_asset" + assert result["changes"]["scope"] == "campaign" + + +def test_apply_create_business_name_asset_customer_scope(): + responses = [ + _FakeMutateOperationResponse("asset_result", "customers/1234567890/assets/1"), + _FakeMutateOperationResponse( + "customer_asset_result", "customers/1234567890/customerAssets/1" + ), + ] + google_ads_service, client = _asset_link_client(responses) + + result = write._apply_create_business_name_asset( + client, + "1234567890", + {"scope": "customer", "campaign_id": "", "business_name": "Example Plumbing"}, + ) + create = google_ads_service.operations[0].asset_operation.create + assert create.text_asset.text == "Example Plumbing" + assert create.type_ == client.enums.AssetTypeEnum.TEXT + link = google_ads_service.operations[1].customer_asset_operation.create + assert link.field_type == client.enums.AssetFieldTypeEnum.BUSINESS_NAME + assert result["link"].endswith("customerAssets/1") + + +# --------------------------------------------------------------------------- +# draft_location_asset +# --------------------------------------------------------------------------- + + +def test_draft_location_asset_requires_gbp_id(config): + result = write.draft_location_asset(config, customer_id="123-456-7890") + assert "business_profile_account_id is required" in result["error"] + + +def test_draft_location_asset_default_name_and_warning(config): + result = write.draft_location_asset( + config, + customer_id="123-456-7890", + business_profile_account_id="9988776655", + ) + assert result["operation"] == "create_location_asset" + assert result["changes"]["asset_set_name"] == "GBP Locations - 9988776655" + assert result["changes"]["scope"] == "customer" + assert result["warnings"] + + +def test_apply_create_location_asset_customer_scope(): + asset_set_service = _FakeAssetSetService() + cas_service = _FakeAssetSetLinkService("customerAssetSets") + client = _FakeClient( + { + "AssetSetService": asset_set_service, + "CustomerAssetSetService": cas_service, + } + ) + result = write._apply_create_location_asset( + client, + "1234567890", + { + "scope": "customer", + "campaign_id": "", + "business_profile_account_id": "9988776655", + "asset_set_name": "GBP Locations - 9988776655", + "label_filters": ["storefront"], + "listing_id_filters": ["12345"], + }, + ) + set_create = asset_set_service.operations[0].create + assert set_create.type_ == client.enums.AssetSetTypeEnum.LOCATION_SYNC + bpls = set_create.location_set.business_profile_location_set + assert bpls.business_account_id == "9988776655" + assert list(bpls.label_filters) == ["storefront"] + assert list(bpls.listing_id_filters) == [12345] + assert result["asset_set"].endswith("assetSets/1") + assert result["customer_asset_set"].endswith("customerAssetSets/1") + + +def test_apply_create_location_asset_campaign_scope(): + asset_set_service = _FakeAssetSetService() + cas_service = _FakeAssetSetLinkService("campaignAssetSets") + client = _FakeClient( + { + "AssetSetService": asset_set_service, + "CampaignAssetSetService": cas_service, + "CampaignService": _FakePathService("campaigns"), + } + ) + result = write._apply_create_location_asset( + client, + "1234567890", + { + "scope": "campaign", + "campaign_id": "1001", + "business_profile_account_id": "9988776655", + "asset_set_name": "GBP Locations", + "label_filters": [], + "listing_id_filters": [], + }, + ) + assert result["campaign_asset_set"].endswith("campaignAssetSets/1") + link = cas_service.operations[0].create + assert link.campaign.endswith("campaigns/1001") + + +# --------------------------------------------------------------------------- +# link_asset_to_customer +# --------------------------------------------------------------------------- + + +def test_link_asset_requires_links(config): + result = write.link_asset_to_customer(config, customer_id="123-456-7890") + assert result["error"] == "At least one link is required" + + +def test_link_asset_rejects_non_numeric_id(config): + result = write.link_asset_to_customer( + config, + customer_id="123-456-7890", + links=[{"asset_id": "abc", "field_type": "BUSINESS_LOGO"}], + ) + assert result["error"] == "Validation failed" + assert any("must be numeric" in d for d in result["details"]) + + +def test_link_asset_rejects_bad_field_type(config): + result = write.link_asset_to_customer( + config, + customer_id="123-456-7890", + links=[{"asset_id": "555", "field_type": "NOT_A_TYPE"}], + ) + assert result["error"] == "Validation failed" + assert any("not valid for" in d for d in result["details"]) + + +def test_link_asset_preview_ok(config): + result = write.link_asset_to_customer( + config, + customer_id="123-456-7890", + links=[{"asset_id": "555", "field_type": "business_logo"}], + ) + assert result["operation"] == "link_asset_to_customer" + assert result["changes"]["links"][0]["field_type"] == "BUSINESS_LOGO" + + +def test_apply_link_asset_to_customer_builds_links(): + cust_service = _FakeCustomerAssetService() + client = _FakeClient( + { + "AssetService": _FakePathService("assets"), + "CustomerAssetService": cust_service, + } + ) + result = write._apply_link_asset_to_customer( + client, + "1234567890", + { + "links": [ + {"asset_id": "555", "field_type": "BUSINESS_LOGO"}, + {"asset_id": "666", "field_type": "MARKETING_IMAGE"}, + ] + }, + ) + assert result["linked_count"] == 2 + op0 = cust_service.operations[0].create + assert op0.asset.endswith("assets/555") + assert op0.field_type == client.enums.AssetFieldTypeEnum.BUSINESS_LOGO