Agent/examples/auto_put_ad_mini/tools/material_recall.py

"""创意素材召回接口适配层(vector 服务)。

业务模型(用户 2026-06-08 确认):
  对一个承接视频(LandingVideo)执行 3 种策略召回素材,合并去重,按 score 取 top N。

接口:POST https://api-internal.piaoquantv.com/videoVector/recallTest/matchByText
鉴权:不需要(内部 API)

本质(用户 2026-06-08 确认):**多路召回 → 排序 → 去重**

实测确认素材库的 4 个有效 configCode(VIDEO_TITLE 和 ALL 对 MATERIAL 模态返回 0):
  - VIDEO_TOPIC      (选题向量)
  - VIDEO_KEYPOINT   (关键点向量)
  - VIDEO_INSPIRATION(灵感点向量)
  - VIDEO_PURPOSE    (目的点向量)

策略:对一个承接视频,选合适的 queryText,**串行调 4 个 configCode**,合并按 materialId 去重(取 max score),按 score 降序返回 top N。

queryText 选择优先级(选第一个非空且非占位符 "-"):
  1. standard_element(标准化元素,实测效果最好)
  2. demand_content_topic(选题,如果非 "-")
  3. title(承接视频标题,兜底)
"""

import logging
import os
from dataclasses import dataclass, field
from typing import List, Optional

import httpx

from tools.video_recall import LandingVideo

logger = logging.getLogger(__name__)

VECTOR_BASE = os.getenv(
    "VECTOR_BASE_URL",
    "https://api-internal.piaoquantv.com/videoVector",
)
VECTOR_MATCH_BY_TEXT = f"{VECTOR_BASE}/recallTest/matchByText"
VECTOR_BATCH_BY_TEXT = f"{VECTOR_BASE}/recallTest/batchByText"
VECTOR_ALL_CONFIG_CODES = f"{VECTOR_BASE}/videoSearch/getAllConfigCodes"


def fetch_all_config_codes() -> dict[str, str]:
    """动态获取 vector 服务支持的全部 configCode → 中文名。

    返回 {configCode: 中文名}。
    召回前先调一次,避免硬编码列表过时。
    """
    resp = httpx.get(VECTOR_ALL_CONFIG_CODES, timeout=15)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") not in (0, 200):
        raise RuntimeError(
            f"getAllConfigCodes 失败:code={data.get('code')} msg={data.get('msg')}"
        )
    return data.get("data") or {}

# 字段维度策略(2026-06-08 用户最终确认)
# - 维度 1 选题:       queryText=demand_content_topic, configCode=VIDEO_TOPIC
# - 维度 2 标准化元素: queryText=standard_element, configCode 按 point_type **唯一对应**:
#     - "灵感点" → INSPIRATION_SUBSTANCE
#     - "关键点" → KEYPOINT_SUBSTANCE
#     - "目的点" → PURPOSE_SUBSTANCE
#     - 其他/空    → 跳过(没办法确定走哪一路,no-guessing)
# - 维度 3 标题:       不做(VIDEO_TITLE 对 MATERIAL 实测返回 0)
POINT_TYPE_TO_SUBSTANCE = {
    "灵感点": "INSPIRATION_SUBSTANCE",
    "关键点": "KEYPOINT_SUBSTANCE",
    "目的点": "PURPOSE_SUBSTANCE",
}
POINT_TYPE_TO_VIDEO_CONFIG = {
    "灵感点": "VIDEO_INSPIRATION",
    "关键点": "VIDEO_KEYPOINT",
    "目的点": "VIDEO_PURPOSE",
}

# fallback(动态接口失败时用)
MATERIAL_EFFECTIVE_CONFIG_CODES = ["VIDEO_TOPIC"] + list(POINT_TYPE_TO_SUBSTANCE.values())

# 占位符值视为无效(piaoquantv 数据里 demandContentTopic 大量为 "-")
PLACEHOLDER_VALUES = {"", "-", None}

# 单 configCode 召回 top N(合并前)
DEFAULT_PER_CC_TOP_N = 100
# 最终合并去重后返回 top N
DEFAULT_FINAL_TOP_N = 20

# Cover URL 黑名单 — 已知尺寸/比例不符合腾讯创意要求的来源
# (2026-06-09 用户决策:踩坑驱动渐进收紧,踩到新模式再加)
#   - "auto_reply_cards_cover" :自动回复卡片业务的 cover,实测 reject code 1801159
EXCLUDED_COVER_URL_PATTERNS = (
    "auto_reply_cards_cover",
)


