Agent/examples/auto_put_ad_mini/tools/landing_plan.py

"""落地页计划注册接口(piaoquantv `xcx/save`)适配层。

业务含义(2026-06-08 用户确认):
  每条腾讯创意挂上去前,先到 piaoquantv 服务侧"注册一个落地计划",
  服务侧返回:
    - pageUrl:小程序跳转路径,**直接作为 creative_components.jump_info.mini_program_path**
    - pageUrl 里嵌入了服务侧生成的 rootSourceId,**直接作为 dynamic_creative_name**(归因锚点)

接口契约(严格按用户给的 curl,不增不减字段):
  POST https://tp-open.piaoquantv.com/contentPlatform/plan/xcx/save
  header: token: <fixed>
  body: {
    "audiencePackage": str,              # 账户级 crowd_package
    "videoList": [                       # 单条逐次调,不批量
      {"videoId": int, "title": str, "cover": str, "video": str, "experimentId": str}
    ]
  }

重复调行为(用户确认):**新建** — 同 video 重复调返回不同 plan_id + 不同 rootSourceId + 不同 shareId。
  客户端无需缓存 plan_id;每条腾讯创意对应一次 xcx/save 调用。
"""

import logging
import os
from dataclasses import dataclass, field
from typing import Optional
from urllib.parse import parse_qs, unquote

import httpx

from tools.video_recall import LandingVideo, PIAOQUANTV_TOKEN

logger = logging.getLogger(__name__)

XCX_SAVE_URL = os.getenv(
    "PIAOQUANTV_XCX_SAVE_URL",
    "https://tp-open.piaoquantv.com/contentPlatform/plan/xcx/save",
)


@dataclass
class LandingPlanResult:
    """xcx/save 返回的落地计划。

    业务核心字段:
      page_url       — 直接传给腾讯创意的 mini_program_path
      root_source_id — 直接用作 dynamic_creative_name(归因锚点)
      share_id       — pageUrl 里的 rootShareId/shareId(实测两者相同)
    """

    plan_id: int                     # piaoquantv 平台计划 ID(自增,8059)
    page_url: str                    # `pages/category?jumpPage=...`
    root_source_id: str              # 从 pageUrl 解析 — 也用作 creative_name
    share_id: str                    # 从 pageUrl 解析
    video_id: int                    # 入参回显,便于追溯
    audience_package: str
    raw: dict = field(default_factory=dict, repr=False)


def _parse_inner_query(page_url: str) -> dict:
    """解析 pageUrl 内层 query。

    pageUrl 是双层 URL:
      外层:`pages/category?jumpPage=<URL-encoded inner>`
      内层:`pages/user-videos?fromGzh=1&rootShareId=...&id=...&shareId=...&rootSourceId=...`

    返回:`{key: value}`(取 parse_qs 的第一个值)
    """
    q_idx = page_url.find("?")
    if q_idx < 0:
        raise ValueError(f"pageUrl 无 query 段: {page_url}")
    outer = page_url[q_idx + 1:]
    if not outer.startswith("jumpPage="):
        raise ValueError(f"pageUrl 外层 query 不是 jumpPage=: {outer[:80]}")
    inner_encoded = outer[len("jumpPage="):]
    inner_url = unquote(inner_encoded)

    inner_q_idx = inner_url.find("?")
    if inner_q_idx < 0:
        raise ValueError(f"jumpPage 解码后无 query: {inner_url}")
    inner_query = inner_url[inner_q_idx + 1:]

    parsed = parse_qs(inner_query, keep_blank_values=True)
    return {k: v[0] for k, v in parsed.items() if v}


def create_landing_plan(
    crowd_package: str,
    video: LandingVideo,
    timeout: int = 30,
) -> LandingPlanResult:
    """调 xcx/save 注册一个落地计划,返回可直接挂创意的 pageUrl + creative_name。

    Args:
        crowd_package: 账户级 crowd_package(来自 account_whitelist)
        video: LandingVideo(必须含 video_id/title/cover_url/video_url/experiment_id)
    """
    if not video.experiment_id:
        raise ValueError(
            f"video_id={video.video_id} 缺 experiment_id,xcx/save 入参不完整"
        )

    body = {
        "audiencePackage": crowd_package,
        "videoList": [{
            "videoId": video.video_id,
            "title": video.title,
            "cover": video.cover_url,
            "video": video.video_url,
            "experimentId": video.experiment_id,
        }],
    }
    headers = {
        "Content-Type": "application/json",
        "token": PIAOQUANTV_TOKEN,
        "Accept": "application/json, text/plain, */*",
    }

    logger.info(
        "[landing_plan] POST xcx/save crowd=%r videoId=%d experimentId=%s",
        crowd_package, video.video_id, video.experiment_id,
    )
    resp = httpx.post(XCX_SAVE_URL, json=body, headers=headers, timeout=timeout)
    resp.raise_for_status()
    data = resp.json()

    if data.get("code") != 0 or not data.get("success"):
        raise RuntimeError(
            f"xcx/save 失败 videoId={video.video_id} "
            f"code={data.get('code')} msg={data.get('msg')}"
        )

    items = data.get("data") or []
    if not items:
        raise RuntimeError(
            f"xcx/save 返回 data 为空 videoId={video.video_id}"
        )
    item = items[0]

    page_url = item.get("pageUrl") or ""
    if not page_url:
        raise RuntimeError(
            f"xcx/save 返回 pageUrl 为空 videoId={video.video_id} item={item}"
        )

    inner = _parse_inner_query(page_url)
    root_source_id = inner.get("rootSourceId") or ""
    share_id = inner.get("shareId") or ""
    if not root_source_id:
        raise RuntimeError(
            f"pageUrl 内层未解析出 rootSourceId: {page_url}"
        )

    logger.info(
        "[landing_plan] OK plan_id=%s rootSourceId=%s",
        item.get("id"), root_source_id,
    )
    return LandingPlanResult(
        plan_id=int(item.get("id") or 0),
        page_url=page_url,
        root_source_id=root_source_id,
        share_id=share_id,
        video_id=video.video_id,
        audience_package=crowd_package,
        raw=item,
    )