maskanx_cm_backend/src/adclaw/meta/objects.py

# -*- coding: utf-8 -*-
"""Pure mappers from a CampaignSpec's stored JSON blobs to Graph API params.

These functions do no I/O and make no Meta calls; they only reshape data so
`MetaClient`'s write methods stay transport-only and this mapping logic is
unit-testable without a fake transport at all. Callers spread the returned
dict as keyword arguments into the matching `MetaClient` method, e.g.:

    await client.create_campaign(ad_account_id, **build_campaign_payload(spec))

Field mapping decisions
------------------------
Phase 2 does not yet have separate AdSet/Ad spec models — `CampaignSpec` is
the only spec in play — so ad-set- and ad-level fields that don't fit
`CampaignSpec`'s campaign-level columns are read out of `spec.advanced`.
This is a deliberate, documented choice rather than an oversight:

- `spec.budget["daily_budget"]` -> ad set `daily_budget`, passed through
  unchanged (already minor currency units / cents, matching
  `adclaw.campaigns.validation` and Graph's own expectation). Lifetime
  budgets are out of scope for this mapping: `daily_budget` is required and
  `build_ad_set_payload` raises `ValueError` if it is missing.
- `spec.targeting["age_min"]` / `["age_max"]` -> Graph
  `targeting.age_min` / `targeting.age_max`.
- `spec.targeting["countries"]` -> Graph
  `targeting.geo_locations.countries`.
- `spec.targeting["genders"]` -> Graph `targeting.genders`, if present.
- `spec.advanced["special_ad_categories"]` -> campaign
  `special_ad_categories` (defaults to an empty list). A bare string is
  normalised to a single-element list rather than exploded into
  characters, since `advanced` is unvalidated free-form JSON and this is
  the compliance field for regulated advertising (housing, credit,
  employment, ...).
- `spec.advanced["optimization_goal"]` / `["billing_event"]` -> ad set
  fields of the same name, defaulting to `LEAD_GENERATION` /
  `IMPRESSIONS` (MaskanX's default lead-gen objective). Both are checked
  against an allowlist (`ALLOWED_OPTIMIZATION_GOALS` /
  `ALLOWED_BILLING_EVENTS`) before being forwarded, since `advanced` is
  client-controlled and `billing_event` in particular determines how the
  ad account is charged; an unrecognised value raises `ValueError` naming
  the offending value and the allowed set rather than reaching Meta.
- `spec.advanced["message"]` / `["headline"]` / `["description"]` /
  `["link"]` -> ad creative `object_story_spec.link_data` fields of the
  same purpose (headline defaults to the campaign name if unset).
"""
from __future__ import annotations

from typing import Any

from ..campaigns.models import CampaignSpec

DEFAULT_OPTIMIZATION_GOAL = "LEAD_GENERATION"
DEFAULT_BILLING_EVENT = "IMPRESSIONS"

# Graph requires a bid strategy on every ad set that carries its own budget,
# and every strategy except this one additionally requires a bid amount or
# bid constraints. Sending nothing is rejected with code 100 / subcode
# 2490487, which is what blocked the first live sync.
#
# LOWEST_COST_WITHOUT_CAP is automatic bidding: Meta spends the daily budget
# as efficiently as it can and never exceeds it. The alternatives set a
# per-result ceiling instead, which controls unit price but not total spend
# — the opposite of what the guardrails need.
DEFAULT_BID_STRATEGY = "LOWEST_COST_WITHOUT_CAP"

# Only one strategy needs nothing else from us. COST_CAP and
# LOWEST_COST_WITH_BID_CAP need a bid amount; LOWEST_COST_WITH_MIN_ROAS
# needs bid_constraints plus a VALUE optimisation goal. CampaignSpec
# carries none of those, so accepting any of them here would just recreate
# subcode 2490487 at sync time — after the campaign object already exists.
# Widen this only together with the fields the strategy requires.
ALLOWED_BID_STRATEGIES = frozenset({"LOWEST_COST_WITHOUT_CAP"})

# advanced.optimization_goal / advanced.billing_event are client-controlled
# and billing_event determines how the ad account is charged, so both are
# checked against an allowlist before being forwarded to a spending API.
ALLOWED_OPTIMIZATION_GOALS = frozenset({
    "LEAD_GENERATION", "LINK_CLICKS", "IMPRESSIONS", "REACH",
    "OFFSITE_CONVERSIONS", "LANDING_PAGE_VIEWS", "THRUPLAY",
    "POST_ENGAGEMENT", "QUALITY_CALL",
})
ALLOWED_BILLING_EVENTS = frozenset({
    "IMPRESSIONS", "LINK_CLICKS", "THRUPLAY", "POST_ENGAGEMENT",
})