@dataclass
class Material:
    """召回的创意素材。

    2026-06-10 升级:从 batchByText 接口接收质量数据(cost/ctr/cvr/roi/impressions 等)。
    """

    material_id: str           # 素材原始 ID
    score: float               # 服务端综合分(wCtr=1 配置下 ≈ qualityScore,与 CTR 强相关)
    title: str = ""
    cover: str = ""
    video_url: str = ""
    # 投放质量数据(来自 materialDetail.quality)— 2026-06-10 升级新增
    cost: Optional[float] = None              # 累计成本(元)
    ctr: Optional[float] = None               # 点击率
    cvr: Optional[float] = None               # 转化率
    roi: Optional[float] = None               # ROI(收入/成本)
    impressions: Optional[int] = None         # 累计曝光
    quality_score: Optional[float] = None     # 服务端 qualityScore
    # 原始 item dict(以备后用)
    raw: dict = field(default_factory=dict, repr=False)


def _call_match_by_text(
    query_text: str,
    config_code: str,
    material_top_n: int = DEFAULT_PER_CC_TOP_N,
    source_labels: Optional[List[str]] = None,
    timeout: int = 30,
) -> List[dict]:
    """单次调 matchByText,只要 MATERIAL 模态,返回原始 items 列表。"""
    body = {
        "queryText": query_text,
        "configCode": config_code,
        "modalities": ["MATERIAL"],
        "videoTopN": 0,
        "articleTopN": 0,
        "materialTopN": material_top_n,
        "topN": material_top_n,
        "displayK": material_top_n,
    }
    if source_labels:
        body["sourceLabels"] = source_labels

    logger.info(
        "[material_recall] matchByText q=%r configCode=%s topN=%d",
        query_text[:40], config_code, material_top_n,
    )
    resp = httpx.post(
        VECTOR_MATCH_BY_TEXT,
        json=body,
        headers={"content-type": "application/json", "accept": "application/json"},
        timeout=timeout,
    )
    resp.raise_for_status()
    data = resp.json()

    code = data.get("code")
    # CommonResponse 可能用 0 或 200 表示成功;以 data 字段为准
    if code not in (0, 200, "0", "200"):
        raise RuntimeError(
            f"vector matchByText 失败:code={code} msg={data.get('msg') or data.get('message')}"
        )

    payload = data.get("data") or {}
    items = payload.get("items") or []
    # 只留 MATERIAL 模态(防御性 — 万一返回混入)
    return [it for it in items if it.get("modality") == "MATERIAL"]


def _pick_query_text(landing: LandingVideo) -> Optional[str]:
    """按优先级选 queryText:standard_element > demand_content_topic > title。
    占位符 "-" / 空串视为无效。"""
    for cand in (landing.standard_element, landing.demand_content_topic, landing.title):
        if cand and cand not in PLACEHOLDER_VALUES:
            return cand
    return None


def _pick_query_text_for_batch(landing: LandingVideo) -> Optional[str]:
    """选 queryText:standard_element > demand_content_topic > title。(保留兼容)"""
    for cand in (landing.standard_element, landing.demand_content_topic, landing.title):
        if cand and cand not in PLACEHOLDER_VALUES:
            return cand
    return None


def _pick_query_strategies_for_batch(landing: LandingVideo) -> List[tuple]:
    """返回 (queryText, configCode, strategy_name) 策略候选(2026-06-10 用户最终确认)。

    只用 2 个维度:
      维度 1 - 标准化元素:queryText=standard_element, configCode 按 point_type 唯一对应
        - "灵感点" → INSPIRATION_SUBSTANCE
        - "关键点" → KEYPOINT_SUBSTANCE
        - "目的点" → PURPOSE_SUBSTANCE
      维度 2 - 选题:queryText=demand_content_topic, configCode=VIDEO_TOPIC

    prepare 阶段:维度 1 失败 → 试维度 2 → 仍失败 → 换 landing。
    """
    strategies = []
    # 维度 1: 标准化元素(优先)
    if landing.standard_element and landing.standard_element not in PLACEHOLDER_VALUES:
        cc = POINT_TYPE_TO_SUBSTANCE.get(landing.point_type)
        if cc:
            strategies.append((landing.standard_element, cc, f"标准化元素-{landing.point_type}"))
    # 维度 2: 选题
    if landing.demand_content_topic and landing.demand_content_topic not in PLACEHOLDER_VALUES:
        strategies.append((landing.demand_content_topic, "VIDEO_TOPIC", "选题"))
    return strategies


