Agent/examples/auto_put_ad_mini/tools/material_recall.py

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

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

接口:POST https://tp-open.piaoquantv.com/contentPlatform/videoVector/recallTest/matchByText
鉴权:复用 contentPlatform 动态登录 token

本质(用户 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 concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Iterable, List, Optional

import httpx

from tools.video_recall import (
    LandingVideo,
    _build_auth_headers,
    _is_not_login_error,
    get_piaoquantv_token,
    refresh_piaoquantv_token,
)

logger = logging.getLogger(__name__)

VECTOR_BASE = os.getenv(
    "VECTOR_BASE_URL",
    "https://tp-open.piaoquantv.com/contentPlatform/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 _request_vector(
    method: str,
    url: str,
    *,
    json_body: Optional[dict] = None,
    timeout: int = 30,
) -> dict:
    """调用 growth-manager vector 转发接口，token 失效时刷新并重试一次。"""
    token = get_piaoquantv_token()

    def _do_request(current_token: str):
        return httpx.request(
            method,
            url,
            json=json_body,
            headers=_build_auth_headers(current_token),
            timeout=timeout,
        )

    resp = _do_request(token)
    if resp.status_code == 401:
        logger.warning("[material_recall] vector 转发接口返回 401，刷新 token 并重试")
        token = refresh_piaoquantv_token()
        resp = _do_request(token)

    resp.raise_for_status()
    data = resp.json()
    if _is_not_login_error(data):
        logger.warning("[material_recall] vector 转发接口返回 Not_LOGIN(1000)，刷新 token 并重试")
        token = refresh_piaoquantv_token()
        resp = _do_request(token)
        resp.raise_for_status()
        data = resp.json()
    return data


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

    返回 {configCode: 中文名}。
    召回前先调一次,避免硬编码列表过时。
    """
    data = _request_vector("GET", VECTOR_ALL_CONFIG_CODES, timeout=15)
    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",
}
SUBSTANCE_TO_VIDEO_CONFIG = {
    "INSPIRATION_SUBSTANCE": "VIDEO_INSPIRATION",
    "KEYPOINT_SUBSTANCE": "VIDEO_KEYPOINT",
    "PURPOSE_SUBSTANCE": "VIDEO_PURPOSE",
}
ODPS_FEATURE_TO_CONFIG = {
    ("解构选题", ""): "VIDEO_TOPIC",
    ("实质", "灵感点"): "INSPIRATION_SUBSTANCE",
    ("实质", "关键点"): "KEYPOINT_SUBSTANCE",
    ("实质", "目的点"): "PURPOSE_SUBSTANCE",
}

# 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 RecallQuery:
    """一条由 ODPS 特征生成的素材召回查询。"""

    query_text: str
    config_code: str
    element_dimension: str
    point_type: str
    standard_element: str
    contribution_score: float = 0.0
    dt: str = ""

    @property
    def strategy_name(self) -> str:
        """返回召回策略名称（解构选题 或 元素维度-点类型）。"""
        if self.element_dimension == "解构选题":
            return "解构选题"
        return f"{self.element_dimension}-{self.point_type}"

    def to_hit(self) -> dict:
        """把召回 query 转成命中记录 dict（写入素材的 recall_hit_queries）。"""
        return {
            "strategy": self.strategy_name,
            "element_dimension": self.element_dimension,
            "point_type": self.point_type,
            "standard_element": self.standard_element,
            "query_text": self.query_text,
            "config_code": self.config_code,
            "contribution_score": self.contribution_score,
            "dt": self.dt,
        }


@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
    visit_uv_30d: Optional[int] = None        # 外部素材配置窗口内访问UV之和（列名兼容历史）
    recall_strategy: str = ""
    recall_query_text: str = ""
    recall_config_code: str = ""
    recall_element_dimension: str = ""
    recall_point_type: str = ""
    recall_standard_element: str = ""
    recall_hit_queries: List[dict] = field(default_factory=list)
    # 原始 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,
    )
    data = _request_vector(
        "POST",
        VECTOR_MATCH_BY_TEXT,
        json_body=body,
        timeout=timeout,
    )

    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 = []
    seen = set()
    for feature in landing.raw.get("element_features") or []:
        if not isinstance(feature, dict):
            continue
        standard_element = (feature.get("standard_element") or "").strip()
        point_type = (feature.get("point_type") or "").strip()
        if standard_element in PLACEHOLDER_VALUES:
            continue
        cc = POINT_TYPE_TO_SUBSTANCE.get(point_type)
        if not cc:
            continue
        key = (standard_element, cc)
        if key in seen:
            continue
        seen.add(key)
        strategies.append((standard_element, cc, f"标准化元素-{point_type}"))

    # 维度 1: 标准化元素(优先)
    if landing.standard_element and landing.standard_element not in PLACEHOLDER_VALUES:
        cc = POINT_TYPE_TO_SUBSTANCE.get(landing.point_type)
        if cc and (landing.standard_element, cc) not in seen:
            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 = SUBSTANCE_TO_VIDEO_CONFIG.get(config_code) or 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,
    )
    data = _request_vector(
        "POST",
        VECTOR_BATCH_BY_TEXT,
        json_body=body,
        timeout=timeout,
    )
    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 _as_float(value, default: float = 0.0) -> float:
    """安全转 float,None 或非法值返回 default。"""
    try:
        if value is None:
            return default
        return float(value)
    except (TypeError, ValueError):
        return default


def _as_int(value, default: int = 0) -> int:
    """安全转 int,None 或非法值返回 default。"""
    try:
        if value is None:
            return default
        return int(float(value))
    except (TypeError, ValueError):
        return default


def _items_to_materials(
    items: List[dict],
    sim_threshold: float,
    *,
    apply_cover_blacklist: bool = True,
    min_cost: Optional[float] = None,
    apply_cost_filter: bool = True,
) -> tuple:
    """把召回 items 过滤 + 转 Material。

    过滤条件:
    - modality=MATERIAL(防御性)
    - cover URL 不在黑名单
    - score >= sim_threshold
    - apply_cost_filter 且 min_cost is not None 时:历史成本(cost, 元) >= min_cost
      (严格,无成本数据也排除)。外部合作素材召回关闭 apply_cost_filter,
      仅以访问 UV 门槛把关。

    CTR / impressions 只作为审批展示和兜底排序参考,不再作为硬筛。
    返回 (materials, stats)。
    """
    out: List[Material] = []
    stats = {
        "blacklist": 0,
        "low_score": 0,
        "low_cost": 0,
        "low_imp": 0,
        "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 apply_cover_blacklist and any(
            pattern in cover for pattern in EXCLUDED_COVER_URL_PATTERNS
        ):
            stats["blacklist"] += 1
            continue
        score = _as_float(it.get("score"))
        if score < sim_threshold:
            stats["low_score"] += 1
            continue
        md = it.get("materialDetail") or {}
        q = md.get("quality") or {}
        raw_cost = q.get("cost")
        cost = _as_float(raw_cost) if raw_cost is not None else None
        if apply_cost_filter and min_cost is not None and (cost is None or cost < min_cost):
            stats["low_cost"] += 1
            continue
        out.append(Material(
            material_id=str(mid),
            score=score,
            title=it.get("title") or "",
            cover=cover,
            video_url=it.get("videoUrl") or "",
            cost=cost,
            ctr=_as_float(q.get("ctr")) if q.get("ctr") is not None else None,
            cvr=_as_float(q.get("cvr")) if q.get("cvr") is not None else None,
            roi=_as_float(q.get("roi")) if q.get("roi") is not None else None,
            impressions=_as_int(q.get("impressions")) if q.get("impressions") is not None else None,
            quality_score=_as_float(q.get("qualityScore")) if q.get("qualityScore") is not None else None,
            raw=it,
        ))
    return out, stats


def _sort_materials_by_policy(materials: List[Material]) -> List[Material]:
    """生产排序策略:先相关性准入,再按历史消耗倒序。"""
    return sorted(
        materials,
        key=lambda m: (
            m.cost is not None,
            m.cost or 0,
            m.roi or 0,
            m.impressions or 0,
            m.ctr or 0,
            m.quality_score or 0,
            m.score or 0,
        ),
        reverse=True,
    )


def _material_rank(material: Material) -> tuple:
    """生成素材排序键（与 _sort_materials_by_policy 同序:消耗优先）。"""
    return (
        material.cost is not None,
        material.cost or 0,
        material.roi or 0,
        material.impressions or 0,
        material.ctr or 0,
        material.quality_score or 0,
        material.score or 0,
    )


def _feature_attr(feature, name: str, default=""):
    """兼容 dict 和对象两种 feature,统一取属性值。"""
    if isinstance(feature, dict):
        return feature.get(name, default)
    return getattr(feature, name, default)


def _build_recall_queries_from_features(
    element_features: Iterable,
    query_limit: int,
) -> List[RecallQuery]:
    """根据 ODPS 特征构建召回查询。

    支持的维度：解构选题映射到 VIDEO_TOPIC；实质加灵感点、关键点或目的点映射
    到对应的 *_SUBSTANCE。
    """
    queries: List[RecallQuery] = []
    seen = set()
    raw_features = list(element_features or [])
    sorted_features = sorted(
        raw_features,
        key=lambda f: (
            0 if str(_feature_attr(f, "element_dimension") or "") == "解构选题" else 1,
            -float(_feature_attr(f, "contribution_score", 0) or 0),
        ),
    )
    for feature in sorted_features:
        element_dimension = str(_feature_attr(feature, "element_dimension") or "").strip()
        point_type = str(_feature_attr(feature, "point_type") or "").strip()
        standard_element = str(_feature_attr(feature, "standard_element") or "").strip()
        if standard_element in PLACEHOLDER_VALUES:
            continue
        config_code = ODPS_FEATURE_TO_CONFIG.get((element_dimension, point_type))
        if not config_code and element_dimension == "解构选题":
            config_code = ODPS_FEATURE_TO_CONFIG.get(("解构选题", ""))
        if not config_code:
            continue
        key = (standard_element, config_code)
        if key in seen:
            continue
        seen.add(key)
        queries.append(RecallQuery(
            query_text=standard_element,
            config_code=config_code,
            element_dimension=element_dimension,
            point_type=point_type,
            standard_element=standard_element,
            contribution_score=float(_feature_attr(feature, "contribution_score", 0) or 0),
            dt=str(_feature_attr(feature, "dt") or ""),
        ))
        if len(queries) >= query_limit:
            break
    return queries


def _call_batch_for_recall_query(
    query: RecallQuery,
    *,
    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],
) -> tuple[RecallQuery, List[dict]]:
    """对单个召回 query 调 batchByText,返回 (query, 原始 items)。供并行调用。"""
    items = _call_batch_by_text(
        query_text=query.query_text,
        config_codes=[query.config_code],
        display_k=display_k,
        days=days,
        sim_threshold=sim_threshold,
        alpha=alpha,
        w_ctr=w_ctr,
        w_cvr=w_cvr,
        w_roi=w_roi,
        w_open_rate=w_open_rate,
        w_fission_rate=w_fission_rate,
        deconstruct_boost=deconstruct_boost,
        source_labels=source_labels,
        modalities=["MATERIAL"],
    )
    return query, items


def _merge_materials_by_policy(query_materials: List[tuple[RecallQuery, List[Material]]]) -> List[Material]:
    """多路召回结果按 material_id 去重合并(保留排序更优者),并记录命中 query,最后按策略排序。"""
    by_mid: dict[str, Material] = {}
    for query, materials in query_materials:
        hit = query.to_hit()
        for material in materials:
            material.recall_hit_queries = [hit]
            material.recall_strategy = query.strategy_name
            material.recall_query_text = query.query_text
            material.recall_config_code = query.config_code
            material.recall_element_dimension = query.element_dimension
            material.recall_point_type = query.point_type
            material.recall_standard_element = query.standard_element
            existing = by_mid.get(material.material_id)
            if existing is None:
                by_mid[material.material_id] = material
                continue
            merged_hits = existing.recall_hit_queries + [
                h for h in material.recall_hit_queries
                if h not in existing.recall_hit_queries
            ]
            if _material_rank(material) > _material_rank(existing):
                material.recall_hit_queries = merged_hits
                by_mid[material.material_id] = material
            else:
                existing.recall_hit_queries = merged_hits
    return _sort_materials_by_policy(list(by_mid.values()))


def recall_materials_for_video(
    landing: LandingVideo,
    final_top_n: Optional[int] = DEFAULT_FINAL_TOP_N,
    source_labels: Optional[List[str]] = None,
    element_features: Optional[Iterable] = None,
    apply_cover_blacklist: bool = True,
    min_cost: Optional[float] = None,
    apply_cost_filter: bool = True,
) -> List[Material]:
    """素材召回:用 ODPS 多维特征并行召回并合并排序。

    流程:
      1. 从 ODPS features 生成 query:解构选题 + 实质三点。
      2. 多 query 并行调用 batchByText。
      3. 汇总、material_id 去重、score>=阈值、按 cost 倒序。

    硬筛:相似度阈值 + 历史成本(cost>=min_cost,min_cost 默认取 config.RECALL_MIN_COST_YUAN)。
    外部合作素材召回传 apply_cost_filter=False 关闭成本门槛,仅以访问 UV 门槛把关。
    曝光/CTR 进入审批表但不拦截。
    """
    from config import (
        RECALL_ALPHA, RECALL_DAYS, RECALL_DECONSTRUCT_BOOST,
        RECALL_DISPLAY_K, RECALL_MIN_COST_YUAN, RECALL_PARALLEL_MAX_WORKERS,
        RECALL_QUERY_LIMIT_PER_VIDEO,
        RECALL_SIM_THRESHOLD, RECALL_SOURCE_LABELS,
        RECALL_W_CTR, RECALL_W_CVR, RECALL_W_FISSION_RATE,
        RECALL_W_OPEN_RATE, RECALL_W_ROI,
    )

    if min_cost is None:
        min_cost = RECALL_MIN_COST_YUAN
    cost_marker = "off" if not apply_cost_filter else f"{min_cost:.0f}"

    if element_features is None:
        element_features = landing.raw.get("element_features") or []
    queries = _build_recall_queries_from_features(
        element_features,
        query_limit=max(1, RECALL_QUERY_LIMIT_PER_VIDEO),
    )
    if not queries:
        logger.warning(
            "[material_recall] landing video_id=%d 无 ODPS 可用召回特征,返回空",
            landing.video_id,
        )
        return []

    logger.info(
        "[material_recall] landing video_id=%d 走 %d 个 ODPS 策略:%s",
        landing.video_id, len(queries),
        "; ".join(f"{q.strategy_name}:{q.query_text}->{q.config_code}" for q in queries),
    )

    query_materials: List[tuple[RecallQuery, List[Material]]] = []
    labels = source_labels or RECALL_SOURCE_LABELS
    max_workers = max(1, min(RECALL_PARALLEL_MAX_WORKERS, len(queries)))
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(
                _call_batch_for_recall_query,
                query,
                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=labels,
            )
            for query in queries
        ]
        for future in as_completed(futures):
            try:
                query, items = future.result()
            except Exception as e:
                logger.error("[material_recall] 并行 batchByText 失败:%s", e)
                continue
            mats, stats = _items_to_materials(
                items,
                RECALL_SIM_THRESHOLD,
                apply_cover_blacklist=apply_cover_blacklist,
                min_cost=min_cost,
                apply_cost_filter=apply_cost_filter,
            )
            query_materials.append((query, mats))
            logger.info(
                "[material_recall]   策略=%s q=%r configCode=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d,⊘ cost<%s %d → 保留 %d",
                query.strategy_name, query.query_text[:30], query.config_code,
                len(items), stats["blacklist"], RECALL_SIM_THRESHOLD,
                stats["low_score"], cost_marker, stats["low_cost"],
                len(mats),
            )

    merged = _merge_materials_by_policy(query_materials)
    logger.info(
        "[material_recall] landing video_id=%d 多维召回合并后保留 %d 条(cost desc)",
        landing.video_id, len(merged),
    )
    if merged:
        return merged if final_top_n is None else merged[:final_top_n]

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

            fmats, fstats = _items_to_materials(
                fallback_items,
                RECALL_SIM_THRESHOLD,
                apply_cover_blacklist=apply_cover_blacklist,
                min_cost=min_cost,
                apply_cost_filter=apply_cost_filter,
            )
            fquery = RecallQuery(
                query_text=query.query_text,
                config_code=fallback_cc,
                element_dimension=query.element_dimension,
                point_type=query.point_type,
                standard_element=query.standard_element,
                contribution_score=query.contribution_score,
                dt=query.dt,
            )
            fmats = _merge_materials_by_policy([(fquery, fmats)])
            logger.info(
                "[material_recall]     fallback=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d,⊘ cost<%s %d → 保留 %d(cost desc)",
                fallback_cc, len(fallback_items), fstats["blacklist"],
                RECALL_SIM_THRESHOLD, fstats["low_score"],
                cost_marker, fstats["low_cost"],
                len(fmats),
            )
            if fmats:
                return fmats if final_top_n is None else fmats[:final_top_n]

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