Agent/examples/auto_put_ad_mini/tools/audience_grant.py

"""腾讯广告人群包解析、授权、可见性验证。

Phase 0 创建广告前调用,保证非泛人群账户的 custom_audience 已经在目标账户可见。
"""

from __future__ import annotations

import logging
import os
import time
from typing import Optional

from tools.ad_api import _check, _get, _post

logger = logging.getLogger(__name__)

DEFAULT_AUDIENCE_SOURCE_ACCOUNT_ID = int(
    os.getenv("TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID", "55615440")
)
AUDIENCE_GRANT_BUSINESS_ID = int(os.getenv("TENCENT_AUDIENCE_GRANT_BUSINESS_ID", "0") or 0)
TARGET_PERMISSION = "GRANT_PERMISSION_TYPE_TARGET"

AUDIENCE_NAME_ALIASES = {
    # 业务配置名保留给内容服务 crowdPackage 使用;腾讯侧真实人群包命名是 R_330+。
    "回流330以上人群": ["R_330+", "R330+"],
}


def _audience_lookup_names(audience_name: str) -> list[str]:
    names = [audience_name]
    for alias in AUDIENCE_NAME_ALIASES.get(audience_name, []):
        if alias and alias not in names:
            names.append(alias)
    return names


def _fetch_custom_audiences(account_id: int) -> list[dict]:
    out: list[dict] = []
    page = 1
    while True:
        resp = _get(
            "/custom_audiences/get",
            {"account_id": account_id, "page": page, "page_size": 100},
        )
        data = _check(resp, "custom_audiences/get")
        out.extend(data.get("list") or [])
        page_info = data.get("page_info") or {}
        total_page = int(page_info.get("total_page") or page)
        if page >= total_page:
            break
        page += 1
    return out


def _is_audience_visible(account_id: int, audience_id: int) -> bool:
    return any(
        int(item.get("audience_id") or 0) == int(audience_id)
        for item in _fetch_custom_audiences(account_id)
    )


def _pick_source_audience(source_account_id: int, audience_name: str) -> dict:
    items = _fetch_custom_audiences(source_account_id)
    usable = [
        item for item in items
        if item.get("status") == "SUCCESS"
        and item.get("online_status") == "ONLINE"
    ]
    lookup_names = _audience_lookup_names(audience_name)

    def score_one(item_name: str, lookup_name: str) -> int:
        if item_name == lookup_name:
            return 3
        if item_name.startswith(f"{lookup_name}_"):
            return 2
        if lookup_name in item_name:
            return 1
        return 0

    def score(item: dict) -> tuple[int, str]:
        name = item.get("name") or ""
        rank = max(score_one(name, lookup_name) for lookup_name in lookup_names)
        return rank, str(item.get("created_time") or "")

    matches = [item for item in usable if score(item)[0] > 0]
    if not matches:
        raise RuntimeError(
            f"source_account={source_account_id} 未找到 ONLINE/SUCCESS 人群包:"
            f" audience_name={audience_name!r}, lookup_names={lookup_names!r}"
        )

    matches.sort(key=score, reverse=True)
    chosen = matches[0]
    logger.info(
        "[audience_grant] source=%d audience_name=%r lookup_names=%r → audience_id=%s full_name=%r",
        source_account_id,
        audience_name,
        lookup_names,
        chosen.get("audience_id"),
        chosen.get("name"),
    )
    return chosen