def _fallback_config_codes_for_strategy(config_code: str, landing: LandingVideo) -> List[str]:
    """batchByText 无结果时给 matchByText 的兼容 configCode。

    2026-06-30 实测:batchByText 对 MATERIAL 返回 0 时,旧 matchByText 仍可在
    VIDEO_KEYPOINT/VIDEO_INSPIRATION 等历史向量字段命中素材。
    """
    out = [config_code]
    video_cc = POINT_TYPE_TO_VIDEO_CONFIG.get(landing.point_type)
    if video_cc and video_cc not in out:
        out.append(video_cc)
    if config_code != "VIDEO_TOPIC" and "VIDEO_TOPIC" not in out:
        out.append("VIDEO_TOPIC")
    return out


def _call_batch_by_text(
    query_text: str,
    config_codes: List[str],
    display_k: int,
    days: int,
    sim_threshold: float,
    alpha: float,
    w_ctr: float, w_cvr: float, w_roi: float,
    w_open_rate: float, w_fission_rate: float,
    deconstruct_boost: float,
    source_labels: List[str],
    modalities: List[str] = None,
    timeout: int = 30,
) -> List[dict]:
    """调用 batchByText 接口(2026-06-10 升级).
    服务端单次 embedding + 多 configCode 并行 ANN + 跨模态过滤 + ranking 加权 + 去重。
    """
    body = {
        "queryText": query_text,
        "configCodes": config_codes,
        "displayK": display_k,
        "modalities": modalities or ["MATERIAL"],
        "sourceLabels": source_labels,
        "days": days,
        "ranking": {
            "simThreshold": sim_threshold,
            "alpha": alpha,
            "wCtr": w_ctr, "wCvr": w_cvr, "wRoi": w_roi,
            "wOpenRate": w_open_rate, "wFissionRate": w_fission_rate,
            "deconstructBoost": deconstruct_boost,
        },
    }
    logger.info(
        "[material_recall] batchByText q=%r configCodes=%d displayK=%d simT=%.2f wCtr=%.2f",
        query_text[:40], len(config_codes), display_k, sim_threshold, w_ctr,
    )
    resp = httpx.post(
        VECTOR_BATCH_BY_TEXT,
        json=body,
        headers={"content-type": "application/json", "accept": "application/json"},
        timeout=timeout,
    )
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") not in (0, 200, "0", "200"):
        raise RuntimeError(
            f"batchByText 失败:code={data.get('code')} msg={data.get('msg') or data.get('message')}"
        )
    payload = data.get("data") or {}
    return payload.get("items") or []


def _items_to_materials(items: List[dict], min_impressions: int, min_ctr: float) -> tuple:
    """把 batchByText items 过滤 + 转 Material(2026-06-10 复用 helper)。

    过滤条件(全部满足才保留):
    - modality=MATERIAL(防御性)
    - cover URL 不在黑名单
    - impressions > min_impressions
    - ctr >= min_ctr(2026-06-10 加,5% 太低)

    返回 (materials, blacklist_n, low_imp_n, low_ctr_n)。
    """
    out: List[Material] = []
    excluded_blacklist = 0
    excluded_low_imp = 0
    excluded_low_ctr = 0
    for it in items:
        if it.get("modality") != "MATERIAL":
            continue
        mid = it.get("materialId") or (str(it["id"]) if it.get("id") is not None else None)
        if not mid:
            continue
        cover = it.get("cover") or ""
        if any(p in cover for p in EXCLUDED_COVER_URL_PATTERNS):
            excluded_blacklist += 1
            continue
        md = it.get("materialDetail") or {}
        q = md.get("quality") or {}
        imp = q.get("impressions") or 0
        if imp <= min_impressions:
            excluded_low_imp += 1
            continue
        ctr = q.get("ctr") or 0.0
        if ctr < min_ctr:
            excluded_low_ctr += 1
            continue
        out.append(Material(
            material_id=str(mid),
            score=float(it.get("score") or 0.0),
            title=it.get("title") or "",
            cover=cover,
            video_url=it.get("videoUrl") or "",
            cost=q.get("cost"),
            ctr=q.get("ctr"),
            cvr=q.get("cvr"),
            roi=q.get("roi"),
            impressions=imp,
            quality_score=q.get("qualityScore"),
            raw=it,
        ))
    return out, excluded_blacklist, excluded_low_imp, excluded_low_ctr


