"""创意素材召回接口适配层(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 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 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", } 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: """One ODPS-derived material recall query.""" 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, ) 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 = [] 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, ) 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 _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, ) -> tuple: """把召回 items 过滤 + 转 Material。 过滤条件: - modality=MATERIAL(防御性) - cover URL 不在黑名单 - score >= sim_threshold CTR / impressions 只作为审批展示和兜底排序参考,不再作为硬筛。 返回 (materials, stats)。 """ out: List[Material] = [] stats = { "blacklist": 0, "low_score": 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 {} out.append(Material( material_id=str(mid), score=score, title=it.get("title") or "", cover=cover, video_url=it.get("videoUrl") or "", cost=_as_float(q.get("cost")) if q.get("cost") is not None else None, 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]: """Build recall queries from ODPS features. Supported dimensions: - 解构选题 -> 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, ) -> List[Material]: """素材召回:用 ODPS 多维特征并行召回并合并排序。 流程: 1. 从 ODPS features 生成 query:解构选题 + 实质三点。 2. 多 query 并行调用 batchByText。 3. 汇总、material_id 去重、score>=阈值、按 cost 倒序。 当前硬筛只保留相似度阈值;曝光/CTR 进入审批表但不拦截。 """ from config import ( RECALL_ALPHA, RECALL_DAYS, RECALL_DECONSTRUCT_BOOST, RECALL_DISPLAY_K, 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 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, ) query_materials.append((query, mats)) logger.info( "[material_recall] 策略=%s q=%r configCode=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d → 保留 %d", query.strategy_name, query.query_text[:30], query.config_code, len(items), stats["blacklist"], RECALL_SIM_THRESHOLD, stats["low_score"], 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, ) 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 → 保留 %d(cost desc)", fallback_cc, len(fallback_items), fstats["blacklist"], RECALL_SIM_THRESHOLD, fstats["low_score"], 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 []