diff --git a/src/adloop/ads/conversion_actions.py b/src/adloop/ads/conversion_actions.py new file mode 100644 index 0000000..9e4c204 --- /dev/null +++ b/src/adloop/ads/conversion_actions.py @@ -0,0 +1,1413 @@ +"""Conversion-action write tools — Google Ads ConversionActionService. + +All operations follow the AdLoop safety pattern: + 1. draft_* → creates a ChangePlan, stores it, returns plan_id + 2. confirm_and_apply(plan_id) → executes via the Google Ads API + +Supported types (conversion_action.type): + AD_CALL — calls from Call assets in ads + WEBSITE_CALL — Google Forwarding Number calls (uses + phone_call_duration_seconds threshold) + WEBPAGE — page-load conversions with code-based tracking + WEBPAGE_CODELESS — page-load conversions detected by Ads (no snippet) + GOOGLE_ANALYTICS_4_CUSTOM — imported from GA4 (custom event) + GOOGLE_ANALYTICS_4_PURCHASE — imported from GA4 (purchase event) + UPLOAD_CALLS, UPLOAD_CLICKS — offline imports + +NOT supported here (Google manages them — mutations are rejected with +MUTATE_NOT_ALLOWED): + SMART_CAMPAIGN_* — auto-created by Smart Campaigns + GOOGLE_HOSTED — auto-created by Google Business Profile / LSA links +""" +from __future__ import annotations + +import hashlib +from typing import TYPE_CHECKING + +from adloop.ads.enums import enum_names + +if TYPE_CHECKING: + from adloop.config import AdLoopConfig + + +# Pulled dynamically from the google-ads SDK at the API version we're +# pinned to (see adloop.ads.client.GOOGLE_ADS_API_VERSION). Keeps the +# validators in sync with whatever the SDK supports — no hand-maintained +# parallel lists to drift. +_VALID_TYPES = enum_names("ConversionActionTypeEnum") +_VALID_CATEGORIES = enum_names("ConversionActionCategoryEnum") +_VALID_COUNTING_TYPES = enum_names("ConversionActionCountingTypeEnum") +_VALID_ATTRIBUTION_MODELS = enum_names("AttributionModelEnum") + +# These types ARE in ConversionActionTypeEnum but Google rejects mutations +# on them with MUTATE_NOT_ALLOWED (they're auto-created by Smart Campaigns, +# Local Services, and Business Profile links). We don't filter them from +# `_VALID_TYPES` — the SDK accepts them syntactically — but warn callers. +_AUTO_MANAGED_TYPES = frozenset({ + "SMART_CAMPAIGN_TRACKED_CALLS", + "SMART_CAMPAIGN_MAP_DIRECTIONS", + "SMART_CAMPAIGN_MAP_CLICKS_TO_CALL", + "SMART_CAMPAIGN_AD_CLICKS_TO_CALL", + "GOOGLE_HOSTED", +}) + + +# --------------------------------------------------------------------------- +# Validators +# --------------------------------------------------------------------------- + + +def _validate_create_inputs( + *, + name: str, + type_: str, + category: str, + counting_type: str, + default_value: float, + currency_code: str, + phone_call_duration_seconds: int, + click_through_window_days: int, + view_through_window_days: int, + attribution_model: str, +) -> list[str]: + errors: list[str] = [] + if not name or not name.strip(): + errors.append("name is required") + if type_ not in _VALID_TYPES: + errors.append( + f"type '{type_}' invalid; valid: {sorted(_VALID_TYPES)}" + ) + if category and category not in _VALID_CATEGORIES: + errors.append( + f"category '{category}' invalid; valid: {sorted(_VALID_CATEGORIES)}" + ) + if counting_type and counting_type not in _VALID_COUNTING_TYPES: + errors.append( + f"counting_type '{counting_type}' invalid; valid: " + f"{sorted(_VALID_COUNTING_TYPES)}" + ) + if default_value < 0: + errors.append("default_value must be >= 0") + if currency_code and len(currency_code) != 3: + errors.append( + f"currency_code '{currency_code}' must be a 3-letter ISO code" + ) + if phone_call_duration_seconds and phone_call_duration_seconds < 0: + errors.append("phone_call_duration_seconds must be >= 0") + if (click_through_window_days + and not (1 <= click_through_window_days <= 90)): + errors.append( + "click_through_window_days must be between 1 and 90" + ) + if (view_through_window_days + and not (1 <= view_through_window_days <= 30)): + errors.append( + "view_through_window_days must be between 1 and 30" + ) + if attribution_model and attribution_model not in _VALID_ATTRIBUTION_MODELS: + errors.append( + f"attribution_model '{attribution_model}' invalid; valid: " + f"{sorted(_VALID_ATTRIBUTION_MODELS)}" + ) + return errors + + +def _validate_update_inputs( + *, + counting_type: str, + default_value: float, + currency_code: str, + phone_call_duration_seconds: int, + click_through_window_days: int, + view_through_window_days: int, + attribution_model: str, +) -> list[str]: + errors: list[str] = [] + if counting_type and counting_type not in _VALID_COUNTING_TYPES: + errors.append( + f"counting_type '{counting_type}' invalid; valid: " + f"{sorted(_VALID_COUNTING_TYPES)}" + ) + if default_value < 0: + errors.append("default_value must be >= 0") + if currency_code and len(currency_code) != 3: + errors.append( + f"currency_code '{currency_code}' must be a 3-letter ISO code" + ) + if phone_call_duration_seconds and phone_call_duration_seconds < 0: + errors.append("phone_call_duration_seconds must be >= 0") + if (click_through_window_days + and not (1 <= click_through_window_days <= 90)): + errors.append( + "click_through_window_days must be between 1 and 90" + ) + if (view_through_window_days + and not (1 <= view_through_window_days <= 30)): + errors.append( + "view_through_window_days must be between 1 and 30" + ) + if attribution_model and attribution_model not in _VALID_ATTRIBUTION_MODELS: + errors.append( + f"attribution_model '{attribution_model}' invalid; valid: " + f"{sorted(_VALID_ATTRIBUTION_MODELS)}" + ) + return errors + + +# --------------------------------------------------------------------------- +# Draft tools (return PREVIEW + plan_id) +# --------------------------------------------------------------------------- + + +def draft_create_conversion_action( + config: AdLoopConfig, + *, + customer_id: str = "", + name: str, + type_: str, + category: str = "DEFAULT", + default_value: float = 0, + currency_code: str = "USD", + always_use_default_value: bool = False, + counting_type: str = "ONE_PER_CLICK", + phone_call_duration_seconds: int = 0, + primary_for_goal: bool = True, + include_in_conversions_metric: bool = True, + click_through_window_days: int = 0, + view_through_window_days: int = 0, + attribution_model: str = "", +) -> dict: + """Draft a new ConversionAction — returns a PREVIEW. + + type_: the ConversionAction.type enum value (AD_CALL, WEBSITE_CALL, + WEBPAGE, WEBPAGE_CODELESS, GOOGLE_ANALYTICS_4_CUSTOM, etc.). + category: the conversion category (PHONE_CALL_LEAD, SUBMIT_LEAD_FORM, + PURCHASE, etc.). Defaults to DEFAULT. + default_value: monetary value attributed to each conversion. + always_use_default_value: when True, transaction values from the + snippet/import are ignored and default_value is used instead. When + False with a positive default_value, Google treats default_value as + a fallback ("tag value with fallback"). Passing a positive + default_value with this flag False is a legal config — the draft + surfaces a warning (see below) but does NOT flip the flag for you. + counting_type: ONE_PER_CLICK (recommended for lead gen — one click, + one conversion no matter how many events fire) or MANY_PER_CLICK + (better for ecommerce where multiple purchases per click are real). + phone_call_duration_seconds: ONLY meaningful for PHONE_CALL_LEAD + category. The call must last at least this many seconds to count. + primary_for_goal: True = drives Smart Bidding optimization; + False = Secondary (records but doesn't affect bidding). + include_in_conversions_metric: True (default) = appears in the + "Conversions" column; False = "All conversions" only. NOTE: this is + IMMUTABLE on create — Google derives it from the category and rejects + any value set in the create mutate. To change it, use + draft_update_conversion_action after the create succeeds. + click_through_window_days / view_through_window_days: attribution + windows. 30/1 is the typical lead-gen pair. + attribution_model: leave empty for the default. For data-driven, + pass GOOGLE_SEARCH_ATTRIBUTION_DATA_DRIVEN. + + 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_conversion_action", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + errors = _validate_create_inputs( + name=name, + type_=type_, + category=category, + counting_type=counting_type, + default_value=default_value, + currency_code=currency_code, + phone_call_duration_seconds=phone_call_duration_seconds, + click_through_window_days=click_through_window_days, + view_through_window_days=view_through_window_days, + attribution_model=attribution_model, + ) + if errors: + return {"error": "Validation failed", "details": errors} + + warnings: list[str] = [] + + # A positive default_value paired with always_use_default_value=False is a + # LEGAL config: Google treats default_value as a fallback when the + # snippet/import supplies no value ("tag value with fallback"). We used to + # silently force the flag to True, which turned that fallback config into + # an unconditional override — a real change in accounting the caller never + # asked for. Surface it as a preview warning instead and leave the flag + # exactly as the caller set it. + if default_value > 0 and not always_use_default_value: + warnings.append( + "default_value is set but always_use_default_value is False: " + "Google will treat default_value as a FALLBACK, used only when " + "the tag/import provides no value. If you want default_value to " + "override every conversion's value, set " + "always_use_default_value=True explicitly." + ) + + if type_ in _AUTO_MANAGED_TYPES: + warnings.append( + f"type '{type_}' is auto-managed by Google (Smart Campaigns / " + "Business Profile). Mutations are rejected with MUTATE_NOT_ALLOWED." + ) + + plan = ChangePlan( + operation="create_conversion_action", + entity_type="conversion_action", + entity_id="", + customer_id=customer_id, + changes={ + "name": name.strip(), + "type": type_, + "category": category, + "default_value": float(default_value), + "currency_code": currency_code.upper(), + "always_use_default_value": bool(always_use_default_value), + "counting_type": counting_type, + "phone_call_duration_seconds": int(phone_call_duration_seconds or 0), + "primary_for_goal": bool(primary_for_goal), + "include_in_conversions_metric": bool(include_in_conversions_metric), + "click_through_window_days": int(click_through_window_days or 0), + "view_through_window_days": int(view_through_window_days or 0), + "attribution_model": attribution_model, + }, + ) + store_plan(plan) + preview = plan.to_preview() + if warnings: + preview["warnings"] = warnings + return preview + + +def draft_update_conversion_action( + config: AdLoopConfig, + *, + customer_id: str = "", + conversion_action_id: str, + name: str = "", + primary_for_goal: bool | None = None, + default_value: float = 0, + currency_code: str = "", + always_use_default_value: bool | None = None, + counting_type: str = "", + phone_call_duration_seconds: int = 0, + include_in_conversions_metric: bool | None = None, + click_through_window_days: int = 0, + view_through_window_days: int = 0, + attribution_model: str = "", +) -> dict: + """Draft a partial UPDATE of an existing ConversionAction — returns PREVIEW. + + Only the parameters you pass non-empty/non-default will be sent to the + API. Use this to rename, demote a Primary to Secondary, change value, + adjust the call-duration threshold, or change attribution settings. + include_in_conversions_metric IS mutable here (unlike on create). + + conversion_action_id: numeric ID. Find via: + SELECT conversion_action.id, conversion_action.name FROM conversion_action + + Note: Google rejects mutations on SMART_CAMPAIGN_* and GOOGLE_HOSTED + types with MUTATE_NOT_ALLOWED. Catch and report this at apply time. + + 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_conversion_action", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not conversion_action_id: + return {"error": "conversion_action_id is required"} + + errors = _validate_update_inputs( + counting_type=counting_type, + default_value=default_value, + currency_code=currency_code, + phone_call_duration_seconds=phone_call_duration_seconds, + click_through_window_days=click_through_window_days, + view_through_window_days=view_through_window_days, + attribution_model=attribution_model, + ) + if errors: + return {"error": "Validation failed", "details": errors} + + # Track which fields the caller actually wants to update so we build + # the right field_mask at apply time. + changes: dict = {"conversion_action_id": str(conversion_action_id)} + if name: + changes["name"] = name.strip() + if primary_for_goal is not None: + changes["primary_for_goal"] = bool(primary_for_goal) + if default_value: + changes["default_value"] = float(default_value) + if currency_code: + changes["currency_code"] = currency_code.upper() + if always_use_default_value is not None: + changes["always_use_default_value"] = bool(always_use_default_value) + if counting_type: + changes["counting_type"] = counting_type + if phone_call_duration_seconds: + changes["phone_call_duration_seconds"] = int(phone_call_duration_seconds) + if include_in_conversions_metric is not None: + changes["include_in_conversions_metric"] = bool( + include_in_conversions_metric + ) + if click_through_window_days: + changes["click_through_window_days"] = int(click_through_window_days) + if view_through_window_days: + changes["view_through_window_days"] = int(view_through_window_days) + if attribution_model: + changes["attribution_model"] = attribution_model + + if len(changes) == 1: # only conversion_action_id + return {"error": "No fields to update"} + + warnings: list[str] = [] + # Same fallback-vs-override nuance as create: on update, a positive + # default_value with always_use_default_value explicitly set to False is + # legal (fallback). Warn rather than silently overriding intent. + if changes.get("default_value", 0) > 0 and ( + changes.get("always_use_default_value") is False + ): + warnings.append( + "default_value is set but always_use_default_value is False: " + "Google will treat default_value as a FALLBACK, used only when " + "the tag/import provides no value. Set " + "always_use_default_value=True explicitly to override every " + "conversion's value." + ) + + plan = ChangePlan( + operation="update_conversion_action", + entity_type="conversion_action", + entity_id=str(conversion_action_id), + customer_id=customer_id, + changes=changes, + ) + store_plan(plan) + preview = plan.to_preview() + if warnings: + preview["warnings"] = warnings + return preview + + +def draft_remove_conversion_action( + config: AdLoopConfig, + *, + customer_id: str = "", + conversion_action_id: str, +) -> dict: + """Draft a REMOVAL of a ConversionAction — returns PREVIEW. + + Removed conversion actions stop counting and disappear from goal lists. + Historical data is preserved. SMART_CAMPAIGN_* and GOOGLE_HOSTED types + cannot be removed via API (Google manages them); the apply will fail + with MUTATE_NOT_ALLOWED for those. + + 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("remove_conversion_action", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + if not conversion_action_id: + return {"error": "conversion_action_id is required"} + + plan = ChangePlan( + operation="remove_conversion_action", + entity_type="conversion_action", + entity_id=str(conversion_action_id), + customer_id=customer_id, + changes={"conversion_action_id": str(conversion_action_id)}, + ) + store_plan(plan) + preview = plan.to_preview() + preview["warnings"] = [ + "Removing a ConversionAction is irreversible. Smart Campaign / GBP-" + "managed types reject mutation with MUTATE_NOT_ALLOWED." + ] + return preview + + +# --------------------------------------------------------------------------- +# Apply handlers +# --------------------------------------------------------------------------- + + +def _apply_create_conversion_action(client: object, cid: str, changes: dict) -> dict: + """Create a new ConversionAction.""" + svc = client.get_service("ConversionActionService") + op = client.get_type("ConversionActionOperation") + ca = op.create + ca.name = changes["name"] + ca.type_ = getattr(client.enums.ConversionActionTypeEnum, changes["type"]) + ca.category = getattr( + client.enums.ConversionActionCategoryEnum, changes["category"] + ) + ca.status = client.enums.ConversionActionStatusEnum.ENABLED + ca.counting_type = getattr( + client.enums.ConversionActionCountingTypeEnum, changes["counting_type"] + ) + ca.value_settings.default_value = changes["default_value"] + ca.value_settings.default_currency_code = changes["currency_code"] + ca.value_settings.always_use_default_value = changes["always_use_default_value"] + ca.primary_for_goal = changes["primary_for_goal"] + # NOTE: include_in_conversions_metric is IMMUTABLE on create — Google + # derives it from the conversion category and rejects any value set in + # the create mutate (IMMUTABLE_FIELD). To change it, use + # draft_update_conversion_action after the create succeeds. + if changes.get("phone_call_duration_seconds"): + ca.phone_call_duration_seconds = changes["phone_call_duration_seconds"] + if changes.get("click_through_window_days"): + ca.click_through_lookback_window_days = changes["click_through_window_days"] + if changes.get("view_through_window_days"): + ca.view_through_lookback_window_days = changes["view_through_window_days"] + if changes.get("attribution_model"): + ca.attribution_model_settings.attribution_model = getattr( + client.enums.AttributionModelEnum, changes["attribution_model"] + ) + + response = svc.mutate_conversion_actions( + customer_id=cid, operations=[op] + ) + return {"resource_name": response.results[0].resource_name} + + +def _apply_update_conversion_action(client: object, cid: str, changes: dict) -> dict: + """Partial update of an existing ConversionAction. + + Builds a FieldMask listing only the fields the caller wanted to update. + """ + from google.protobuf import field_mask_pb2 + + svc = client.get_service("ConversionActionService") + op = client.get_type("ConversionActionOperation") + ca = op.update + ca.resource_name = svc.conversion_action_path( + cid, changes["conversion_action_id"] + ) + + paths: list[str] = [] + + if "name" in changes: + ca.name = changes["name"] + paths.append("name") + if "primary_for_goal" in changes: + ca.primary_for_goal = changes["primary_for_goal"] + paths.append("primary_for_goal") + if "default_value" in changes: + ca.value_settings.default_value = changes["default_value"] + paths.append("value_settings.default_value") + if "currency_code" in changes: + ca.value_settings.default_currency_code = changes["currency_code"] + paths.append("value_settings.default_currency_code") + if "always_use_default_value" in changes: + ca.value_settings.always_use_default_value = changes["always_use_default_value"] + paths.append("value_settings.always_use_default_value") + if "counting_type" in changes: + ca.counting_type = getattr( + client.enums.ConversionActionCountingTypeEnum, changes["counting_type"] + ) + paths.append("counting_type") + if "phone_call_duration_seconds" in changes: + ca.phone_call_duration_seconds = changes["phone_call_duration_seconds"] + paths.append("phone_call_duration_seconds") + if "include_in_conversions_metric" in changes: + ca.include_in_conversions_metric = changes["include_in_conversions_metric"] + paths.append("include_in_conversions_metric") + if "click_through_window_days" in changes: + ca.click_through_lookback_window_days = changes["click_through_window_days"] + paths.append("click_through_lookback_window_days") + if "view_through_window_days" in changes: + ca.view_through_lookback_window_days = changes["view_through_window_days"] + paths.append("view_through_lookback_window_days") + if "attribution_model" in changes: + ca.attribution_model_settings.attribution_model = getattr( + client.enums.AttributionModelEnum, changes["attribution_model"] + ) + paths.append("attribution_model_settings.attribution_model") + + op.update_mask.CopyFrom(field_mask_pb2.FieldMask(paths=paths)) + response = svc.mutate_conversion_actions( + customer_id=cid, operations=[op] + ) + return {"resource_name": response.results[0].resource_name} + + +def _apply_remove_conversion_action(client: object, cid: str, changes: dict) -> dict: + """Remove a ConversionAction (sets status=REMOVED).""" + svc = client.get_service("ConversionActionService") + op = client.get_type("ConversionActionOperation") + op.remove = svc.conversion_action_path( + cid, changes["conversion_action_id"] + ) + response = svc.mutate_conversion_actions( + customer_id=cid, operations=[op] + ) + return {"resource_name": response.results[0].resource_name} + + +# =========================================================================== +# Offline conversion uploads — ConversionUploadService +# =========================================================================== +# +# Two upload paths live here: +# +# 1. Call conversions (UploadCallConversions) — matches phone calls back to +# ad clicks by caller_id (E.164 phone). The caller_id is REQUIRED raw by +# Google for matching and CANNOT be hashed. It is stored in the plan so +# apply can rebuild the upload without re-reading the CSV, but it is +# REDACTED in the audit log and preview display (see _redact_caller_id +# and _redact_changes_for_audit in write.py). +# +# 2. Enhanced Conversions for Leads (UploadClickConversions with +# user_identifiers) — matches hashed PII (email / phone / name) back to +# logged-in Google users who clicked our ads. PII is normalized and +# SHA-256-hashed AT PREVIEW TIME; only the hashes are stored in the plan. +# Raw PII never lands in plan.changes and never reaches the audit log. +# +# Security invariant shared by both: apply builds the upload protos from +# ``plan.changes["rows"]`` (frozen at preview time), NOT by re-reading the +# CSV. What you previewed is exactly what gets uploaded, and no raw PII is +# re-read at apply time. +# --------------------------------------------------------------------------- + + +def _sha256_hex(value: str) -> str: + """SHA-256 a UTF-8 string, return lowercase hex. Empty in → empty out.""" + if not value: + return "" + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _normalize_email(email: str) -> str: + """Normalize an email for Enhanced Conversions: trim + lowercase. + + Google's canonicalization for EC is trim + lowercase. (Gmail dot/plus + stripping is NOT applied by Google's EC matcher — it matches on the + literal normalized address — so we deliberately do NOT strip dots or + +tags. Doing so would REDUCE the match rate.) + """ + return (email or "").strip().lower() + + +def _normalize_name(name: str) -> str: + """Normalize a first/last name for EC: trim + lowercase.""" + return (name or "").strip().lower() + + +def _normalize_phone_e164(phone: str) -> str: + """Best-effort E.164 normalization for a phone number. + + Rules (deliberately conservative — Google requires E.164 for EC phone + hashing and CallConversion.caller_id): + * Strip spaces, hyphens, parens, dots. + * A leading "00" is the international-access prefix → replace with "+". + * A single leading domestic trunk "0" (common in EU national format, + e.g. UK "020 7946 0018") is dropped — but ONLY one zero, and ONLY + when there's no "+" already. We do NOT strip every leading zero. + * Italy is the notable exception: Italian fixed-line numbers KEEP their + leading 0 in E.164 (e.g. Rome "+39 06 …"). We can't reliably detect + country from a bare national number, so the safe, documented rule is: + if the number already carries a country code (starts with "+"), we + never touch interior digits. A bare Italian number passed without a + "+" can't be disambiguated here — callers should pass Italian numbers + in full "+39…" form. This keeps the common EU trunk-zero case correct + without corrupting Italy's retained-zero numbers that arrive as "+39…". + + Returns the number with a leading "+" when we could infer one; otherwise + returns the cleaned digits unchanged (Google will reject a non-E.164 + number, surfaced as a per-row failure rather than silently mangled). + """ + s = (phone or "").strip() + if not s: + return "" + has_plus = s.startswith("+") + digits = "".join(ch for ch in s if ch.isdigit()) + if not digits: + return "" + if has_plus: + # Already carries a country code — trust it verbatim (this is the + # path that preserves Italy's retained leading zero, e.g. +3906…). + return "+" + digits + # "00" international access prefix → "+" + if digits.startswith("00"): + return "+" + digits[2:] + # Strip EXACTLY ONE domestic trunk zero (national dialing format). + if digits.startswith("0"): + return digits[1:] + return digits + + +def _gaql_escape(s: str) -> str: + """Escape a string literal for interpolation into a GAQL WHERE clause. + + GAQL uses BACKSLASH escaping (NOT SQL-style doubled quotes). Escape the + backslash first, then the single quote. Handles names like ``O'Brien``. + """ + return s.replace("\\", "\\\\").replace("'", "\\'") + + +def _consent_from_param(consent: dict | None) -> dict | None: + """Validate + normalize the ``consent`` tool parameter. + + Accepts ``{"ad_user_data": "GRANTED"|"DENIED"|"UNSPECIFIED", + "ad_personalization": ...}``. Missing keys default to UNSPECIFIED. + Returns a plain dict stored in the plan (JSON-safe, non-PII), or None + if no consent was supplied at all. + """ + if not consent: + return None + valid = {"UNSPECIFIED", "UNKNOWN", "GRANTED", "DENIED"} + out: dict[str, str] = {} + for key in ("ad_user_data", "ad_personalization"): + raw = str(consent.get(key, "UNSPECIFIED") or "UNSPECIFIED").upper() + if raw not in valid: + raise ValueError( + f"consent.{key}='{raw}' is invalid. Use one of: " + "GRANTED, DENIED, UNSPECIFIED." + ) + out[key] = raw + return out + + +def _apply_consent(client: object, conversion: object, consent: dict | None) -> None: + """Set conversion.consent.{ad_user_data,ad_personalization} from a plan dict. + + Maps the stored string values to the ConsentStatus enum. A None/empty + consent leaves the proto default (UNSPECIFIED) — which is the correct + "not provided" signal for Google. + """ + if not consent: + return + status_enum = client.enums.ConsentStatusEnum + aud = consent.get("ad_user_data", "UNSPECIFIED") + ap = consent.get("ad_personalization", "UNSPECIFIED") + conversion.consent.ad_user_data = getattr(status_enum, aud) + conversion.consent.ad_personalization = getattr(status_enum, ap) + + +# --------------------------------------------------------------------------- +# Timestamp + CSV parsing shared with the call-conversion path +# --------------------------------------------------------------------------- + +_EXPECTED_CALL_HEADERS = [ + "Caller's Phone Number", + "Call Start Time", + "Conversion Name", + "Conversion Time", + "Conversion Value", + "Conversion Currency", +] + + +def _normalize_call_timestamp(ts: str) -> str: + """Google Ads API wants 'yyyy-mm-dd HH:MM:SS+|-HH:MM'. + + Our CSV writes ISO 8601 with 'T' separator and trailing 'Z' + (e.g. '2026-02-26T16:49:44.567Z'). Convert: strip fractional + seconds, replace 'T' with space, replace 'Z' with '+00:00'. + """ + s = (ts or "").strip() + if not s: + return s + if "." in s: + head, tail = s.split(".", 1) + tz = "" + for marker in ("+", "-", "Z"): + idx = tail.find(marker) + if idx >= 0: + tz = tail[idx:] + break + s = head + (tz or "") + s = s.replace("T", " ") + if s.endswith("Z"): + s = s[:-1] + "+00:00" + return s + + +def _parse_call_conversion_csv(csv_path: str) -> tuple[list[dict], list[str]]: + """Read the AdLoop-generated phone-conversions CSV. + + Returns (rows, errors). Rows are dicts keyed by canonical column name. + Skips the optional `Parameters:TimeZone=...` row at the top. + + The ``caller_id`` (E.164 phone) is retained RAW — Google requires it for + call-to-click matching and it cannot be hashed. Callers that persist + these rows (the draft path does, into the plan) are responsible for + redacting caller_id from any audit log or preview surface. + """ + import csv + from pathlib import Path + + errors: list[str] = [] + path = Path(csv_path).expanduser() + if not path.exists(): + return [], [f"CSV not found at {path}"] + + with path.open("r", newline="") as f: + reader = csv.reader(f) + rows_iter = iter(reader) + header: list[str] | None = None + for raw in rows_iter: + if not raw: + continue + first = (raw[0] or "").strip() + if first.startswith(("Parameters:", "#", "###")): + continue + header = [c.strip() for c in raw] + break + if header is None: + return [], ["CSV is empty (no header row found)"] + + missing = [c for c in _EXPECTED_CALL_HEADERS if c not in header] + if missing: + errors.append( + f"CSV missing required columns: {missing}. Got: {header}" + ) + return [], errors + + col = {name: header.index(name) for name in _EXPECTED_CALL_HEADERS} + out: list[dict] = [] + for line_num, raw in enumerate(rows_iter, start=2): + if not raw or all((c or "").strip() == "" for c in raw): + continue + if (raw[0] or "").strip().startswith(("#", "Parameters:")): + continue + try: + value_str = raw[col["Conversion Value"]].strip() + value = float(value_str) if value_str else 0.0 + except (ValueError, IndexError): + errors.append(f"Row {line_num}: invalid Conversion Value") + continue + out.append({ + "caller_id": _normalize_phone_e164( + raw[col["Caller's Phone Number"]] + ), + "call_start_time": _normalize_call_timestamp( + raw[col["Call Start Time"]] + ), + "conversion_name": raw[col["Conversion Name"]].strip(), + "conversion_time": _normalize_call_timestamp( + raw[col["Conversion Time"]] + ), + "conversion_value": value, + "currency_code": ( + raw[col["Conversion Currency"]].strip().upper() or "USD" + ), + }) + return out, errors + + +def _redact_caller_id(caller_id: str) -> str: + """Mask an E.164 phone for display/logging: keep the leading digits and + the last 4, star the middle. e.g. '+15555550142' -> '+155***0142'. + """ + s = (caller_id or "").strip() + if not s: + return "" + if len(s) <= 6: + return "***" + head = s[:4] + tail = s[-4:] + return f"{head}***{tail}" + + +def draft_upload_call_conversions( + config: AdLoopConfig, + *, + customer_id: str = "", + csv_path: str, + partial_failure: bool = True, + consent: dict | None = None, +) -> dict: + """Draft an upload of call conversions from CSV — returns a PREVIEW. + + Reads any CSV matching Google Ads' call-upload schema and previews what + would be sent to ConversionUploadService.UploadCallConversions. + + Required CSV columns: Caller's Phone Number, Call Start Time, Conversion + Name, Conversion Time, Conversion Value, Conversion Currency. An optional + ``Parameters:TimeZone=...`` row at the top is ignored. + + The ``Conversion Name`` value MUST exactly match an existing conversion + action whose type is UPLOAD_CALLS. + + ``partial_failure`` (default True) lets Google accept the rows that parse + successfully and report only the bad ones — recommended. + + ``consent`` (GDPR/EEA): a dict like + ``{"ad_user_data": "GRANTED", "ad_personalization": "DENIED"}``. Values: + GRANTED / DENIED / UNSPECIFIED. Required for EEA traffic. Defaults to + UNSPECIFIED when omitted. + + PII note: the caller phone number is required raw by Google for matching, + so it is stored in the plan (needed by apply) but REDACTED in the preview + and the audit log. Call confirm_and_apply with the returned plan_id. + """ + from adloop.safety.guards import SafetyViolation, check_blocked_operation + from adloop.safety.preview import ChangePlan, store_plan + + try: + check_blocked_operation("upload_call_conversions", config.safety) + except SafetyViolation as e: + return {"error": str(e)} + + try: + consent_norm = _consent_from_param(consent) + except ValueError as e: + return {"error": str(e)} + + rows, parse_errors = _parse_call_conversion_csv(csv_path) + if parse_errors and not rows: + return {"error": "CSV parse failed", "details": parse_errors} + if not rows: + return {"error": "CSV contained zero conversion rows"} + + distinct_actions = sorted({r["conversion_name"] for r in rows}) + total_value = sum(r["conversion_value"] for r in rows) + + # Freeze the exact rows apply will upload (caller_id RAW — required by + # Google). These live in the plan; the audit path redacts caller_id. + frozen_rows = [ + { + "caller_id": r["caller_id"], + "call_start_time": r["call_start_time"], + "conversion_name": r["conversion_name"], + "conversion_time": r["conversion_time"], + "conversion_value": r["conversion_value"], + "currency_code": r["currency_code"], + } + for r in rows + ] + + plan = ChangePlan( + operation="upload_call_conversions", + entity_type="call_conversion_batch", + entity_id=str(len(rows)), + customer_id=customer_id, + changes={ + "row_count": len(rows), + "total_value": round(total_value, 2), + "currency_hint": rows[0]["currency_code"] if rows else "USD", + "distinct_conversion_actions": distinct_actions, + "partial_failure": bool(partial_failure), + "consent": consent_norm, + "parse_warnings": parse_errors, + # RAW caller_id lives here (apply needs it). REDACTED for audit. + "rows": frozen_rows, + # Display sample uses REDACTED caller ids only. + "sample_rows": [ + { + "caller_id": _redact_caller_id(r["caller_id"]), + "call_start_time": r["call_start_time"], + "conversion_name": r["conversion_name"], + "conversion_value": r["conversion_value"], + } + for r in rows[:3] + ], + }, + ) + store_plan(plan) + return plan.to_preview() + + +def _resolve_conversion_action_ids( + client: object, cid: str, names: list[str] +) -> dict[str, str]: + """Look up conversion_action.resource_name for a list of names. + + Raises ValueError if any name isn't found OR isn't of type UPLOAD_CALLS. + """ + if not names: + return {} + + ga_service = client.get_service("GoogleAdsService") + quoted = ", ".join(f"'{_gaql_escape(n)}'" for n in names) + query = ( + "SELECT conversion_action.id, conversion_action.name, " + "conversion_action.resource_name, conversion_action.type, " + "conversion_action.status " + "FROM conversion_action " + f"WHERE conversion_action.name IN ({quoted}) " + "AND conversion_action.status != 'REMOVED'" + ) + response = ga_service.search(customer_id=cid, query=query) + + mapping: dict[str, str] = {} + bad_types: list[str] = [] + for row in response: + ca = row.conversion_action + ca_type = ca.type_.name if hasattr(ca.type_, "name") else str(ca.type_) + if ca_type != "UPLOAD_CALLS": + bad_types.append(f"{ca.name} (type={ca_type})") + continue + mapping[ca.name] = ca.resource_name + + missing = [n for n in names if n not in mapping] + if bad_types: + raise ValueError( + "Some conversion actions exist but are not of type UPLOAD_CALLS " + f"(needed for call uploads): {bad_types}. " + "Create a new conversion action via " + "draft_create_conversion_action(type_='UPLOAD_CALLS', ...) " + "or via Google Ads UI: Tools → Conversions → New → " + "Import → Other data sources → Track conversions from calls." + ) + if missing: + raise ValueError( + f"Conversion action(s) not found: {missing}. " + "Verify the 'Conversion Name' column in the CSV matches exactly." + ) + return mapping + + +def _apply_upload_call_conversions( + client: object, cid: str, changes: dict +) -> dict: + """Execute the call-conversion upload via ConversionUploadService. + + Builds the upload protos from ``changes["rows"]`` (frozen at preview + time) — the CSV is NOT re-read. Returns counts of successes / failures + plus per-row error details. + """ + rows = changes.get("rows") or [] + if not rows: + return {"error": "Plan contained zero call-conversion rows"} + + distinct = sorted({r["conversion_name"] for r in rows}) + action_resources = _resolve_conversion_action_ids(client, cid, distinct) + consent = changes.get("consent") + + payload: list = [] + for r in rows: + cc = client.get_type("CallConversion") + cc.caller_id = r["caller_id"] + cc.call_start_date_time = r["call_start_time"] + cc.conversion_action = action_resources[r["conversion_name"]] + cc.conversion_date_time = r["conversion_time"] + cc.conversion_value = float(r["conversion_value"]) + cc.currency_code = r["currency_code"] + _apply_consent(client, cc, consent) + payload.append(cc) + + upload_service = client.get_service("ConversionUploadService") + response = upload_service.upload_call_conversions( + customer_id=cid, + conversions=payload, + partial_failure=bool(changes.get("partial_failure", True)), + ) + + results = list(response.results) + # Google only populates the result row's ``conversion_action`` for rows + # that were actually accepted. Echoed identifiers (caller_id) come back + # even for FAILED rows, so counting those overreports success — key off + # conversion_action being populated instead. + success_count = sum( + 1 for r in results if getattr(r, "conversion_action", "") + ) + failure_count = len(results) - success_count + + row_errors: list[dict] = [] + partial = getattr(response, "partial_failure_error", None) + if partial and partial.message: + row_errors.append({ + "type": "partial_failure", + "message": partial.message, + "code": getattr(partial, "code", None), + }) + + return { + "uploaded_total": len(payload), + "success_count": success_count, + "failure_count": failure_count, + "conversion_actions_used": action_resources, + "row_errors": row_errors, + } + + +# --------------------------------------------------------------------------- +# Enhanced Conversions for Leads — UploadClickConversions w/ user_identifiers +# +# The CSV here carries RAW PII (email / phone / name). We normalize and +# SHA-256-hash it AT PREVIEW TIME and store only the hashes in the plan. +# Raw PII is never persisted and never reaches the audit log. +# --------------------------------------------------------------------------- + +_EXPECTED_EC_HEADERS = [ + "Email", + "Phone Number", + "First Name", + "Last Name", + "Conversion Name", + "Conversion Time", + "Conversion Value", + "Conversion Currency", +] + +# Optional CSV columns — parsed when present, omitted otherwise. Order ID is +# Google's dedup key for ClickConversion uploads: if set, re-uploading the +# same (conversion_action, order_id) pair is idempotent; if absent, re-uploads +# double-count. +_OPTIONAL_EC_HEADERS = [ + "Order ID", +] + + +def _parse_ec_for_leads_csv(csv_path: str) -> tuple[list[dict], list[str]]: + """Parse the EC-for-Leads CSV and hash PII at parse time. + + Required columns: Email, Phone Number, First Name, Last Name, + Conversion Name, Conversion Time, Conversion Value, Conversion Currency. + Optional: Order ID (Google's dedup key — strongly recommended so + re-uploads of the same source row don't double-count). + + The Email / Phone Number / First Name / Last Name columns hold RAW PII. + Each is normalized (email→trim+lowercase, phone→E.164, names→trim+ + lowercase) and then SHA-256-hashed here. Returned rows contain ONLY the + hashes (``*_sha256`` keys) plus non-PII fields — the raw values never + leave this function. + """ + import csv + from pathlib import Path + + errors: list[str] = [] + path = Path(csv_path).expanduser() + if not path.exists(): + return [], [f"CSV not found at {path}"] + + with path.open("r", newline="") as f: + reader = csv.reader(f) + rows_iter = iter(reader) + header: list[str] | None = None + for raw in rows_iter: + if not raw: + continue + first = (raw[0] or "").strip() + if first.startswith(("Parameters:", "#")): + continue + header = [c.strip() for c in raw] + break + if header is None: + return [], ["CSV is empty"] + + missing = [c for c in _EXPECTED_EC_HEADERS if c not in header] + if missing: + return [], [ + f"CSV missing required columns: {missing}. Got: {header}" + ] + col = {n: header.index(n) for n in _EXPECTED_EC_HEADERS} + optional_col = { + n: header.index(n) for n in _OPTIONAL_EC_HEADERS if n in header + } + out: list[dict] = [] + for line_num, raw in enumerate(rows_iter, start=2): + if not raw or all((c or "").strip() == "" for c in raw): + continue + try: + value_str = raw[col["Conversion Value"]].strip() + value = float(value_str) if value_str else 0.0 + except (ValueError, IndexError): + errors.append(f"Row {line_num}: invalid Conversion Value") + continue + order_id = "" + if "Order ID" in optional_col: + try: + order_id = raw[optional_col["Order ID"]].strip() + except IndexError: + pass + # Normalize THEN hash. Raw values are discarded immediately. + email_norm = _normalize_email(raw[col["Email"]]) + phone_norm = _normalize_phone_e164(raw[col["Phone Number"]]) + first_norm = _normalize_name(raw[col["First Name"]]) + last_norm = _normalize_name(raw[col["Last Name"]]) + out.append({ + "email_sha256": _sha256_hex(email_norm), + "phone_sha256": _sha256_hex(phone_norm), + "first_name_sha256": _sha256_hex(first_norm), + "last_name_sha256": _sha256_hex(last_norm), + "conversion_name": raw[col["Conversion Name"]].strip(), + "conversion_time": _normalize_call_timestamp( + raw[col["Conversion Time"]] + ), + "conversion_value": value, + "currency_code": ( + raw[col["Conversion Currency"]].strip().upper() or "USD" + ), + "order_id": order_id, + }) + return out, errors + + +def draft_upload_enhanced_conversions_for_leads( + config: AdLoopConfig, + *, + customer_id: str = "", + csv_path: str, + partial_failure: bool = True, + consent: dict | None = None, +) -> dict: + """Draft an Enhanced Conversions for Leads upload — returns PREVIEW. + + Reads a CSV of RAW lead PII, normalizes + SHA-256-hashes the + Email / Phone / First Name / Last Name columns, and previews what will + be pushed via ConversionUploadService.UploadClickConversions with + user_identifiers populated. Only the hashes are stored in the plan — + raw PII never lands in plan.changes or the audit log. + + The target conversion action must be of type UPLOAD_CLICKS (EC for Leads + layers user-identifier matching on top of click conversions). Works + retroactively — no "action must exist before the call" constraint like + UPLOAD_CALLS has. + + Optional ``Order ID`` CSV column → ClickConversion.order_id, Google's + dedup key; without it, re-uploads double-count matched conversions. + + ``consent`` (GDPR/EEA): a dict like + ``{"ad_user_data": "GRANTED", "ad_personalization": "DENIED"}``. Values: + GRANTED / DENIED / UNSPECIFIED. Required for EEA traffic. Defaults to + UNSPECIFIED when omitted. + + 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( + "upload_enhanced_conversions_for_leads", config.safety + ) + except SafetyViolation as e: + return {"error": str(e)} + + try: + consent_norm = _consent_from_param(consent) + except ValueError as e: + return {"error": str(e)} + + rows, parse_errors = _parse_ec_for_leads_csv(csv_path) + if parse_errors and not rows: + return {"error": "CSV parse failed", "details": parse_errors} + if not rows: + return {"error": "CSV contained zero conversion rows"} + + distinct_actions = sorted({r["conversion_name"] for r in rows}) + total_value = sum(r["conversion_value"] for r in rows) + with_email = sum(1 for r in rows if r["email_sha256"]) + with_phone = sum(1 for r in rows if r["phone_sha256"]) + with_order_id = sum(1 for r in rows if r.get("order_id")) + + dedup_warnings: list[str] = [] + if with_order_id == 0: + dedup_warnings.append( + "No Order ID column present. Re-uploading this CSV will " + "double-count any matched conversions because Google has no " + "dedup key. Add an Order ID column (e.g. a stable source row " + "identifier) so re-uploads are idempotent." + ) + elif with_order_id < len(rows): + dedup_warnings.append( + f"Only {with_order_id} of {len(rows)} rows have an Order ID. " + "Rows without one will double-count on re-upload." + ) + + # Freeze the hashed rows apply will upload. These are already SHA-256 + # hashes — NO raw PII. Safe to persist in the plan and (order_id/value/ + # currency/time/action only) surface in the audit log. + frozen_rows = [ + { + "email_sha256": r["email_sha256"], + "phone_sha256": r["phone_sha256"], + "first_name_sha256": r["first_name_sha256"], + "last_name_sha256": r["last_name_sha256"], + "conversion_name": r["conversion_name"], + "conversion_time": r["conversion_time"], + "conversion_value": r["conversion_value"], + "currency_code": r["currency_code"], + "order_id": r.get("order_id", ""), + } + for r in rows + ] + + plan = ChangePlan( + operation="upload_enhanced_conversions_for_leads", + entity_type="ec_for_leads_batch", + entity_id=str(len(rows)), + customer_id=customer_id, + changes={ + "row_count": len(rows), + "total_value": round(total_value, 2), + "currency_hint": rows[0]["currency_code"] if rows else "USD", + "rows_with_email": with_email, + "rows_with_phone": with_phone, + "rows_with_order_id": with_order_id, + "distinct_conversion_actions": distinct_actions, + "partial_failure": bool(partial_failure), + "consent": consent_norm, + "parse_warnings": parse_errors, + "dedup_warnings": dedup_warnings, + # Hashed-only rows — no raw PII. Apply builds protos from here. + "rows": frozen_rows, + "sample_rows": [ + { + "email_sha256": (r["email_sha256"][:16] + "...") + if r["email_sha256"] else "", + "phone_sha256": (r["phone_sha256"][:16] + "...") + if r["phone_sha256"] else "", + "conversion_name": r["conversion_name"], + "conversion_value": r["conversion_value"], + "conversion_time": r["conversion_time"], + "order_id": r.get("order_id", ""), + } + for r in rows[:3] + ], + }, + ) + store_plan(plan) + return plan.to_preview() + + +def _resolve_upload_clicks_action( + client: object, cid: str, names: list[str] +) -> dict[str, str]: + """Look up conversion_action.resource_name for UPLOAD_CLICKS actions.""" + if not names: + return {} + + ga_service = client.get_service("GoogleAdsService") + quoted = ", ".join(f"'{_gaql_escape(n)}'" for n in names) + query = ( + "SELECT conversion_action.id, conversion_action.name, " + "conversion_action.resource_name, conversion_action.type, " + "conversion_action.status " + "FROM conversion_action " + f"WHERE conversion_action.name IN ({quoted}) " + "AND conversion_action.status != 'REMOVED'" + ) + response = ga_service.search(customer_id=cid, query=query) + + mapping: dict[str, str] = {} + wrong_type: list[str] = [] + for row in response: + ca = row.conversion_action + ca_type = ca.type_.name if hasattr(ca.type_, "name") else str(ca.type_) + if ca_type != "UPLOAD_CLICKS": + wrong_type.append(f"{ca.name} (type={ca_type})") + continue + mapping[ca.name] = ca.resource_name + + missing = [n for n in names if n not in mapping] + if wrong_type: + raise ValueError( + "Some conversion actions are not of type UPLOAD_CLICKS " + "(required for Enhanced Conversions for Leads uploads): " + f"{wrong_type}. Use an UPLOAD_CLICKS-type action — create one " + "via draft_create_conversion_action(type_='UPLOAD_CLICKS', ...)." + ) + if missing: + raise ValueError( + f"Conversion action(s) not found: {missing}. " + "Verify the 'Conversion Name' column in the CSV." + ) + return mapping + + +def _apply_upload_enhanced_conversions_for_leads( + client: object, cid: str, changes: dict +) -> dict: + """Execute EC-for-Leads upload via ConversionUploadService. + + Builds the upload protos from ``changes["rows"]`` (SHA-256 hashes frozen + at preview time) — the CSV is NOT re-read, so no raw PII is touched here. + """ + rows = changes.get("rows") or [] + if not rows: + return {"error": "Plan contained zero EC-for-leads rows"} + + distinct = sorted({r["conversion_name"] for r in rows}) + action_resources = _resolve_upload_clicks_action(client, cid, distinct) + consent = changes.get("consent") + + payload: list = [] + for r in rows: + cc = client.get_type("ClickConversion") + cc.conversion_action = action_resources[r["conversion_name"]] + cc.conversion_date_time = r["conversion_time"] + cc.conversion_value = float(r["conversion_value"]) + cc.currency_code = r["currency_code"] + + # Order ID is Google's dedup key for ClickConversion. When set, + # re-uploading the same (conversion_action, order_id) is a no-op; + # without it, every upload counts as a fresh conversion. + if r.get("order_id"): + cc.order_id = r["order_id"] + + _apply_consent(client, cc, consent) + + # Build user_identifiers from the hashed PII. Google matches the + # hashed email/phone/name to logged-in users who clicked our ads. + if r["email_sha256"]: + uid = client.get_type("UserIdentifier") + uid.hashed_email = r["email_sha256"] + cc.user_identifiers.append(uid) + if r["phone_sha256"]: + uid = client.get_type("UserIdentifier") + uid.hashed_phone_number = r["phone_sha256"] + cc.user_identifiers.append(uid) + if r["first_name_sha256"] and r["last_name_sha256"]: + uid = client.get_type("UserIdentifier") + uid.address_info.hashed_first_name = r["first_name_sha256"] + uid.address_info.hashed_last_name = r["last_name_sha256"] + cc.user_identifiers.append(uid) + + if not cc.user_identifiers: + continue + payload.append(cc) + + upload_service = client.get_service("ConversionUploadService") + response = upload_service.upload_click_conversions( + customer_id=cid, + conversions=payload, + partial_failure=bool(changes.get("partial_failure", True)), + ) + + results = list(response.results) + # The Google Ads API only populates the result row's ``conversion_action`` + # for rows that actually matched and were accepted. Failed rows come back + # with empty conversion_action (and a corresponding partial_failure_error + # entry). Echoed user_identifiers are NOT a success signal — the API + # echoes them back for failed rows too — so key success off + # conversion_action being populated. + success_count = sum( + 1 for r in results if getattr(r, "conversion_action", "") + ) + failure_count = len(results) - success_count + + row_errors: list[dict] = [] + partial = getattr(response, "partial_failure_error", None) + if partial and partial.message: + row_errors.append({ + "type": "partial_failure", + "message": partial.message, + "code": getattr(partial, "code", None), + }) + + return { + "uploaded_total": len(payload), + "success_count": success_count, + "failure_count": failure_count, + "conversion_actions_used": action_resources, + "row_errors": row_errors, + } diff --git a/src/adloop/ads/write.py b/src/adloop/ads/write.py index ed25b96..2dcd3ed 100644 --- a/src/adloop/ads/write.py +++ b/src/adloop/ads/write.py @@ -1777,6 +1777,55 @@ def _extract_error_message(exc: Exception) -> str: return fallback if fallback else repr(exc) +def _redact_changes_for_audit(operation: str, changes: dict) -> dict: + """Return an audit-safe copy of a plan's ``changes`` dict. + + Some upload operations carry data that must never hit the audit log + verbatim: + + * ``upload_call_conversions`` — each frozen row holds a raw E.164 + ``caller_id``. Google requires it raw for call-to-click matching (it + cannot be hashed), so it lives in the plan for apply — but it is PII and + must be REDACTED here (e.g. ``+155***0142``). Sample rows already carry + redacted ids; we redact the ``rows`` list too. + * ``upload_enhanced_conversions_for_leads`` — rows already contain only + SHA-256 hashes (no raw PII), so they are safe. We still drop the bulky + per-row hash blob from the audit record to keep it compact and to avoid + logging identifier hashes at row granularity; the non-PII summary + counters (row_count, total_value, rows_with_email/phone/order_id) remain. + + Non-upload operations are returned unchanged. + """ + if operation == "upload_call_conversions": + from adloop.ads.conversion_actions import _redact_caller_id + + redacted = dict(changes) + rows = redacted.get("rows") + if isinstance(rows, list): + redacted["rows"] = [ + {**{k: v for k, v in row.items() if k != "caller_id"}, + "caller_id": _redact_caller_id(row.get("caller_id", ""))} + for row in rows + ] + return redacted + + if operation == "upload_enhanced_conversions_for_leads": + redacted = dict(changes) + # Rows are hash-only, but drop the row-level hash blob from the audit + # trail — the summary counters above it are sufficient for an audit. + if "rows" in redacted: + redacted = { + k: v for k, v in redacted.items() if k != "rows" + } + redacted["rows_redacted"] = ( + f"{changes.get('row_count', len(changes.get('rows') or []))} " + "hashed rows omitted from audit log (SHA-256, no raw PII)" + ) + return redacted + + return changes + + def confirm_and_apply( config: AdLoopConfig, *, @@ -1802,6 +1851,8 @@ def confirm_and_apply( if config.safety.require_dry_run: dry_run = True + audit_changes = _redact_changes_for_audit(plan.operation, plan.changes) + if dry_run: log_mutation( config.safety.log_file, @@ -1809,7 +1860,7 @@ def confirm_and_apply( customer_id=plan.customer_id, entity_type=plan.entity_type, entity_id=plan.entity_id, - changes=plan.changes, + changes=audit_changes, dry_run=True, result="dry_run_success", ) @@ -1817,7 +1868,7 @@ def confirm_and_apply( "status": "DRY_RUN_SUCCESS", "plan_id": plan.plan_id, "operation": plan.operation, - "changes": plan.changes, + "changes": audit_changes, } if forced_by_config: # The caller passed dry_run=false but safety.require_dry_run @@ -1858,7 +1909,7 @@ def confirm_and_apply( customer_id=plan.customer_id, entity_type=plan.entity_type, entity_id=plan.entity_id, - changes=plan.changes, + changes=audit_changes, dry_run=False, result="error", error=error_message, @@ -1871,7 +1922,7 @@ def confirm_and_apply( customer_id=plan.customer_id, entity_type=plan.entity_type, entity_id=plan.entity_id, - changes=plan.changes, + changes=audit_changes, dry_run=False, result="success", ) @@ -2533,6 +2584,16 @@ def _execute_plan(config: AdLoopConfig, plan: object) -> dict: client = get_ads_client(config) cid = normalize_customer_id(plan.customer_id) + # Conversion-action CRUD lives in its own module; import lazily so the + # dispatch table (and this module) don't take the dependency at import time. + from adloop.ads.conversion_actions import ( + _apply_create_conversion_action, + _apply_remove_conversion_action, + _apply_update_conversion_action, + _apply_upload_call_conversions, + _apply_upload_enhanced_conversions_for_leads, + ) + dispatch = { "create_campaign": _apply_create_campaign, "create_ad_group": _apply_create_ad_group, @@ -2554,6 +2615,13 @@ 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_conversion_action": _apply_create_conversion_action, + "update_conversion_action": _apply_update_conversion_action, + "remove_conversion_action": _apply_remove_conversion_action, + "upload_call_conversions": _apply_upload_call_conversions, + "upload_enhanced_conversions_for_leads": ( + _apply_upload_enhanced_conversions_for_leads + ), } handler = dispatch.get(plan.operation) diff --git a/src/adloop/server.py b/src/adloop/server.py index 5ebe0ff..6f296ce 100644 --- a/src/adloop/server.py +++ b/src/adloop/server.py @@ -2070,6 +2070,207 @@ def draft_key_event( ) +@mcp.tool(annotations=_WRITE) +@_safe +def draft_create_conversion_action( + name: str, + type_: str, + category: str = "DEFAULT", + default_value: float = 0, + currency_code: str = "USD", + always_use_default_value: bool = False, + counting_type: str = "ONE_PER_CLICK", + phone_call_duration_seconds: int = 0, + primary_for_goal: bool = True, + include_in_conversions_metric: bool = True, + click_through_window_days: int = 0, + view_through_window_days: int = 0, + attribution_model: str = "", + customer_id: str = "", +) -> dict: + """Draft a new Google Ads ConversionAction — returns a PREVIEW. + + type_: AD_CALL, WEBSITE_CALL, WEBPAGE, WEBPAGE_CODELESS, + GOOGLE_ANALYTICS_4_CUSTOM, etc. category: PHONE_CALL_LEAD, + SUBMIT_LEAD_FORM, PURCHASE, ... (default DEFAULT). + + A positive default_value with always_use_default_value=False is a legal + "fallback" config — the preview warns but does NOT flip the flag. + include_in_conversions_metric is IMMUTABLE on create (change it later via + draft_update_conversion_action). Call confirm_and_apply with the returned + plan_id to execute. + """ + from adloop.ads.conversion_actions import ( + draft_create_conversion_action as _impl, + ) + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + name=name, + type_=type_, + category=category, + default_value=default_value, + currency_code=currency_code, + always_use_default_value=always_use_default_value, + counting_type=counting_type, + phone_call_duration_seconds=phone_call_duration_seconds, + primary_for_goal=primary_for_goal, + include_in_conversions_metric=include_in_conversions_metric, + click_through_window_days=click_through_window_days, + view_through_window_days=view_through_window_days, + attribution_model=attribution_model, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_update_conversion_action( + conversion_action_id: str, + name: str = "", + primary_for_goal: bool | None = None, + default_value: float = 0, + currency_code: str = "", + always_use_default_value: bool | None = None, + counting_type: str = "", + phone_call_duration_seconds: int = 0, + include_in_conversions_metric: bool | None = None, + click_through_window_days: int = 0, + view_through_window_days: int = 0, + attribution_model: str = "", + customer_id: str = "", +) -> dict: + """Draft a partial UPDATE of an existing ConversionAction — returns PREVIEW. + + Only the parameters you pass non-empty/non-default are sent to the API. + Use to rename, demote a Primary to Secondary, change value, adjust the + call-duration threshold, or change attribution. Find the ID via: + SELECT conversion_action.id, conversion_action.name FROM conversion_action. + Call confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.conversion_actions import ( + draft_update_conversion_action as _impl, + ) + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + conversion_action_id=conversion_action_id, + name=name, + primary_for_goal=primary_for_goal, + default_value=default_value, + currency_code=currency_code, + always_use_default_value=always_use_default_value, + counting_type=counting_type, + phone_call_duration_seconds=phone_call_duration_seconds, + include_in_conversions_metric=include_in_conversions_metric, + click_through_window_days=click_through_window_days, + view_through_window_days=view_through_window_days, + attribution_model=attribution_model, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_remove_conversion_action( + conversion_action_id: str, + customer_id: str = "", +) -> dict: + """Draft a REMOVAL of a ConversionAction — returns a PREVIEW. + + Removal stops counting and drops the action from goal lists; historical + data is preserved but the action is irreversible. SMART_CAMPAIGN_* and + GOOGLE_HOSTED types reject removal with MUTATE_NOT_ALLOWED. Call + confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.conversion_actions import ( + draft_remove_conversion_action as _impl, + ) + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + conversion_action_id=conversion_action_id, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_upload_call_conversions( + csv_path: str, + partial_failure: bool = True, + consent: dict | None = None, + customer_id: str = "", +) -> dict: + """Draft an offline CALL-conversion upload from a CSV — returns a PREVIEW. + + Uploads phone-call conversions to ConversionUploadService. + UploadCallConversions, matching each call back to the ad click by the + caller's phone number. The CSV must have columns: Caller's Phone Number, + Call Start Time, Conversion Name, Conversion Time, Conversion Value, + Conversion Currency. The Conversion Name must match an existing + UPLOAD_CALLS-type conversion action exactly. + + PII: the caller phone number is required RAW by Google (it cannot be + hashed) — it is redacted in the preview and audit log but stored in the + plan so apply uploads exactly what you previewed (no CSV re-read). + + consent (GDPR/EEA): {"ad_user_data": "GRANTED"|"DENIED"|"UNSPECIFIED", + "ad_personalization": ...}. Defaults to UNSPECIFIED. Call + confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.conversion_actions import ( + draft_upload_call_conversions as _impl, + ) + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + csv_path=csv_path, + partial_failure=partial_failure, + consent=consent, + ) + + +@mcp.tool(annotations=_WRITE) +@_safe +def draft_upload_enhanced_conversions_for_leads( + csv_path: str, + partial_failure: bool = True, + consent: dict | None = None, + customer_id: str = "", +) -> dict: + """Draft an Enhanced Conversions for LEADS upload from a CSV — PREVIEW. + + Uploads lead conversions to ConversionUploadService.UploadClickConversions + with user_identifiers, matching hashed customer PII back to the Google + users who clicked the ads. The CSV holds RAW PII in columns: Email, Phone + Number, First Name, Last Name (plus Conversion Name, Conversion Time, + Conversion Value, Conversion Currency; optional Order ID dedup key). + + PII is normalized and SHA-256-hashed AT PREVIEW TIME — only the hashes are + stored in the plan. Raw email/phone/name never land in the plan or the + audit log. The target conversion action must be UPLOAD_CLICKS-type. + + Add an Order ID column so re-uploads dedup instead of double-counting. + + consent (GDPR/EEA): {"ad_user_data": "GRANTED"|"DENIED"|"UNSPECIFIED", + "ad_personalization": ...}. Defaults to UNSPECIFIED. Call + confirm_and_apply with the returned plan_id to execute. + """ + from adloop.ads.conversion_actions import ( + draft_upload_enhanced_conversions_for_leads as _impl, + ) + + return _impl( + current_config(), + customer_id=customer_id or current_config().ads.customer_id, + csv_path=csv_path, + partial_failure=partial_failure, + consent=consent, + ) + + @mcp.tool(annotations=_READONLY) @_safe def validate_tracking( diff --git a/tests/test_conversion_actions.py b/tests/test_conversion_actions.py new file mode 100644 index 0000000..000cef1 --- /dev/null +++ b/tests/test_conversion_actions.py @@ -0,0 +1,1544 @@ +"""Tests for conversion-action write tools (create / update / remove).""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from google.ads.googleads.client import GoogleAdsClient + +from adloop.ads import conversion_actions, 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 +# --------------------------------------------------------------------------- + + +class _FakeResult: + def __init__(self, resource_name: str = ""): + self.resource_name = resource_name + + +class _FakeConversionActionService: + def __init__(self, results: list[_FakeResult] | None = None): + self.operations: list = [] + self.results_to_return = results or [] + + def conversion_action_path(self, customer_id: str, ca_id: str) -> str: + return f"customers/{customer_id}/conversionActions/{ca_id}" + + def mutate_conversion_actions( + self, customer_id: str, operations: list + ) -> object: + self.operations = operations + return SimpleNamespace(results=self.results_to_return) + + +class _FakeClient: + def __init__(self, services: dict[str, object] | None = None): + 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 or {} + + def get_service(self, name: str) -> object: + return self._services[name] + + +@pytest.fixture(autouse=True) +def clear_pending_plans(): + # v0.12 runtime: plans live in a swappable store, scoped per tenant. + 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 _stored_plan(result: dict): + """Fetch the stored ChangePlan for a draft result (current tenant).""" + plan = preview_store.get_plan(result["plan_id"]) + assert plan is not None, "plan was not stored" + return plan + + +# --------------------------------------------------------------------------- +# Validation tests for draft_create_conversion_action +# --------------------------------------------------------------------------- + + +class TestDraftCreateConversionActionValidation: + def _ok_args(self, **overrides): + defaults = dict( + customer_id="1234567890", + name="Calls from Ads", + type_="AD_CALL", + category="PHONE_CALL_LEAD", + default_value=250, + currency_code="USD", + ) + defaults.update(overrides) + return defaults + + def test_happy_path(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args() + ) + assert "error" not in result + plan = _stored_plan(result) + assert plan.changes["name"] == "Calls from Ads" + assert plan.changes["type"] == "AD_CALL" + assert plan.changes["default_value"] == 250.0 + assert plan.changes["currency_code"] == "USD" + assert plan.changes["counting_type"] == "ONE_PER_CLICK" + assert plan.changes["primary_for_goal"] is True + + def test_name_required(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(name="") + ) + assert result["error"] == "Validation failed" + assert any("name is required" in d for d in result["details"]) + + def test_invalid_type(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(type_="MADE_UP_TYPE") + ) + assert result["error"] == "Validation failed" + assert any("MADE_UP_TYPE" in d for d in result["details"]) + + def test_invalid_category(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(category="WRONG_CATEGORY") + ) + assert result["error"] == "Validation failed" + assert any("WRONG_CATEGORY" in d for d in result["details"]) + + def test_invalid_counting_type(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(counting_type="WRONG") + ) + assert result["error"] == "Validation failed" + assert any("counting_type" in d for d in result["details"]) + + def test_negative_default_value_rejected(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(default_value=-1) + ) + assert result["error"] == "Validation failed" + assert any("default_value" in d for d in result["details"]) + + def test_invalid_currency_length(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(currency_code="USDX") + ) + assert result["error"] == "Validation failed" + assert any("currency_code" in d for d in result["details"]) + + def test_invalid_click_through_window(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(click_through_window_days=120) + ) + assert result["error"] == "Validation failed" + assert any("click_through_window_days" in d for d in result["details"]) + + def test_invalid_view_through_window(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(view_through_window_days=60) + ) + assert result["error"] == "Validation failed" + assert any("view_through_window_days" in d for d in result["details"]) + + def test_invalid_attribution_model(self, config): + result = conversion_actions.draft_create_conversion_action( + config, **self._ok_args(attribution_model="MAGIC") + ) + assert result["error"] == "Validation failed" + assert any("attribution_model" in d for d in result["details"]) + + def test_phone_call_duration_threshold_persisted(self, config): + result = conversion_actions.draft_create_conversion_action( + config, + **self._ok_args( + type_="WEBSITE_CALL", + phone_call_duration_seconds=90, + ), + ) + plan = _stored_plan(result) + assert plan.changes["phone_call_duration_seconds"] == 90 + + def test_default_value_with_fallback_flag_warns_not_flips(self, config): + """Maintainer fix #2: a positive default_value paired with + always_use_default_value=False is a LEGAL "tag value with fallback" + config. The draft must emit a PREVIEW WARNING and leave the flag + exactly as the caller set it — NOT silently force it to True (which + would turn a fallback into an unconditional override).""" + result = conversion_actions.draft_create_conversion_action( + config, + **self._ok_args(default_value=400, always_use_default_value=False), + ) + assert "error" not in result + # The flag is NOT force-set. + plan = _stored_plan(result) + assert plan.changes["default_value"] == 400.0 + assert plan.changes["always_use_default_value"] is False + # A warning surfaces the fallback-vs-override behavior. + assert "warnings" in result + assert any( + "fallback" in w.lower() and "always_use_default_value" in w + for w in result["warnings"] + ) + + def test_zero_default_value_no_warning(self, config): + """No warning when default_value is 0 — callers may legitimately + want to use snippet/import-provided values.""" + result = conversion_actions.draft_create_conversion_action( + config, + **self._ok_args(default_value=0, always_use_default_value=False), + ) + assert "error" not in result + plan = _stored_plan(result) + assert plan.changes["always_use_default_value"] is False + assert "warnings" not in result + + def test_explicit_always_use_default_value_true_no_warning(self, config): + """Explicit True is preserved and produces no fallback warning.""" + result = conversion_actions.draft_create_conversion_action( + config, + **self._ok_args(default_value=500, always_use_default_value=True), + ) + assert "error" not in result + plan = _stored_plan(result) + assert plan.changes["always_use_default_value"] is True + assert "warnings" not in result + + +# --------------------------------------------------------------------------- +# draft_update_conversion_action +# --------------------------------------------------------------------------- + + +class TestDraftUpdateConversionAction: + def test_id_required(self, config): + result = conversion_actions.draft_update_conversion_action( + config, customer_id="1", conversion_action_id="" + ) + assert "conversion_action_id is required" in result["error"] + + def test_no_fields_to_update_rejected(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + ) + assert "No fields to update" in result["error"] + + def test_partial_update_only_includes_specified(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + name="Calls from Ads (>=90s)", + primary_for_goal=False, + default_value=250, + currency_code="USD", + ) + plan = _stored_plan(result) + # specified fields present + assert plan.changes["name"] == "Calls from Ads (>=90s)" + assert plan.changes["primary_for_goal"] is False + assert plan.changes["default_value"] == 250.0 + assert plan.changes["currency_code"] == "USD" + # unspecified fields absent + assert "counting_type" not in plan.changes + assert "click_through_window_days" not in plan.changes + + def test_promote_to_primary(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + primary_for_goal=True, + ) + plan = _stored_plan(result) + assert plan.changes["primary_for_goal"] is True + + def test_demote_to_secondary(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + primary_for_goal=False, + ) + plan = _stored_plan(result) + assert plan.changes["primary_for_goal"] is False + + def test_invalid_counting_type_rejected(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + counting_type="BAD", + ) + assert result["error"] == "Validation failed" + + def test_phone_duration_persisted(self, config): + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + phone_call_duration_seconds=90, + ) + plan = _stored_plan(result) + assert plan.changes["phone_call_duration_seconds"] == 90 + + def test_include_in_conversions_metric_is_mutable_on_update(self, config): + """Unlike create (where it's IMMUTABLE), the update path accepts + include_in_conversions_metric and passes it through.""" + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + include_in_conversions_metric=False, + ) + plan = _stored_plan(result) + assert plan.changes["include_in_conversions_metric"] is False + + def test_fallback_flag_warns_on_update(self, config): + """Fix #2 also applies on update: positive default_value with the + flag explicitly False warns rather than overriding intent.""" + result = conversion_actions.draft_update_conversion_action( + config, + customer_id="1", + conversion_action_id="6797442210", + default_value=300, + always_use_default_value=False, + ) + plan = _stored_plan(result) + assert plan.changes["always_use_default_value"] is False + assert plan.changes["default_value"] == 300.0 + assert "warnings" in result + assert any("fallback" in w.lower() for w in result["warnings"]) + + +# --------------------------------------------------------------------------- +# draft_remove_conversion_action +# --------------------------------------------------------------------------- + + +class TestDraftRemoveConversionAction: + def test_id_required(self, config): + result = conversion_actions.draft_remove_conversion_action( + config, customer_id="1", conversion_action_id="" + ) + assert "conversion_action_id is required" in result["error"] + + def test_emits_irreversible_warning(self, config): + result = conversion_actions.draft_remove_conversion_action( + config, customer_id="1", conversion_action_id="6797442210" + ) + assert "warnings" in result + assert any("irreversible" in w.lower() for w in result["warnings"]) + plan = _stored_plan(result) + assert plan.operation == "remove_conversion_action" + assert plan.entity_id == "6797442210" + + +# --------------------------------------------------------------------------- +# Apply handlers — exercised against fake services +# --------------------------------------------------------------------------- + + +class TestApplyCreateConversionAction: + def test_websitecall_with_duration_threshold(self): + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/100")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_create_conversion_action( + client, + "1", + { + "name": "Website Call (GFN >=90s)", + "type": "WEBSITE_CALL", + "category": "PHONE_CALL_LEAD", + "default_value": 250.0, + "currency_code": "USD", + "always_use_default_value": True, + "counting_type": "ONE_PER_CLICK", + "phone_call_duration_seconds": 90, + "primary_for_goal": True, + "include_in_conversions_metric": True, + "click_through_window_days": 30, + "view_through_window_days": 1, + "attribution_model": "GOOGLE_SEARCH_ATTRIBUTION_DATA_DRIVEN", + }, + ) + + assert len(ca_svc.operations) == 1 + ca = ca_svc.operations[0].create + assert ca.name == "Website Call (GFN >=90s)" + assert ca.type_ == client.enums.ConversionActionTypeEnum.WEBSITE_CALL + assert ca.category == client.enums.ConversionActionCategoryEnum.PHONE_CALL_LEAD + assert ca.value_settings.default_value == 250.0 + assert ca.value_settings.default_currency_code == "USD" + assert ca.value_settings.always_use_default_value is True + assert ca.counting_type == client.enums.ConversionActionCountingTypeEnum.ONE_PER_CLICK + assert ca.primary_for_goal is True + assert ca.phone_call_duration_seconds == 90 + assert ca.click_through_lookback_window_days == 30 + assert ca.view_through_lookback_window_days == 1 + + def test_does_not_set_include_in_conversions_metric_on_create(self): + """Regression: Google's API treats include_in_conversions_metric + as IMMUTABLE on create (derived from category). Setting it in the + create mutate raises IMMUTABLE_FIELD. The apply function must + leave the proto field unset; callers who need to change it must + use draft_update_conversion_action after the create succeeds.""" + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/100")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_create_conversion_action( + client, + "1", + { + "name": "Example Co - Call from Ad", + "type": "AD_CALL", + "category": "PHONE_CALL_LEAD", + "default_value": 400.0, + "currency_code": "USD", + "always_use_default_value": True, + "counting_type": "ONE_PER_CLICK", + "phone_call_duration_seconds": 0, + "primary_for_goal": False, + # Caller passes True (the tool's default), but the apply + # function must NOT propagate it into the proto on create. + "include_in_conversions_metric": True, + "click_through_window_days": 30, + "view_through_window_days": 0, + "attribution_model": "", + }, + ) + + assert len(ca_svc.operations) == 1 + ca = ca_svc.operations[0].create + # The proto3-optional field must not be explicitly set, otherwise + # Google rejects the mutate with IMMUTABLE_FIELD. + assert not ca._pb.HasField("include_in_conversions_metric") + + +class TestApplyUpdateConversionAction: + def test_partial_update_fieldmask(self): + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/6797442210")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_update_conversion_action( + client, + "1", + { + "conversion_action_id": "6797442210", + "name": "Calls from Ads (>=90s)", + "default_value": 250.0, + "currency_code": "USD", + "always_use_default_value": True, + "counting_type": "ONE_PER_CLICK", + "primary_for_goal": True, + }, + ) + + op = ca_svc.operations[0] + ca = op.update + assert ca.resource_name == "customers/1/conversionActions/6797442210" + assert ca.name == "Calls from Ads (>=90s)" + assert ca.value_settings.default_value == 250.0 + assert ca.counting_type == client.enums.ConversionActionCountingTypeEnum.ONE_PER_CLICK + assert ca.primary_for_goal is True + # Field mask reflects exactly the keys we set + mask_paths = list(op.update_mask.paths) + assert "name" in mask_paths + assert "value_settings.default_value" in mask_paths + assert "value_settings.default_currency_code" in mask_paths + assert "value_settings.always_use_default_value" in mask_paths + assert "counting_type" in mask_paths + assert "primary_for_goal" in mask_paths + # Fields we didn't pass shouldn't be in the mask + assert "phone_call_duration_seconds" not in mask_paths + + def test_update_only_phone_duration(self): + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/6797442210")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_update_conversion_action( + client, + "1", + { + "conversion_action_id": "6797442210", + "phone_call_duration_seconds": 90, + }, + ) + + op = ca_svc.operations[0] + ca = op.update + assert ca.phone_call_duration_seconds == 90 + mask_paths = list(op.update_mask.paths) + assert mask_paths == ["phone_call_duration_seconds"] + + def test_update_include_in_conversions_metric_fieldmask(self): + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/6797442210")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_update_conversion_action( + client, + "1", + { + "conversion_action_id": "6797442210", + "include_in_conversions_metric": False, + }, + ) + + op = ca_svc.operations[0] + ca = op.update + assert ca.include_in_conversions_metric is False + assert list(op.update_mask.paths) == ["include_in_conversions_metric"] + + +class TestApplyRemoveConversionAction: + def test_remove_sets_resource_name(self): + ca_svc = _FakeConversionActionService( + [_FakeResult("customers/1/conversionActions/6797442210")] + ) + client = _FakeClient({"ConversionActionService": ca_svc}) + + conversion_actions._apply_remove_conversion_action( + client, + "1", + {"conversion_action_id": "6797442210"}, + ) + + op = ca_svc.operations[0] + assert op.remove == "customers/1/conversionActions/6797442210" + + +# --------------------------------------------------------------------------- +# MCP tool registration + dispatch wiring +# --------------------------------------------------------------------------- + + +class TestMCPRegistration: + @pytest.fixture(scope="class") + @classmethod + def tools_by_name(cls): + import asyncio + from adloop.server import mcp + + async def _list(): + return await mcp.list_tools() + + tools = asyncio.run(_list()) + return {t.name: t for t in tools} + + def test_three_conversion_action_tools_registered(self, tools_by_name): + for name in ( + "draft_create_conversion_action", + "draft_update_conversion_action", + "draft_remove_conversion_action", + ): + assert name in tools_by_name, f"{name} not registered" + + def test_create_required_params(self, tools_by_name): + required = ( + tools_by_name["draft_create_conversion_action"] + .parameters.get("required", []) + ) + assert "name" in required + assert "type_" in required + + def test_update_requires_id(self, tools_by_name): + required = ( + tools_by_name["draft_update_conversion_action"] + .parameters.get("required", []) + ) + assert "conversion_action_id" in required + + def test_remove_requires_id(self, tools_by_name): + required = ( + tools_by_name["draft_remove_conversion_action"] + .parameters.get("required", []) + ) + assert "conversion_action_id" in required + + def test_dispatch_routes(self): + """_execute_plan must map the three CRUD operations to the module's + apply handlers (same dispatch-dict style as the other Ads writes).""" + import inspect + src = inspect.getsource(write._execute_plan) + assert '"create_conversion_action": _apply_create_conversion_action' in src + assert '"update_conversion_action": _apply_update_conversion_action' in src + assert '"remove_conversion_action": _apply_remove_conversion_action' in src + + +# =========================================================================== +# Offline conversion uploads — helpers, hashing, redaction, GAQL escape +# =========================================================================== +# +# Fixtures use only FAKE PII: emails user@example.com, phones +15555550142, +# names Test User, order ids ORD-001. Hash assertions are pinned to the +# SHA-256 hex of those known fakes. + +import hashlib + + +def _sha(s: str) -> str: + return hashlib.sha256(s.encode("utf-8")).hexdigest() + + +# Known-fake canonical hashes (normalize THEN sha256). +_EMAIL_HASH = _sha("user@example.com") +_PHONE_HASH = _sha("+15555550142") +_FIRST_HASH = _sha("test") +_LAST_HASH = _sha("user") + + +class TestSha256Hashing: + def test_email_normalized_then_hashed(self): + # Trim + lowercase, then SHA-256. + h = conversion_actions._sha256_hex( + conversion_actions._normalize_email(" User@Example.COM ") + ) + assert h == _EMAIL_HASH + + def test_phone_normalized_then_hashed(self): + h = conversion_actions._sha256_hex( + conversion_actions._normalize_phone_e164("+1 (555) 555-0142") + ) + assert h == _PHONE_HASH + + def test_name_normalized_then_hashed(self): + assert conversion_actions._sha256_hex( + conversion_actions._normalize_name(" Test ") + ) == _FIRST_HASH + assert conversion_actions._sha256_hex( + conversion_actions._normalize_name("USER") + ) == _LAST_HASH + + def test_empty_hashes_to_empty(self): + assert conversion_actions._sha256_hex("") == "" + + +class TestNormalizePhoneE164: + def test_us_number_preserved(self): + assert ( + conversion_actions._normalize_phone_e164("+1 (555) 555-0142") + == "+15555550142" + ) + + def test_double_zero_international_prefix_becomes_plus(self): + assert ( + conversion_actions._normalize_phone_e164("0044 20 7946 0018") + == "+442079460018" + ) + + def test_strips_exactly_one_domestic_trunk_zero(self): + # UK national format "020 …" -> the single leading 0 is dropped. + assert ( + conversion_actions._normalize_phone_e164("020 7946 0018") + == "2079460018" + ) + + def test_does_not_strip_multiple_leading_zeros_as_trunk(self): + # A leading "00" is the international prefix, NOT two trunk zeros: + # only the "00"->"+" rule fires, no extra zero-stripping. + assert conversion_actions._normalize_phone_e164("0012025550000") == ( + "+12025550000" + ) + + def test_italy_leading_zero_preserved_when_plus_present(self): + # Italian fixed-line numbers KEEP the leading 0 in E.164. + assert ( + conversion_actions._normalize_phone_e164("+39 06 6982 1234") + == "+390669821234" + ) + + def test_empty(self): + assert conversion_actions._normalize_phone_e164("") == "" + assert conversion_actions._normalize_phone_e164(" ") == "" + + +class TestGaqlEscape: + def test_apostrophe_escaped_with_backslash(self): + # GAQL uses backslash escaping, NOT SQL-style doubled quotes. + assert conversion_actions._gaql_escape("O'Brien Lead") == ( + "O\\'Brien Lead" + ) + + def test_backslash_escaped_first(self): + assert conversion_actions._gaql_escape("a\\b") == "a\\\\b" + + def test_used_in_resolve_query(self): + # The resolver must interpolate the escaped name into the WHERE. + ads = _FakeGoogleAdsService([ + _FakeSearchRow( + "O'Brien Lead", "customers/1/conversionActions/1", + type_name="UPLOAD_CALLS", + ) + ]) + conversion_actions._resolve_conversion_action_ids( + _client_with(upload_service=_FakeUploadService(), ads_service=ads), + "1", + ["O'Brien Lead"], + ) + assert "O\\'Brien Lead" in ads.last_query + # Must NOT contain the SQL-style doubled-quote form. + assert "O''Brien" not in ads.last_query + + +class TestRedactCallerId: + def test_masks_middle(self): + assert conversion_actions._redact_caller_id("+15555550142") == ( + "+155***0142" + ) + + def test_short_number_fully_masked(self): + assert conversion_actions._redact_caller_id("12345") == "***" + + def test_empty(self): + assert conversion_actions._redact_caller_id("") == "" + + +class TestConsentParam: + def test_none_returns_none(self): + assert conversion_actions._consent_from_param(None) is None + assert conversion_actions._consent_from_param({}) is None + + def test_defaults_missing_keys_to_unspecified(self): + out = conversion_actions._consent_from_param( + {"ad_user_data": "GRANTED"} + ) + assert out == { + "ad_user_data": "GRANTED", + "ad_personalization": "UNSPECIFIED", + } + + def test_invalid_value_raises(self): + with pytest.raises(ValueError): + conversion_actions._consent_from_param({"ad_user_data": "YES"}) + + +# --------------------------------------------------------------------------- +# Fakes for the upload paths +# --------------------------------------------------------------------------- + + +class _FakeUploadService: + def __init__(self, results_count: int = 0, error_message: str = ""): + self.called_with: dict | None = None + self._results_count = results_count + self._error_message = error_message + + def upload_call_conversions( + self, *, customer_id, conversions, partial_failure + ): + self.called_with = { + "customer_id": customer_id, + "conversions": list(conversions), + "partial_failure": partial_failure, + } + # Mark the first N results as accepted (conversion_action populated). + results = [] + for i, c in enumerate(conversions): + results.append(SimpleNamespace( + conversion_action=( + c.conversion_action if i < self._results_count else "" + ), + caller_id=c.caller_id, # API echoes this back even on failure + )) + partial = ( + SimpleNamespace(message=self._error_message, code=0) + if self._error_message + else SimpleNamespace(message="", code=0) + ) + return SimpleNamespace(results=results, partial_failure_error=partial) + + +class _FakeClickUploadService: + def __init__(self, results_count: int = 0, error_message: str = ""): + self.called_with: dict | None = None + self._results_count = results_count + self._error_message = error_message + + def upload_click_conversions( + self, *, customer_id, conversions, partial_failure + ): + self.called_with = { + "customer_id": customer_id, + "conversions": list(conversions), + "partial_failure": partial_failure, + } + results = [] + for i, c in enumerate(conversions): + results.append(SimpleNamespace( + conversion_action=( + c.conversion_action if i < self._results_count else "" + ), + gclid="", + # API echoes user_identifiers back even for FAILED rows. + user_identifiers=list(c.user_identifiers), + )) + partial = ( + SimpleNamespace(message=self._error_message, code=0) + if self._error_message + else SimpleNamespace(message="", code=0) + ) + return SimpleNamespace(results=results, partial_failure_error=partial) + + +class _FakeSearchRow: + def __init__( + self, name: str, resource_name: str, type_name: str = "UPLOAD_CALLS" + ): + self.conversion_action = SimpleNamespace( + name=name, + resource_name=resource_name, + type_=SimpleNamespace(name=type_name), + status=SimpleNamespace(name="ENABLED"), + id=int(resource_name.split("/")[-1]), + ) + + +class _FakeGoogleAdsService: + def __init__(self, rows: list): + self._rows = rows + self.last_query = "" + + def search(self, *, customer_id, query): + self.last_query = query + return iter(self._rows) + + +def _client_with(*, upload_service, ads_service): + return _FakeClient({ + "ConversionUploadService": upload_service, + "GoogleAdsService": ads_service, + }) + + +def _ec_client_with(*, upload, ads): + return _FakeClient({ + "ConversionUploadService": upload, + "GoogleAdsService": ads, + }) + + +# --------------------------------------------------------------------------- +# Call conversions — parse, draft (redaction/consent), apply-from-rows +# --------------------------------------------------------------------------- + +_CALL_HEADER = ( + "Caller's Phone Number,Call Start Time,Conversion Name," + "Conversion Time,Conversion Value,Conversion Currency\n" +) + + +class TestParseCallConversionCsv: + def test_missing_file(self, tmp_path): + rows, errors = conversion_actions._parse_call_conversion_csv( + str(tmp_path / "nope.csv") + ) + assert rows == [] + assert any("not found" in e for e in errors) + + def test_skips_parameters_row_and_normalizes(self, tmp_path): + p = tmp_path / "phone.csv" + p.write_text( + "Parameters:TimeZone=America/Los_Angeles,,,,,\n" + + _CALL_HEADER + + "+15555550142,2026-03-01T12:00:00Z,My Action," + "2026-03-01T13:00:00Z,250.00,usd\n" + ) + rows, errors = conversion_actions._parse_call_conversion_csv(str(p)) + assert errors == [] + assert len(rows) == 1 + assert rows[0]["caller_id"] == "+15555550142" + assert rows[0]["call_start_time"] == "2026-03-01 12:00:00+00:00" + assert rows[0]["currency_code"] == "USD" + + def test_missing_required_column(self, tmp_path): + p = tmp_path / "phone.csv" + p.write_text( + "Caller's Phone Number,Conversion Name,Conversion Time," + "Conversion Value,Conversion Currency\n" + "+15555550142,X,2026-03-01T13:00:00Z,10,USD\n" + ) + rows, errors = conversion_actions._parse_call_conversion_csv(str(p)) + assert rows == [] + assert any("Call Start Time" in e for e in errors) + + +class TestDraftUploadCallConversions: + def _write(self, tmp_path): + p = tmp_path / "phone.csv" + p.write_text( + _CALL_HEADER + + "+15555550142,2026-03-01T12:00:00Z,A," + "2026-03-01T13:00:00Z,250.00,USD\n" + "+15555550143,2026-03-02T12:00:00Z,A," + "2026-03-02T13:00:00Z,500.00,USD\n" + "+15555550144,2026-03-03T12:00:00Z,B," + "2026-03-03T13:00:00Z,75.00,USD\n" + ) + return str(p) + + def test_missing_csv_returns_error(self, config, tmp_path): + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", + csv_path=str(tmp_path / "missing.csv"), + ) + assert "error" in result + + def test_happy_path_preview(self, config, tmp_path): + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", csv_path=path, + ) + assert result["operation"] == "upload_call_conversions" + assert result["entity_type"] == "call_conversion_batch" + c = result["changes"] + assert c["row_count"] == 3 + assert c["total_value"] == 825.00 + assert c["distinct_conversion_actions"] == ["A", "B"] + + def test_preview_never_contains_csv_path(self, config, tmp_path): + # Apply reads from plan rows, not the CSV — path is not persisted. + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", csv_path=path, + ) + assert "csv_path" not in result["changes"] + + def test_sample_rows_redact_caller_id(self, config, tmp_path): + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", csv_path=path, + ) + for s in result["changes"]["sample_rows"]: + assert "***" in s["caller_id"] + assert s["caller_id"] != "+15555550142" + + def test_plan_rows_keep_raw_caller_id(self, config, tmp_path): + # Rows in the plan MUST keep the raw caller_id (apply needs it). + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", csv_path=path, + ) + plan = _stored_plan(result) + assert plan.changes["rows"][0]["caller_id"] == "+15555550142" + + def test_consent_stored_in_plan(self, config, tmp_path): + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + config, customer_id="1234567890", csv_path=path, + consent={"ad_user_data": "GRANTED", + "ad_personalization": "DENIED"}, + ) + plan = _stored_plan(result) + assert plan.changes["consent"] == { + "ad_user_data": "GRANTED", "ad_personalization": "DENIED", + } + + def test_safety_blocked_operation(self, tmp_path): + cfg = AdLoopConfig( + ads=AdsConfig(customer_id="123-456-7890"), + safety=SafetyConfig( + blocked_operations=["upload_call_conversions"] + ), + ) + path = self._write(tmp_path) + result = conversion_actions.draft_upload_call_conversions( + cfg, customer_id="1234567890", csv_path=path, + ) + assert "error" in result + + +class TestApplyUploadCallConversions: + def _changes(self, consent=None): + rows = [ + { + "caller_id": "+15555550142", + "call_start_time": "2026-03-01 12:00:00+00:00", + "conversion_name": "My Action", + "conversion_time": "2026-03-01 13:00:00+00:00", + "conversion_value": 250.0, + "currency_code": "USD", + }, + { + "caller_id": "+15555550143", + "call_start_time": "2026-03-02 12:00:00+00:00", + "conversion_name": "My Action", + "conversion_time": "2026-03-02 13:00:00+00:00", + "conversion_value": 500.0, + "currency_code": "USD", + }, + ] + changes = {"rows": rows, "partial_failure": True} + if consent is not None: + changes["consent"] = consent + return changes + + def test_builds_protos_from_rows_no_csv(self, tmp_path): + upload = _FakeUploadService(results_count=2) + ads = _FakeGoogleAdsService([ + _FakeSearchRow("My Action", "customers/1/conversionActions/777") + ]) + client = _client_with(upload_service=upload, ads_service=ads) + + result = conversion_actions._apply_upload_call_conversions( + client, "1", self._changes() + ) + assert result["uploaded_total"] == 2 + assert result["success_count"] == 2 + assert result["failure_count"] == 0 + sent = upload.called_with["conversions"] + assert sent[0].caller_id == "+15555550142" + assert sent[0].conversion_action == ( + "customers/1/conversionActions/777" + ) + assert sent[0].conversion_value == 250.0 + assert sent[0].call_start_date_time == "2026-03-01 12:00:00+00:00" + + def test_success_count_keys_off_conversion_action_not_caller_id( + self, tmp_path + ): + # Only 1 of 2 rows accepted — caller_id is echoed back on BOTH, so a + # caller_id-based count would wrongly report 2. We must report 1. + upload = _FakeUploadService(results_count=1) + ads = _FakeGoogleAdsService([ + _FakeSearchRow("My Action", "customers/1/conversionActions/777") + ]) + client = _client_with(upload_service=upload, ads_service=ads) + result = conversion_actions._apply_upload_call_conversions( + client, "1", self._changes() + ) + assert result["success_count"] == 1 + assert result["failure_count"] == 1 + + def test_zero_matched_reports_zero_success(self, tmp_path): + upload = _FakeUploadService(results_count=0) + ads = _FakeGoogleAdsService([ + _FakeSearchRow("My Action", "customers/1/conversionActions/777") + ]) + client = _client_with(upload_service=upload, ads_service=ads) + result = conversion_actions._apply_upload_call_conversions( + client, "1", self._changes() + ) + assert result["success_count"] == 0 + assert result["failure_count"] == 2 + + def test_consent_applied_to_protos(self, tmp_path): + upload = _FakeUploadService(results_count=2) + ads = _FakeGoogleAdsService([ + _FakeSearchRow("My Action", "customers/1/conversionActions/777") + ]) + client = _client_with(upload_service=upload, ads_service=ads) + conversion_actions._apply_upload_call_conversions( + client, "1", + self._changes(consent={ + "ad_user_data": "GRANTED", + "ad_personalization": "DENIED", + }), + ) + sent = upload.called_with["conversions"] + enums = client.enums.ConsentStatusEnum + assert sent[0].consent.ad_user_data == enums.GRANTED + assert sent[0].consent.ad_personalization == enums.DENIED + + def test_wrong_type_raises(self, tmp_path): + upload = _FakeUploadService() + ads = _FakeGoogleAdsService([ + _FakeSearchRow( + "My Action", "customers/1/conversionActions/777", + type_name="UPLOAD_CLICKS", + ) + ]) + client = _client_with(upload_service=upload, ads_service=ads) + with pytest.raises(ValueError) as exc: + conversion_actions._apply_upload_call_conversions( + client, "1", self._changes() + ) + assert "UPLOAD_CALLS" in str(exc.value) + + def test_action_not_found_raises(self, tmp_path): + upload = _FakeUploadService() + ads = _FakeGoogleAdsService([]) + client = _client_with(upload_service=upload, ads_service=ads) + with pytest.raises(ValueError) as exc: + conversion_actions._apply_upload_call_conversions( + client, "1", self._changes() + ) + assert "not found" in str(exc.value) + + def test_empty_rows_returns_error(self, tmp_path): + upload = _FakeUploadService() + ads = _FakeGoogleAdsService([]) + client = _client_with(upload_service=upload, ads_service=ads) + result = conversion_actions._apply_upload_call_conversions( + client, "1", {"rows": [], "partial_failure": True} + ) + assert "error" in result + assert upload.called_with is None + + +# --------------------------------------------------------------------------- +# EC for Leads — parse+hash, draft, apply-from-rows, order_id, success_count +# --------------------------------------------------------------------------- + +_EC_HEADER = ( + "Email,Phone Number,First Name,Last Name,Conversion Name," + "Conversion Time,Conversion Value,Conversion Currency" +) + + +class TestParseEcForLeadsCsvHashesPii: + def _write(self, tmp_path, content): + p = tmp_path / "ec.csv" + p.write_text(content) + return str(p) + + def test_raw_pii_is_normalized_then_hashed(self, tmp_path): + # Raw fakes in the CSV; the parser must return hashes only. + path = self._write( + tmp_path, + _EC_HEADER + "\n" + "User@Example.com,+1 (555) 555-0142,Test,User,My Action," + "2026-03-01T12:00:00Z,250.00,USD\n", + ) + rows, errors = conversion_actions._parse_ec_for_leads_csv(path) + assert errors == [] + r = rows[0] + assert r["email_sha256"] == _EMAIL_HASH + assert r["phone_sha256"] == _PHONE_HASH + assert r["first_name_sha256"] == _FIRST_HASH + assert r["last_name_sha256"] == _LAST_HASH + # No raw-PII keys leak out of the parser. + assert "email" not in r and "phone" not in r + assert "first_name" not in r and "last_name" not in r + + def test_blank_pii_hashes_to_empty(self, tmp_path): + path = self._write( + tmp_path, + _EC_HEADER + "\n" + ",+15555550142,,,My Action,2026-03-01T12:00:00Z,200.00,USD\n", + ) + rows, _ = conversion_actions._parse_ec_for_leads_csv(path) + assert rows[0]["email_sha256"] == "" + assert rows[0]["first_name_sha256"] == "" + assert rows[0]["phone_sha256"] == _PHONE_HASH + + def test_order_id_optional_column(self, tmp_path): + path = self._write( + tmp_path, + _EC_HEADER + ",Order ID\n" + "user@example.com,+15555550142,Test,User,My Action," + "2026-03-01T12:00:00Z,250.00,USD,ORD-001\n", + ) + rows, _ = conversion_actions._parse_ec_for_leads_csv(path) + assert rows[0]["order_id"] == "ORD-001" + + def test_order_id_defaults_empty_when_absent(self, tmp_path): + path = self._write( + tmp_path, + _EC_HEADER + "\n" + "user@example.com,+15555550142,Test,User,My Action," + "2026-03-01T12:00:00Z,250.00,USD\n", + ) + rows, _ = conversion_actions._parse_ec_for_leads_csv(path) + assert rows[0]["order_id"] == "" + + def test_missing_required_column(self, tmp_path): + path = self._write( + tmp_path, + "Email,Phone Number,Conversion Name,Conversion Time," + "Conversion Value,Conversion Currency\n" + "user@example.com,+15555550142,X,2026-03-01T12:00:00Z,10,USD\n", + ) + rows, errors = conversion_actions._parse_ec_for_leads_csv(path) + assert rows == [] + assert any("First Name" in e or "Last Name" in e for e in errors) + + +class TestDraftUploadEcForLeads: + def _write(self, tmp_path, *, order_id=False): + header = _EC_HEADER + (",Order ID" if order_id else "") + r1 = ("user@example.com,+15555550142,Test,User,Job Close," + "2026-03-01T12:00:00Z,500.00,USD") + r2 = (",+15555550143,,,Job Close," + "2026-03-02T12:00:00Z,1500.00,USD") + if order_id: + r1 += ",ORD-001" + r2 += ",ORD-002" + p = tmp_path / "ec.csv" + p.write_text(header + "\n" + r1 + "\n" + r2 + "\n") + return str(p) + + def test_happy_path_preview(self, config, tmp_path): + path = self._write(tmp_path) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + ) + ) + assert result["operation"] == "upload_enhanced_conversions_for_leads" + c = result["changes"] + assert c["row_count"] == 2 + assert c["total_value"] == 2000.00 + assert c["rows_with_email"] == 1 + assert c["rows_with_phone"] == 2 + assert c["distinct_conversion_actions"] == ["Job Close"] + + def test_no_raw_pii_in_plan_changes(self, config, tmp_path): + # The stored plan must contain ONLY hashes for PII, never raw values. + path = self._write(tmp_path) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + ) + ) + plan = _stored_plan(result) + blob = repr(plan.changes) + assert "user@example.com" not in blob + assert "+15555550142" not in blob + assert "Test" not in blob and "User" not in blob + # But the hashes ARE present in the frozen rows. + assert plan.changes["rows"][0]["email_sha256"] == _EMAIL_HASH + + def test_sample_rows_truncate_hashes(self, config, tmp_path): + path = self._write(tmp_path) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + ) + ) + assert "..." in result["changes"]["sample_rows"][0]["email_sha256"] + + def test_dedup_warning_when_no_order_id(self, config, tmp_path): + path = self._write(tmp_path, order_id=False) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + ) + ) + c = result["changes"] + assert c["rows_with_order_id"] == 0 + assert len(c["dedup_warnings"]) == 1 + assert "double-count" in c["dedup_warnings"][0] + + def test_dedup_no_warning_full_coverage(self, config, tmp_path): + path = self._write(tmp_path, order_id=True) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + ) + ) + c = result["changes"] + assert c["rows_with_order_id"] == 2 + assert c["dedup_warnings"] == [] + + def test_dedup_partial_coverage_warns(self, config, tmp_path): + p = tmp_path / "ec.csv" + p.write_text( + _EC_HEADER + ",Order ID\n" + "user@example.com,+15555550142,Test,User,Job Close," + "2026-03-01T12:00:00Z,500.00,USD,ORD-001\n" + "user2@example.com,+15555550143,Test,User,Job Close," + "2026-03-02T12:00:00Z,1500.00,USD,\n" + ) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=str(p), + ) + ) + c = result["changes"] + assert c["rows_with_order_id"] == 1 + assert any("1 of 2" in w for w in c["dedup_warnings"]) + + def test_consent_stored(self, config, tmp_path): + path = self._write(tmp_path) + result = ( + conversion_actions.draft_upload_enhanced_conversions_for_leads( + config, customer_id="1234567890", csv_path=path, + consent={"ad_user_data": "DENIED", + "ad_personalization": "GRANTED"}, + ) + ) + plan = _stored_plan(result) + assert plan.changes["consent"] == { + "ad_user_data": "DENIED", "ad_personalization": "GRANTED", + } + + +class TestApplyUploadEcForLeads: + def _changes(self, *, order_id=False, consent=None): + rows = [ + { + "email_sha256": _EMAIL_HASH, + "phone_sha256": _PHONE_HASH, + "first_name_sha256": _FIRST_HASH, + "last_name_sha256": _LAST_HASH, + "conversion_name": "My Job", + "conversion_time": "2026-03-01 12:00:00+00:00", + "conversion_value": 500.0, + "currency_code": "USD", + "order_id": "ORD-001" if order_id else "", + }, + { + "email_sha256": "", + "phone_sha256": _PHONE_HASH, + "first_name_sha256": "", + "last_name_sha256": "", + "conversion_name": "My Job", + "conversion_time": "2026-03-02 12:00:00+00:00", + "conversion_value": 1500.0, + "currency_code": "USD", + "order_id": "ORD-002" if order_id else "", + }, + ] + changes = {"rows": rows, "partial_failure": True} + if consent is not None: + changes["consent"] = consent + return changes + + def _client(self, results_count, type_name="UPLOAD_CLICKS"): + upload = _FakeClickUploadService(results_count=results_count) + ads = _FakeGoogleAdsService([ + _FakeSearchRow( + "My Job", "customers/1/conversionActions/999", + type_name=type_name, + ) + ]) + return _ec_client_with(upload=upload, ads=ads), upload + + def test_builds_user_identifiers_from_hashes_no_csv(self, tmp_path): + client, upload = self._client(results_count=2) + result = ( + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes() + ) + ) + assert result["uploaded_total"] == 2 + assert result["success_count"] == 2 + sent = upload.called_with["conversions"] + # Row 1: email + phone + name = 3 identifiers. + assert len(sent[0].user_identifiers) == 3 + assert sent[0].user_identifiers[0].hashed_email == _EMAIL_HASH + assert sent[0].user_identifiers[1].hashed_phone_number == _PHONE_HASH + assert ( + sent[0].user_identifiers[2].address_info.hashed_first_name + == _FIRST_HASH + ) + # Row 2: phone only = 1 identifier. + assert len(sent[1].user_identifiers) == 1 + + def test_success_count_keys_off_conversion_action(self, tmp_path): + # user_identifiers are echoed back on ALL rows; only 1 matched. + client, _ = self._client(results_count=1) + result = ( + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes() + ) + ) + assert result["success_count"] == 1 + assert result["failure_count"] == 1 + + def test_zero_matched_reports_zero_success(self, tmp_path): + client, _ = self._client(results_count=0) + result = ( + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes() + ) + ) + assert result["success_count"] == 0 + assert result["failure_count"] == 2 + + def test_order_id_propagated_to_proto(self, tmp_path): + client, upload = self._client(results_count=2) + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes(order_id=True) + ) + sent = upload.called_with["conversions"] + assert sent[0].order_id == "ORD-001" + assert sent[1].order_id == "ORD-002" + + def test_order_id_unset_when_absent(self, tmp_path): + client, upload = self._client(results_count=2) + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes(order_id=False) + ) + sent = upload.called_with["conversions"] + assert sent[0].order_id == "" + + def test_consent_applied(self, tmp_path): + client, upload = self._client(results_count=2) + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", + self._changes(consent={ + "ad_user_data": "GRANTED", + "ad_personalization": "DENIED", + }), + ) + sent = upload.called_with["conversions"] + enums = client.enums.ConsentStatusEnum + assert sent[0].consent.ad_user_data == enums.GRANTED + assert sent[0].consent.ad_personalization == enums.DENIED + + def test_wrong_type_rejected(self, tmp_path): + client, _ = self._client(results_count=0, type_name="UPLOAD_CALLS") + with pytest.raises(ValueError) as exc: + conversion_actions._apply_upload_enhanced_conversions_for_leads( + client, "1", self._changes() + ) + assert "UPLOAD_CLICKS" in str(exc.value) + + def test_gaql_escape_used_for_ec_resolver(self, tmp_path): + ads = _FakeGoogleAdsService([ + _FakeSearchRow( + "O'Brien Lead", "customers/1/conversionActions/9", + type_name="UPLOAD_CLICKS", + ) + ]) + conversion_actions._resolve_upload_clicks_action( + _ec_client_with(upload=_FakeClickUploadService(), ads=ads), + "1", ["O'Brien Lead"], + ) + assert "O\\'Brien Lead" in ads.last_query + assert "O''Brien" not in ads.last_query + + +# --------------------------------------------------------------------------- +# Audit-log redaction (write._redact_changes_for_audit) +# --------------------------------------------------------------------------- + + +class TestAuditRedaction: + def test_call_conversion_caller_id_redacted_in_audit(self): + changes = { + "row_count": 1, + "rows": [{ + "caller_id": "+15555550142", + "call_start_time": "2026-03-01 12:00:00+00:00", + "conversion_name": "A", + "conversion_time": "2026-03-01 13:00:00+00:00", + "conversion_value": 250.0, + "currency_code": "USD", + }], + } + redacted = write._redact_changes_for_audit( + "upload_call_conversions", changes + ) + blob = repr(redacted) + assert "+15555550142" not in blob + assert "+155***0142" in blob + # Original object is NOT mutated. + assert changes["rows"][0]["caller_id"] == "+15555550142" + + def test_ec_rows_dropped_from_audit(self): + changes = { + "row_count": 1, + "rows_with_email": 1, + "rows": [{ + "email_sha256": _EMAIL_HASH, + "phone_sha256": _PHONE_HASH, + "first_name_sha256": _FIRST_HASH, + "last_name_sha256": _LAST_HASH, + "conversion_name": "Job", + "conversion_time": "2026-03-01 12:00:00+00:00", + "conversion_value": 500.0, + "currency_code": "USD", + "order_id": "", + }], + } + redacted = write._redact_changes_for_audit( + "upload_enhanced_conversions_for_leads", changes + ) + assert "rows" not in redacted + assert "rows_redacted" in redacted + # Summary counters survive. + assert redacted["rows_with_email"] == 1 + # Original object is NOT mutated. + assert "rows" in changes + + def test_non_upload_ops_pass_through(self): + changes = {"campaign_name": "X"} + assert ( + write._redact_changes_for_audit("create_campaign", changes) + is changes + ) + + +# --------------------------------------------------------------------------- +# Upload tool registration + dispatch wiring +# --------------------------------------------------------------------------- + + +class TestUploadToolRegistration: + @pytest.fixture(scope="class") + @classmethod + def tools_by_name(cls): + import asyncio + from adloop.server import mcp + + async def _list(): + return await mcp.list_tools() + + tools = asyncio.run(_list()) + return {t.name: t for t in tools} + + def test_upload_tools_registered(self, tools_by_name): + assert "draft_upload_call_conversions" in tools_by_name + assert ( + "draft_upload_enhanced_conversions_for_leads" in tools_by_name + ) + + def test_call_upload_requires_csv_path(self, tools_by_name): + required = ( + tools_by_name["draft_upload_call_conversions"] + .parameters.get("required", []) + ) + assert "csv_path" in required + + def test_ec_upload_requires_csv_path(self, tools_by_name): + required = ( + tools_by_name["draft_upload_enhanced_conversions_for_leads"] + .parameters.get("required", []) + ) + assert "csv_path" in required + + def test_upload_dispatch_wired(self): + import inspect + src = inspect.getsource(write._execute_plan) + assert ( + '"upload_call_conversions": _apply_upload_call_conversions' + in src + ) + assert "upload_enhanced_conversions_for_leads" in src