Agent/examples/auto_put_ad_mini/tools/scene_spec.py

"""腾讯广告"版位定投场景"标签查询 + 进程内缓存(P0 一账户 N 广告差异化基础设施)。

数据流(2026-06-09 用户确认):
  主循环启动时 → get_wechat_position_tags(account_id)
  缓存中:返回内存里的 dict
  缓存过期/未填:调腾讯 /scene_spec_tags/get(type=WECHAT_POSITION) → 缓存 → 返回

用途:
  - 运行时校验 config.WECHAT_POSITION_TARGETED_PRESET 中的 ID 仍然合法(腾讯未弃用)
  - 给飞书审批表展示中文场景名(P0-C 扩展时)
  - 日志可读化(把 1024795 → "小程序激励式广告")

API 参考:
  POST /scene_spec_tags/get
  必填:account_id + type (enum,实测 WECHAT_POSITION 可用)
  返回:data.list[{id, name, description, parent_id, targeting_name}]
"""

import logging
import time
from typing import Optional

from tools.ad_api import _get

logger = logging.getLogger(__name__)

# 缓存键 → "type 名"(如 'WECHAT_POSITION');TTL 1 小时(用户 2026-06-09 确认)
_CACHE: dict = {}
_TTL_SECONDS = 3600


def _now() -> float:
    return time.time()


def get_wechat_position_tags(
    account_id: int,
    force_refresh: bool = False,
) -> dict[int, str]:
    """查询 WECHAT_POSITION 全部场景 ID → 中文 description 映射(带 1h 缓存)。

    Args:
        account_id: 腾讯账户 ID(腾讯接口要求必填,但场景列表跟账户无关 — 任一账户的返回值通用)
        force_refresh: 强制刷新缓存

    Returns:
        {场景 ID(int): 中文名(str)}  — 13 个映射(2026-06-09 实测)
    """
    cache_key = "WECHAT_POSITION"
    cached = _CACHE.get(cache_key)
    if (
        not force_refresh
        and cached is not None
        and (_now() - cached["ts"]) < _TTL_SECONDS
    ):
        return cached["data"]

    logger.info(
        "[scene_spec] cache miss/expired,调 /scene_spec_tags/get type=%s",
        cache_key,
    )
    resp = _get("/scene_spec_tags/get", {
        "account_id": account_id, "type": cache_key,
    })
    if resp.get("code") != 0:
        raise RuntimeError(
            f"scene_spec_tags/get 失败 code={resp.get('code')} "
            f"msg={resp.get('message_cn') or resp.get('message')}"
        )
    items = (resp.get("data") or {}).get("list") or []
    mapping = {int(it["id"]): str(it.get("description") or "") for it in items}

    _CACHE[cache_key] = {"data": mapping, "ts": _now()}
    logger.info(
        "[scene_spec] 缓存已更新 type=%s 共 %d 项",
        cache_key, len(mapping),
    )
    return mapping


def validate_preset_ids(account_id: int, preset_ids: list[int]) -> dict:
    """校验给定 ID 列表是否都在腾讯当前合法场景集合内。

    用于主循环启动时检查 config.WECHAT_POSITION_TARGETED_PRESET 仍有效。

    Returns:
        {"all_valid": bool, "invalid_ids": list[int], "mapping": {id: 中文}}
    """
    mapping = get_wechat_position_tags(account_id)
    invalid = [i for i in preset_ids if i not in mapping]
    return {
        "all_valid": not invalid,
        "invalid_ids": invalid,
        "mapping": {i: mapping.get(i, "<未知>") for i in preset_ids},
    }


def describe_position_ids(account_id: int, ids: list[int]) -> str:
    """把场景 ID 列表转成 "公众号文章底部+中部+..." 的可读字符串(日志 / 审批表用)。"""
    if not ids:
        return "(无定投)"
    mapping = get_wechat_position_tags(account_id)
    return " + ".join(mapping.get(i, f"<ID {i}>") for i in ids)