Agent/examples/auto_put_ad_mini/tools/delivery_config.py

"""Account-level bidding and placement config parsed from Feishu/DB."""

from __future__ import annotations

from dataclasses import dataclass
from functools import lru_cache
import json
from pathlib import Path

BID_MODE_AVERAGE_COST = "average_cost"
BID_MODE_MAX_CONVERSION = "max_conversion"


@dataclass(frozen=True)
class PlacementConfig:
    automatic_site_enabled: bool
    site_set: list[str]


@dataclass(frozen=True)
class GeoLocationConfig:
    location_types: list[str]
    region_ids: list[int]


_BID_SCENE_VALUES = {
    "": BID_MODE_AVERAGE_COST,
    "平均成本": BID_MODE_AVERAGE_COST,
    "稳定成本": BID_MODE_AVERAGE_COST,
    "最大转化量": BID_MODE_MAX_CONVERSION,
    "最大转化量出价": BID_MODE_MAX_CONVERSION,
    "average_cost": BID_MODE_AVERAGE_COST,
    "max_conversion": BID_MODE_MAX_CONVERSION,
    # 兼容已经同步过的旧值。
    "BID_SCENE_NORMAL_AVERAGE": BID_MODE_AVERAGE_COST,
    "BID_SCENE_NORMAL_MAX": BID_MODE_MAX_CONVERSION,
}

_PLACEMENT_GROUPS = {
    "微信朋友圈": ["SITE_SET_MOMENTS"],
    "朋友圈": ["SITE_SET_MOMENTS"],
    "微信公众号与小程序": ["SITE_SET_WECHAT", "SITE_SET_WECHAT_PLUGIN"],
    "微信公众号": ["SITE_SET_WECHAT"],
    "微信插件": ["SITE_SET_WECHAT_PLUGIN"],
    "搜索场景": ["SITE_SET_SEARCH_SCENE"],
    "腾讯平台与内容媒体": [
        "SITE_SET_TENCENT_NEWS",
        "SITE_SET_TENCENT_VIDEO",
        "SITE_SET_KANDIAN",
        "SITE_SET_QQ_MUSIC_GAME",
    ],
    "腾讯营销联盟": ["SITE_SET_MOBILE_UNION"],
}

REGION_DICTIONARY_PATH = (
    Path(__file__).resolve().parents[1]
    / "data"
    / "tencent_constants"
    / "regions_all.json"
)


def validate_region_dictionary() -> Path:
    """Fail fast when the versioned Tencent region asset is missing or malformed."""
    path = REGION_DICTIONARY_PATH
    if not path.is_file():
        raise RuntimeError(f"腾讯地域字典缺失: {path}")
    try:
        regions = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"腾讯地域字典不可读取: {path}: {exc}") from exc
    if not isinstance(regions, list) or not regions:
        raise RuntimeError(f"腾讯地域字典必须是非空数组: {path}")
    for index, region in enumerate(regions):
        if not isinstance(region, dict) or "id" not in region or "name" not in region:
            raise RuntimeError(f"腾讯地域字典第 {index} 项缺少 id/name: {path}")
    return path


def parse_bid_scene(raw: object) -> str:
    text = str(raw or "").strip()
    if text in _BID_SCENE_VALUES:
        return _BID_SCENE_VALUES[text]
    raise ValueError(f"未知出价方式:{raw}")


def parse_placement_config(raw: object) -> PlacementConfig | None:
    text = str(raw or "").strip()
    if not text:
        return None
    if text.upper() == "AIM+":
        return PlacementConfig(automatic_site_enabled=True, site_set=[])

    site_set: list[str] = []
    for part in text.replace("，", ",").split(","):
        name = part.strip()
        if not name:
            continue
        values = _PLACEMENT_GROUPS.get(name)
        if values is None:
            raise ValueError(f"未知投放版位:{name}")
        for value in values:
            if value not in site_set:
                site_set.append(value)
    if not site_set:
        raise ValueError(f"投放版位为空:{raw}")
    return PlacementConfig(automatic_site_enabled=False, site_set=site_set)


@lru_cache(maxsize=1)
def _region_name_mapping() -> dict[str, int]:
    path = validate_region_dictionary()
    regions = json.loads(path.read_text(encoding="utf-8"))
    mapping: dict[str, int] = {}
    for region in regions:
        name = str(region.get("name") or "").strip()
        if not name:
            continue
        region_id = int(region["id"])
        mapping.setdefault(name, region_id)
        for suffix in ("特别行政区", "壮族自治区", "回族自治区", "维吾尔自治区", "自治区", "省", "市"):
            if name.endswith(suffix):
                mapping.setdefault(name[:-len(suffix)], region_id)
                break
    return mapping


def parse_geo_location_config(raw: object) -> GeoLocationConfig | None:
    """Parse Feishu geography: blank keeps the delivery template; 不限 omits geo targeting."""
    text = str(raw or "").strip()
    if not text or text in {"排地域", "默认地域"}:
        return None
    if text.lower() in {"不限", "不限地域", "全国", "all"}:
        return GeoLocationConfig(location_types=[], region_ids=[])

    names = [part.strip() for part in text.replace("，", ",").split(",") if part.strip()]
    if not names:
        raise ValueError(f"地域为空:{raw}")
    name_mapping = _region_name_mapping()
    region_ids: list[int] = []
    for name in names:
        try:
            region_id = int(name)
        except ValueError:
            region_id = name_mapping.get(name, 0)
        if not region_id:
            raise ValueError(f"未知地域:{name}")
        if region_id not in region_ids:
            region_ids.append(region_id)
    return GeoLocationConfig(location_types=["LIVE_IN"], region_ids=region_ids)