def recall_materials_for_video(
    landing: LandingVideo,
    final_top_n: int = DEFAULT_FINAL_TOP_N,
    source_labels: Optional[List[str]] = None,
) -> List[Material]:
    """素材召回(2026-06-10 用户最终:2 维度 strategy + fallback,batchByText 单 configCode)。

    流程:
      1. _pick_query_strategies_for_batch(landing) 返回 1~2 个策略候选
         维度 1: (standard_element, *_SUBSTANCE 按 point_type) — 优先
         维度 2: (demand_content_topic, VIDEO_TOPIC)
      2. 对每个策略调一次 batchByText(只传 1 个 configCode)
         首次返回 ≥ 1 条 impressions>阈值 的 material → break,直接返回
      3. 全部策略都失败 → 返回空 → 上层 prepare 换 landing

    阈值不变保证质量(用户 2026-06-10 确认):依赖"多次尝试"提升命中率。
    """
    from config import (
        RECALL_ALPHA, RECALL_DAYS, RECALL_DECONSTRUCT_BOOST,
        RECALL_DISPLAY_K, RECALL_MIN_CTR, RECALL_MIN_IMPRESSIONS,
        RECALL_SIM_THRESHOLD, RECALL_SOURCE_LABELS,
        RECALL_W_CTR, RECALL_W_CVR, RECALL_W_FISSION_RATE,
        RECALL_W_OPEN_RATE, RECALL_W_ROI,
    )

    strategies = _pick_query_strategies_for_batch(landing)
    if not strategies:
        logger.warning(
            "[material_recall] landing video_id=%d 无可用策略(2 维度都缺),返回空",
            landing.video_id,
        )
        return []

    logger.info(
        "[material_recall] landing video_id=%d 走 %d 个策略",
        landing.video_id, len(strategies),
    )

    for query_text, config_code, name in strategies:
        logger.info(
            "[material_recall]   策略=%s q=%r configCode=%s",
            name, query_text[:30], config_code,
        )
        try:
            items = _call_batch_by_text(
                query_text=query_text,
                config_codes=[config_code],          # ← 只传当前策略的 1 个
                display_k=RECALL_DISPLAY_K,
                days=RECALL_DAYS,
                sim_threshold=RECALL_SIM_THRESHOLD,
                alpha=RECALL_ALPHA,
                w_ctr=RECALL_W_CTR, w_cvr=RECALL_W_CVR, w_roi=RECALL_W_ROI,
                w_open_rate=RECALL_W_OPEN_RATE, w_fission_rate=RECALL_W_FISSION_RATE,
                deconstruct_boost=RECALL_DECONSTRUCT_BOOST,
                source_labels=source_labels or RECALL_SOURCE_LABELS,
                modalities=["MATERIAL"],
            )
        except Exception as e:
            logger.error(
                "[material_recall] 策略 %s 失败,试下一个:%s", name, e,
            )
            continue

        mats, n_bl, n_low_imp, n_low_ctr = _items_to_materials(
            items, RECALL_MIN_IMPRESSIONS, RECALL_MIN_CTR,
        )
        logger.info(
            "[material_recall]     服务端返回 %d 条,⊘ 黑名单 %d,⊘ imp<=%d %d,⊘ ctr<%.2f %d → 保留 %d",
            len(items), n_bl,
            RECALL_MIN_IMPRESSIONS, n_low_imp,
            RECALL_MIN_CTR, n_low_ctr,
            len(mats),
        )
        if mats:
            return mats[:final_top_n]

        # batchByText 当前可能对 MATERIAL 返回 0;降级到历史 matchByText 路径。
        for fallback_cc in _fallback_config_codes_for_strategy(config_code, landing):
            try:
                fallback_items = _call_match_by_text(
                    query_text=query_text,
                    config_code=fallback_cc,
                    material_top_n=RECALL_DISPLAY_K,
                    source_labels=source_labels or RECALL_SOURCE_LABELS,
                )
            except Exception as e:
                logger.error(
                    "[material_recall] fallback matchByText %s 失败,试下一个:%s",
                    fallback_cc, e,
                )
                continue

            fmats, fn_bl, fn_low_imp, fn_low_ctr = _items_to_materials(
                fallback_items, RECALL_MIN_IMPRESSIONS, RECALL_MIN_CTR,
            )
            logger.info(
                "[material_recall]     fallback=%s 返回 %d 条,⊘ 黑名单 %d,⊘ imp<=%d %d,⊘ ctr<%.2f %d → 保留 %d",
                fallback_cc, len(fallback_items), fn_bl,
                RECALL_MIN_IMPRESSIONS, fn_low_imp,
                RECALL_MIN_CTR, fn_low_ctr,
                len(fmats),
            )
            if fmats:
                fmats.sort(
                    key=lambda m: (
                        m.ctr or 0,
                        m.impressions or 0,
                        m.quality_score or 0,
                        m.score or 0,
                    ),
                    reverse=True,
                )
                return fmats[:final_top_n]
        # else 继续下一个策略

    logger.info(
        "[material_recall] landing video_id=%d 所有 %d 策略全失败,返回空(上层换 landing)",
        landing.video_id, len(strategies),
    )
    return []