def build_campaign_payload(spec: CampaignSpec) -> dict[str, Any]:
    """Map a CampaignSpec onto `MetaClient.create_campaign` kwargs.

    Raises `ValueError` if `spec.objective` is unset — Graph requires an
    objective on every campaign and there is no sensible default to guess.
    """
    if not spec.objective:
        raise ValueError(
            "CampaignSpec.objective is required to create a Meta campaign.",
        )
    # A bare string ("HOUSING") must become a single-element list, not be
    # exploded into characters by list(); advanced is unvalidated JSON from
    # the API and special_ad_categories is the compliance field for
    # regulated advertising, so this cannot be left to Meta's opaque error.
    raw_categories = spec.advanced.get("special_ad_categories") or []
    categories = (
        [raw_categories] if isinstance(raw_categories, str) else list(raw_categories)
    )
    return {
        "name": spec.name,
        "objective": spec.objective,
        "special_ad_categories": categories,
    }


def build_ad_set_payload(spec: CampaignSpec, campaign_id: str) -> dict[str, Any]:
    """Map a CampaignSpec onto `MetaClient.create_ad_set` kwargs.

    Raises `ValueError` if `spec.budget["daily_budget"]` is unset — this
    mapping does not yet support lifetime-budget campaigns.
    """
    daily_budget = spec.budget.get("daily_budget")
    if daily_budget is None:
        raise ValueError(
            "CampaignSpec.budget.daily_budget is required to create an ad "
            "set (lifetime budgets are not yet supported).",
        )

    targeting: dict[str, Any] = {}
    age_min = spec.targeting.get("age_min")
    age_max = spec.targeting.get("age_max")
    if age_min is not None:
        targeting["age_min"] = age_min
    if age_max is not None:
        targeting["age_max"] = age_max
    countries = spec.targeting.get("countries")
    if countries:
        targeting["geo_locations"] = {"countries": list(countries)}
    genders = spec.targeting.get("genders")
    if genders:
        targeting["genders"] = list(genders)

    # Graph requires targeting on every ad set. Catching it here names the
    # missing field; letting it through produces an opaque Meta rejection
    # part-way into building the object chain.
    if not targeting:
        raise ValueError(
            "CampaignSpec.targeting is empty, so the ad set has no audience. "
            "Set at least one of countries, age_min, age_max or genders.",
        )

    optimization_goal = spec.advanced.get(
        "optimization_goal", DEFAULT_OPTIMIZATION_GOAL,
    )
    if optimization_goal not in ALLOWED_OPTIMIZATION_GOALS:
        raise ValueError(
            f"Unsupported optimization_goal {optimization_goal!r}. Allowed: "
            f"{sorted(ALLOWED_OPTIMIZATION_GOALS)}",
        )
    billing_event = spec.advanced.get("billing_event", DEFAULT_BILLING_EVENT)
    if billing_event not in ALLOWED_BILLING_EVENTS:
        raise ValueError(
            f"Unsupported billing_event {billing_event!r}. Allowed: "
            f"{sorted(ALLOWED_BILLING_EVENTS)}",
        )

    bid_strategy = spec.advanced.get("bid_strategy", DEFAULT_BID_STRATEGY)
    if bid_strategy not in ALLOWED_BID_STRATEGIES:
        raise ValueError(
            f"Unsupported bid_strategy {bid_strategy!r}. Allowed: "
            f"{sorted(ALLOWED_BID_STRATEGIES)}. The others require a bid "
            f"amount or bid constraints, which CampaignSpec does not carry.",
        )

    return {
        "campaign_id": campaign_id,
        "name": f"{spec.name} - Ad Set",
        "daily_budget": daily_budget,
        "targeting": targeting,
        "optimization_goal": optimization_goal,
        "billing_event": billing_event,
        "bid_strategy": bid_strategy,
    }


def build_creative_payload(
    spec: CampaignSpec,
    page_id: str,
    image_hash: str,
) -> dict[str, Any]:
    """Map a CampaignSpec onto `MetaClient.create_ad_creative` kwargs."""
    return {
        "name": f"{spec.name} - Creative",
        "page_id": page_id,
        "message": spec.advanced.get("message", ""),
        "headline": spec.advanced.get("headline") or spec.name,
        "description": spec.advanced.get("description", ""),
        "link": spec.advanced.get("link", ""),
        "image_hash": image_hash,
    }