def _load_account_config(account_id: int) -> Optional[dict]:
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT account_id, audience_name, audience_pack_id,
                       audience_source_account_id, audience_grant_status
                FROM ad_creation_account_config
                WHERE account_id=%s AND enabled=TRUE
                """,
                (account_id,),
            )
            return cur.fetchone()
    finally:
        conn.close()


def _write_audience_state(
    account_id: int,
    audience_id: Optional[int],
    audience_name: str,
    source_account_id: Optional[int],
    status: str,
    verified: bool = False,
) -> None:
    from db.connection import get_connection

    verified_sql = "audience_verified_at=NOW()," if verified else ""
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                f"""
                UPDATE ad_creation_account_config
                SET audience_pack_id=%s,
                    audience_tier_label=%s,
                    audience_source_account_id=%s,
                    audience_grant_status=%s,
                    {verified_sql}
                    updated_by='auto-audience-grant'
                WHERE account_id=%s
                """,
                (audience_id, audience_name, source_account_id, status, account_id),
            )
            cur.execute(
                """
                UPDATE account_whitelist
                SET audience_pack_id=%s,
                    audience_tier_label=%s,
                    updated_by='auto-audience-grant'
                WHERE account_id=%s
                """,
                (audience_id, audience_name, account_id),
            )
        conn.commit()
    finally:
        conn.close()


def _grant_audience_to_account(
    source_account_id: int,
    target_account_id: int,
    audience_id: int,
    grant_business_id: int,
) -> None:
    body = {
        "account_id": source_account_id,
        "audience_id_list": [int(audience_id)],
        "grant_type": "GRANT_TYPE_BUSINESS",
        "grant_spec": {
            "grant_to_business_spec": {
                "grant_business_id": int(grant_business_id),
                "grant_scope_type": "GRANT_SCOPE_TYPE_ACCOUNT",
                "grant_account_permission": [
                    {
                        "account_id": int(target_account_id),
                        "grant_permission_type_list": [TARGET_PERMISSION],
                    }
                ],
            }
        },
    }
    resp = _post("/audience_grant_relations/add", body)
    if resp.get("code") not in (0, "0"):
        raise RuntimeError(
            f"audience_grant_relations/add 失败 code={resp.get('code')} "
            f"msg={resp.get('message_cn') or resp.get('message')}"
        )


def ensure_account_audience_grant(
    account_id: int,
    poll_attempts: int = 12,
    poll_interval_seconds: int = 5,
) -> dict:
    """确保 account_id 的待投放人群包已在目标账户可见。

    Returns:
        {"status": str, "audience_id": Optional[int], "audience_name": str}
    """
    cfg = _load_account_config(account_id)
    if not cfg:
        raise RuntimeError(f"account_id={account_id} 未配置 ad_creation_account_config")

    audience_name = (cfg.get("audience_name") or "").strip()
    if not audience_name:
        raise RuntimeError(f"account_id={account_id} audience_name 为空")

    if audience_name == "泛人群":
        _write_audience_state(
            account_id=account_id,
            audience_id=None,
            audience_name=audience_name,
            source_account_id=None,
            status="not_required_no_audience_pack",
            verified=True,
        )
        logger.info("[audience_grant] account=%d 泛人群,无需授权", account_id)
        return {"status": "not_required_no_audience_pack", "audience_id": None, "audience_name": audience_name}

    source_account_id = int(
        cfg.get("audience_source_account_id") or DEFAULT_AUDIENCE_SOURCE_ACCOUNT_ID
    )
    audience_id = cfg.get("audience_pack_id")
    if audience_id is None:
        chosen = _pick_source_audience(source_account_id, audience_name)
        audience_id = int(chosen["audience_id"])
        _write_audience_state(
            account_id=account_id,
            audience_id=audience_id,
            audience_name=audience_name,
            source_account_id=source_account_id,
            status="resolved",
        )
    audience_id = int(audience_id)

    if _is_audience_visible(account_id, audience_id):
        _write_audience_state(
            account_id=account_id,
            audience_id=audience_id,
            audience_name=audience_name,
            source_account_id=source_account_id,
            status="verified",
            verified=True,
        )
        logger.info(
            "[audience_grant] account=%d audience_id=%d 已可见",
            account_id, audience_id,
        )
        return {"status": "verified", "audience_id": audience_id, "audience_name": audience_name}

    if not AUDIENCE_GRANT_BUSINESS_ID:
        raise RuntimeError(
            "缺少 TENCENT_AUDIENCE_GRANT_BUSINESS_ID,无法自动授权人群包"
        )

    logger.info(
        "[audience_grant] 授权 source=%d audience_id=%d → target=%d bm=%d",
        source_account_id,
        audience_id,
        account_id,
        AUDIENCE_GRANT_BUSINESS_ID,
    )
    _grant_audience_to_account(
        source_account_id=source_account_id,
        target_account_id=account_id,
        audience_id=audience_id,
        grant_business_id=AUDIENCE_GRANT_BUSINESS_ID,
    )
    _write_audience_state(
        account_id=account_id,
        audience_id=audience_id,
        audience_name=audience_name,
        source_account_id=source_account_id,
        status="grant_submitted",
    )

    for i in range(poll_attempts):
        if i:
            time.sleep(poll_interval_seconds)
        if _is_audience_visible(account_id, audience_id):
            _write_audience_state(
                account_id=account_id,
                audience_id=audience_id,
                audience_name=audience_name,
                source_account_id=source_account_id,
                status="verified",
                verified=True,
            )
            logger.info(
                "[audience_grant] account=%d audience_id=%d 授权后验证通过 attempt=%d",
                account_id,
                audience_id,
                i + 1,
            )
            return {"status": "verified", "audience_id": audience_id, "audience_name": audience_name}

    raise RuntimeError(
        f"人群包授权后目标账户仍不可见: account={account_id} audience_id={audience_id}"
    )