支持英文版本的 show_desc

app/domains/recommend/demand_recommend/__init__.py

@@ -0,0 +1,20 @@
+from ._const import DemandRecommendConst, MatchMethod, ConfigCode, DemandSource
+from ._utils import DemandRecord, MatchStrategy, MatchResult, DemandStrategyParser, parse_recall_items
+from ._mapper import DemandRecommendMapper
+from .match_engine import DemandVideoMatchEngine
+from .entrance import DemandRecommendTask
+
+__all__ = [
+    "DemandRecommendConst",
+    "MatchMethod",
+    "ConfigCode",
+    "DemandSource",
+    "DemandRecord",
+    "MatchStrategy",
+    "MatchResult",
+    "DemandStrategyParser",
+    "DemandRecommendMapper",
+    "DemandVideoMatchEngine",
+    "DemandRecommendTask",
+    "parse_recall_items",
+]

app/domains/recommend/demand_recommend/_const.py

@@ -0,0 +1,65 @@
+from enum import StrEnum
+
+
+class MatchMethod(StrEnum):
+    """匹配方式，由 demand 的 match_video_rule 字段推导"""
+    TEXT = "text"               # 文本检索 → /recallTest/matchByText
+    VIDEO_ID = "video_id"       # 以视频搜视频 → /recallTest/matchByVideoId
+    CONTENT_ID = "content_id"   # 以内容搜视频 → /videoSearch/matchTopNVideo
+
+
+class ConfigCode(StrEnum):
+    """向量匹配维度"""
+    VIDEO_TOPIC = "VIDEO_TOPIC"
+    VIDEO_INSPIRATION = "VIDEO_INSPIRATION"
+    VIDEO_KEYPOINT = "VIDEO_KEYPOINT"
+    VIDEO_PURPOSE = "VIDEO_PURPOSE"
+
+
+class DemandSource(StrEnum):
+    """需求来源类型"""
+    PRIORI = "先验需求"          # 只有文本描述
+    POSTERIOR = "后验需求"       # 有后验视频ID
+    SCENE = "场景需求"           # 有场景内容ID
+
+
+class DemandRecommendConst:
+    """需求-视频匹配常量"""
+
+    # ── 策略关键词 → configCode 映射 ──
+    STRATEGY_CONFIG_MAP = {
+        "灵感": ConfigCode.VIDEO_INSPIRATION,
+        "创意": ConfigCode.VIDEO_INSPIRATION,
+        "关键": ConfigCode.VIDEO_KEYPOINT,
+        "要点": ConfigCode.VIDEO_KEYPOINT,
+        "目的": ConfigCode.VIDEO_PURPOSE,
+        "转化": ConfigCode.VIDEO_PURPOSE,
+    }
+    DEFAULT_CONFIG_CODE = ConfigCode.VIDEO_TOPIC
+
+    # ── match_video_rule 关键词 → match_method 映射 ──
+    RULE_METHOD_MAP = {
+        "场景已看视频": MatchMethod.CONTENT_ID,
+        "票圈推荐库-关键词": MatchMethod.TEXT,
+        "票圈推荐库-向量": MatchMethod.VIDEO_ID,
+    }
+
+    # ── 默认值 ──
+    DEFAULT_TOPN = 10
+    MAX_WORKERS = 5          # 并发匹配 worker 数
+    API_TIMEOUT = 30         # 匹配 API 超时(秒)
+
+    # ── API 路径 ──
+    BASE_URL = "http://api-internal.piaoquantv.com/videoVector"
+
+    class ApiPath:
+        MATCH_BY_TEXT = "/recallTest/matchByText"
+        MATCH_BY_VIDEO_ID = "/recallTest/matchByVideoId"
+        MATCH_TOP_N_VIDEO = "/videoSearch/matchTopNVideo"
+        DECONSTRUCT = "/videoSearch/deconstruct"
+        GET_DECONSTRUCT_RESULT = "/videoSearch/getDeconstructResult"
+        GET_ALL_CONFIG_CODES = "/videoSearch/getAllConfigCodes"
+
+    # ── 匹配结果表 ──
+    MATCH_RESULT_TABLE = "demand_video_match"
+    MATCH_RESULT_DB = "long_articles"

app/domains/recommend/demand_recommend/_mapper.py

@@ -0,0 +1,120 @@
+import json
+from typing import Any, Dict, List
+
+from app.core.database import DatabaseManager
+
+from ._const import DemandRecommendConst
+from ._utils import DemandRecord, MatchResult
+
+
+class DemandRecommendMapper:
+    """需求-视频匹配数据层"""
+
+    def __init__(self, pool: DatabaseManager):
+        self.pool = pool
+        self.table = DemandRecommendConst.MATCH_RESULT_TABLE
+        self.db = DemandRecommendConst.MATCH_RESULT_DB
+
+    # ── 需求表读取（暂为占位，待实际表结构确认后补全 SQL） ──
+
+    async def fetch_demands_by_dt(self, dt: str) -> List[DemandRecord]:
+        """读取指定日期的需求记录列表
+
+        TODO: 替换为实际表名和字段映射
+        """
+        # query = """
+        #     SELECT * FROM demand_info
+        #     WHERE dt = %s
+        # """
+        # rows = await self.pool.async_fetch(query=query, params=(dt,), db_name=self.db)
+        # return [DemandRecord.from_dict(r) for r in (rows or [])]
+        return []
+
+    # ── 匹配结果写入 ──
+
+    async def save_match_results(self, results: List[MatchResult]) -> int:
+        """批量写入匹配结果（IGNORE 已存在的 uk 冲突行）"""
+        if not results:
+            return 0
+
+        query = f"""
+            INSERT IGNORE INTO {self.table}
+                (dt, demand_id, match_experiment_id, match_method, config_code,
+                 video_id, score, rank_position, video_title, video_detail)
+            VALUES
+                (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+        """
+        params = [
+            (
+                r.dt,
+                r.demand_id,
+                r.match_experiment_id,
+                r.match_method,
+                r.config_code,
+                r.video_id,
+                r.score,
+                r.rank_position,
+                r.video_title,
+                json.dumps(r.video_detail, ensure_ascii=False) if r.video_detail else None,
+            )
+            for r in results
+        ]
+        try:
+            affected = await self.pool.async_save(
+                query=query, params=params, db_name=self.db, batch=True
+            )
+            return affected
+        except Exception:
+            # async_save 内部已 rollback + log
+            return 0
+
+    # ── 匹配结果读取（供排序阶段使用） ──
+
+    async def fetch_matches_by_demand(self, demand_id: str) -> List[Dict[str, Any]]:
+        """按需求ID读取有效匹配结果"""
+        query = f"""
+            SELECT id, dt, demand_id, match_experiment_id, match_method, config_code,
+                   video_id, score, rank_position, video_title, video_detail
+            FROM {self.table}
+            WHERE demand_id = %s AND status = 1
+            ORDER BY config_code, rank_position
+        """
+        rows = await self.pool.async_fetch(query=query, params=(demand_id,), db_name=self.db)
+        return rows or []
+
+    async def fetch_matches_by_experiment(
+        self, experiment_id: str, limit: int = 1000
+    ) -> List[Dict[str, Any]]:
+        """按实验ID读取有效匹配结果"""
+        query = f"""
+            SELECT id, dt, demand_id, match_experiment_id, match_method, config_code,
+                   video_id, score, rank_position, video_title, video_detail
+            FROM {self.table}
+            WHERE match_experiment_id = %s AND status = 1
+            ORDER BY demand_id, config_code, rank_position
+            LIMIT %s
+        """
+        rows = await self.pool.async_fetch(
+            query=query, params=(experiment_id, limit), db_name=self.db
+        )
+        return rows or []
+
+    # ── 失效管理 ──
+
+    async def invalidate_by_dt(self, dt: str) -> int:
+        """将指定日期的匹配结果标记为失效"""
+        query = f"""
+            UPDATE {self.table}
+            SET status = 0
+            WHERE dt = %s AND status = 1
+        """
+        return await self.pool.async_save(query=query, params=(dt,), db_name=self.db) or 0
+
+    async def invalidate_by_demand(self, demand_id: str) -> int:
+        """将指定需求的匹配结果标记为失效"""
+        query = f"""
+            UPDATE {self.table}
+            SET status = 0
+            WHERE demand_id = %s AND status = 1
+        """
+        return await self.pool.async_save(query=query, params=(demand_id,), db_name=self.db) or 0

app/domains/recommend/demand_recommend/_utils.py

@@ -0,0 +1,263 @@
+import re
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from ._const import (
+    ConfigCode,
+    DemandRecommendConst,
+    DemandSource,
+    MatchMethod,
+)
+
+
+# ──────────────────────────────────────────────
+# Dataclasses
+# ──────────────────────────────────────────────
+
+@dataclass
+class DemandRecord:
+    """上游需求表的一行原始数据（字段暂按结果表推测，后续按实际表结构对齐）"""
+    dt: str = ""
+    action_type: str = ""
+    match_experiment_id: str = ""
+    demand_source_crowd: str = ""
+    demand_strategy: str = ""
+    match_strategy: str = ""
+    match_video_rule: str = ""
+    demand_id: str = ""
+    crowd_channel: str = ""
+    crowd_segment: str = ""
+    crowd_package: str = ""
+    conversion_target: str = ""
+    partner: str = ""
+    account: str = ""
+    scene_value: str = ""
+    demand_source: str = ""
+    drive_dim_time: str = ""
+    drive_dim_space: str = ""
+    demand_filter_strategy: str = ""
+    demand_video_id: int = 0
+    demand_video_title: str = ""
+    scene_content_id: str = ""
+    scene_content_title: str = ""
+    demand_topic: str = ""
+    demand_feature_points: str = ""
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "DemandRecord":
+        return cls(
+            dt=str(data.get("dt", "")),
+            action_type=str(data.get("action_type", "")),
+            match_experiment_id=str(data.get("match_experiment_id", "")),
+            demand_source_crowd=str(data.get("demand_source_crowd", "")),
+            demand_strategy=str(data.get("demand_strategy", "")),
+            match_strategy=str(data.get("match_strategy", "")),
+            match_video_rule=str(data.get("match_video_rule", "")),
+            demand_id=str(data.get("demand_id", "")),
+            crowd_channel=str(data.get("crowd_channel", "")),
+            crowd_segment=str(data.get("crowd_segment", "")),
+            crowd_package=str(data.get("crowd_package", "")),
+            conversion_target=str(data.get("conversion_target", "")),
+            partner=str(data.get("partner", "")),
+            account=str(data.get("account", "")),
+            scene_value=str(data.get("scene_value", "")),
+            demand_source=str(data.get("demand_source", "")),
+            drive_dim_time=str(data.get("drive_dim_time", "")),
+            drive_dim_space=str(data.get("drive_dim_space", "")),
+            demand_filter_strategy=str(data.get("demand_filter_strategy", "")),
+            demand_video_id=int(data.get("demand_video_id", 0) or 0),
+            demand_video_title=str(data.get("demand_video_title", "")),
+            scene_content_id=str(data.get("scene_content_id", "")),
+            scene_content_title=str(data.get("scene_content_title", "")),
+            demand_topic=str(data.get("demand_topic", "")),
+            demand_feature_points=str(data.get("demand_feature_points", "")),
+        )
+
+
+@dataclass
+class MatchStrategy:
+    """从 DemandRecord 解析出的匹配执行策略"""
+    demand_id: str
+    experiment_id: str
+    dt: str
+    match_methods: List[str] = field(default_factory=list)
+    config_codes: List[str] = field(default_factory=list)
+    top_n: int = DemandRecommendConst.DEFAULT_TOPN
+    query_text: str = ""
+    video_id: int = 0
+    content_id: str = ""
+    filter_rule: str = ""
+    multi_recall_fusion: bool = False
+
+
+@dataclass
+class MatchResult:
+    """单条匹配结果"""
+    dt: str
+    demand_id: str
+    match_experiment_id: str
+    match_method: str
+    config_code: str
+    video_id: int
+    score: float
+    rank_position: int = 0
+    video_title: str = ""
+    video_detail: Optional[Dict[str, Any]] = None
+
+
+# ──────────────────────────────────────────────
+# Strategy Parser
+# ──────────────────────────────────────────────
+
+class DemandStrategyParser:
+    """解析 DemandRecord → MatchStrategy"""
+
+    @staticmethod
+    def select_config_codes(match_strategy: str) -> List[str]:
+        """从匹配策略文本中推导 configCode 列表"""
+        if not match_strategy:
+            return [DemandRecommendConst.DEFAULT_CONFIG_CODE]
+
+        codes: List[str] = []
+        for keyword, code in DemandRecommendConst.STRATEGY_CONFIG_MAP.items():
+            if keyword in match_strategy and code not in codes:
+                codes.append(code)
+
+        if not codes:
+            codes.append(DemandRecommendConst.DEFAULT_CONFIG_CODE)
+        return codes
+
+    @staticmethod
+    def select_match_methods(demand: DemandRecord) -> List[str]:
+        """从需求行的 match_video_rule + 可用字段推导匹配方式"""
+        if not demand.match_video_rule:
+            return DemandStrategyParser._fallback_methods(demand)
+
+        methods: List[str] = []
+        for keyword, method in DemandRecommendConst.RULE_METHOD_MAP.items():
+            if keyword in demand.match_video_rule:
+                if method == MatchMethod.VIDEO_ID and demand.demand_video_id > 0:
+                    methods.append(method)
+                elif method == MatchMethod.CONTENT_ID and demand.scene_content_id:
+                    methods.append(method)
+                elif method == MatchMethod.TEXT and (
+                    demand.demand_topic or demand.demand_feature_points
+                ):
+                    methods.append(method)
+
+        if not methods:
+            return DemandStrategyParser._fallback_methods(demand)
+        return methods
+
+    @staticmethod
+    def _fallback_methods(demand: DemandRecord) -> List[str]:
+        """当 match_video_rule 无法解析时，按可用字段兜底推导"""
+        methods: List[str] = []
+        if demand.demand_video_id > 0:
+            methods.append(MatchMethod.VIDEO_ID)
+        if demand.scene_content_id:
+            methods.append(MatchMethod.CONTENT_ID)
+        if demand.demand_topic or demand.demand_feature_points:
+            methods.append(MatchMethod.TEXT)
+        if not methods:
+            methods.append(MatchMethod.TEXT)  # 最终兜底
+        return methods
+
+    @staticmethod
+    def parse_top_n(match_strategy: str) -> int:
+        """从匹配策略中解析 topN 参数，缺省 10"""
+        if not match_strategy:
+            return DemandRecommendConst.DEFAULT_TOPN
+        m = re.search(r"topN[=:]?\s*(\d+)", match_strategy, re.IGNORECASE)
+        if m:
+            return int(m.group(1))
+        return DemandRecommendConst.DEFAULT_TOPN
+
+    @classmethod
+    def parse(cls, demand: DemandRecord) -> MatchStrategy:
+        """完整解析一条需求记录为匹配策略"""
+        return MatchStrategy(
+            demand_id=demand.demand_id,
+            experiment_id=demand.match_experiment_id,
+            dt=demand.dt,
+            match_methods=cls.select_match_methods(demand),
+            config_codes=cls.select_config_codes(demand.match_strategy),
+            top_n=cls.parse_top_n(demand.match_strategy),
+            query_text=build_query_text(demand.demand_topic, demand.demand_feature_points),
+            video_id=demand.demand_video_id,
+            content_id=demand.scene_content_id,
+            filter_rule=demand.demand_filter_strategy,
+            multi_recall_fusion=("多路" in (demand.match_strategy or ""))
+                                or (len(cls.select_match_methods(demand)) > 1),
+        )
+
+
+# ──────────────────────────────────────────────
+# Helpers
+# ──────────────────────────────────────────────
+
+def build_query_text(topic: str, feature_points: str) -> str:
+    """拼接选题 + 特征点为检索文本"""
+    parts = [p for p in [topic, feature_points] if p and p.strip()]
+    return "。".join(parts) if parts else ""
+
+
+def parse_recall_items(
+    api_response: Dict[str, Any],
+    strategy: MatchStrategy,
+    match_method: str,
+    config_code: str,
+) -> List[MatchResult]:
+    """解析 API 返回结果为 MatchResult 列表"""
+    if not api_response or api_response.get("code") != 0:
+        return []
+
+    data = api_response.get("data")
+    if not data:
+        return []
+
+    # matchTopNVideo 返回 data 直接是 list
+    if isinstance(data, list):
+        items = data
+    else:
+        # recallTest 返回 data.items[]
+        items = data.get("items", [])
+
+    results: List[MatchResult] = []
+    for rank, item in enumerate(items, start=1):
+        vid = item.get("id") or item.get("videoId", 0)
+        if not vid:
+            continue
+        results.append(MatchResult(
+            dt=strategy.dt,
+            demand_id=strategy.demand_id,
+            match_experiment_id=strategy.experiment_id,
+            match_method=match_method,
+            config_code=config_code,
+            video_id=int(vid),
+            score=float(item.get("score", 0)),
+            rank_position=rank,
+            video_title=str(item.get("title", "")),
+            video_detail=item.get("videoDetail"),
+        ))
+    return results
+
+
+def merge_multi_recall(
+    result_groups: List[List[MatchResult]],
+    top_n: int,
+) -> List[MatchResult]:
+    """多路召回结果合并：按 video_id 去重，保留最高分，取 top_n"""
+    merged: Dict[int, MatchResult] = {}
+    for group in result_groups:
+        for r in group:
+            if r.video_id in merged:
+                if r.score > merged[r.video_id].score:
+                    merged[r.video_id] = r
+            else:
+                merged[r.video_id] = r
+
+    sorted_results = sorted(merged.values(), key=lambda x: x.score, reverse=True)
+    for i, r in enumerate(sorted_results[:top_n], start=1):
+        r.rank_position = i
+    return sorted_results[:top_n]

app/domains/recommend/demand_recommend/api_server.md

@@ -0,0 +1,723 @@
+# Video Search & Vector Recall API 接口文档
+
+> 项目: video-vector-server  
+> 文档版本: 2026-05-07  
+> 适用 Controller: `VideoSearchController`、`VectorRecallTestController`
+
+---
+
+## 通用约定
+
+### 基础路径
+
+| Controller | 前缀 |
+|---|---|
+| VideoSearchController | `/videoSearch` |
+| VectorRecallTestController | `/recallTest` |
+
+### 统一响应格式 `CommonResponse<T>`
+
+所有接口均返回以下 JSON 结构:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": <T>
+}
+```
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `code` | `int` | 业务状态码。`0` 表示成功 |
+| `msg` | `String` | 状态消息。成功时为 `"success"` |
+| `data` | `T` (泛型) | 业务数据,具体结构见各接口定义。失败时可能为 `null` |
+
+### 全局行为
+
+- **CORS**: 由全局拦截器 `CrosDomainAllowInterceptor` 统一处理,前端无需额外配置
+- **鉴权**: MVP 阶段不校验鉴权,所有接口均可直接调用
+
+---
+
+## 一、VideoSearchController (`/videoSearch`)
+
+视频解构与相似视频匹配。
+
+### 1.1 触发内容解构
+
+```
+POST /videoSearch/deconstruct
+```
+
+**用途**: 对指定内容(长文/图文/视频)发起 AI 解构任务,提取选题、灵感点、关键点、目的点等结构化信息。
+
+**Request Body** (`application/json`):
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `bizType` | `Integer` | 否 | 业务类型。`0` = 投流 |
+| `contentType` | `Integer` | 否 | 内容类型: `1` = 长文, `2` = 图文, `3` = 视频 |
+| `channelContentId` | `String` | 否 | 业务内容ID(帖子ID/视频ID) |
+| `title` | `String` | 否 | 标题 |
+| `bodyText` | `String` | 否 | 正文内容 |
+| `videoUrl` | `String` | 否 | 视频地址 |
+| `imageList` | `List<String>` | 否 | 图片URL列表 |
+| `channelAccountId` | `String` | 否 | 作者ID |
+| `channelAccountName` | `String` | 否 | 作者名称 |
+
+**Request 示例**:
+
+```json
+{
+  "bizType": 0,
+  "contentType": 3,
+  "channelContentId": "abc123",
+  "title": "产品种草视频",
+  "bodyText": "这是一款非常好用的护肤品...",
+  "videoUrl": "https://example.com/video.mp4",
+  "imageList": ["https://example.com/img1.jpg"],
+  "channelAccountId": "account_001",
+  "channelAccountName": "美妆达人小A"
+}
+```
+
+**Response**:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": "<taskId>"
+}
+```
+
+`data` 为 `String` 类型,返回解构任务的唯一标识 `taskId`,可用于后续调用 `getDeconstructResult` 查询结果。
+
+---
+
+### 1.2 查询解构结果
+
+```
+POST /videoSearch/getDeconstructResult
+```
+
+**用途**: 根据任务ID或业务内容ID查询解构任务的处理结果。
+
+**Request Body** (`application/json`):
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `taskId` | `String` | 否 | 解构任务ID(由 `deconstruct` 接口返回) |
+| `bizType` | `Integer` | 否 | 业务类型: `0` = 选题, `1` = 创作, `2` = 制作 |
+| `contentType` | `Integer` | 否 | 内容类型: `1` = 长文, `2` = 图文, `3` = 视频 |
+| `channelContentId` | `String` | 否 | 业务内容ID(帖子ID/视频ID) |
+| `forceRefresh` | `Boolean` | 否 | 是否强制从远程API重新拉取,不走Redis缓存 |
+
+> **查询逻辑**: 至少提供 `taskId` 或 `channelContentId` 之一。若提供 `taskId` 则按任务ID查询;若提供 `channelContentId` 则按内容ID查询最新结果。
+
+**Request 示例**:
+
+```json
+{
+  "taskId": "task_20260507_001",
+  "bizType": 0,
+  "contentType": 3,
+  "channelContentId": "abc123",
+  "forceRefresh": false
+}
+```
+
+**Response**: `data` 为 `JSONObject`,结构为解构引擎返回的原始结果JSON,包含但不限于:
+
+- 选题(topic)
+- 灵感点(inspiration points)
+- 关键点(key points)
+- 目的点(purpose points)
+- 各点的实质词汇及贡献度分数(0~1)
+
+---
+
+### 1.3 按内容ID获取解构结果
+
+```
+GET /videoSearch/getDeconstructResultByChannelContentId
+```
+
+**用途**: 通过业务内容ID直接获取解构结果(简化版,仅需 channelContentId)。
+
+**Query Parameters**:
+
+| 参数 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `channelContentId` | `String` | 是 | 业务内容ID(帖子ID/视频ID) |
+
+**示例请求**:
+
+```
+GET /videoSearch/getDeconstructResultByChannelContentId?channelContentId=abc123
+```
+
+**Response**: `data` 为 `JSONObject`,与接口 1.2 返回结构一致。
+
+---
+
+### 1.4 相似视频匹配(Top-N 召回)
+
+```
+POST /videoSearch/matchTopNVideo
+```
+
+**用途**: 基于内容解构结果,在向量库中检索最相似的 Top-N 个视频。
+
+**Request Body** (`application/json`):
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `channelContentId` | `String` | 否 | 参考内容的业务ID,用于自动获取该内容的解构向量 |
+| `configCode` | `String` | 否 | 向量配置编码,指定搜索维度。不传使用默认配置 |
+| `queryText` | `String` | 否 | 查询文本,将被向量化后检索 |
+| `queryVector` | `List<Float>` | 否 | 直接传入查询向量,优先级高于 `queryText` |
+| `topN` | `Integer` | 否 | 返回结果数量,默认 `10` |
+
+> **检索逻辑**: `queryVector` > `queryText` > `channelContentId`。即若传入 `queryVector` 直接用于检索;否则若有 `queryText` 则将其向量化;否则使用 `channelContentId` 的解构向量进行检索。
+
+**Request 示例**:
+
+```json
+{
+  "channelContentId": "abc123",
+  "configCode": "VIDEO_TOPIC",
+  "topN": 10
+}
+```
+
+按文本查询:
+
+```json
+{
+  "configCode": "VIDEO_INSPIRATION",
+  "queryText": "如何做美妆教程",
+  "topN": 20
+}
+```
+
+**Response**: `data` 为 `List<VideoMatchResult>`:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": [
+    {
+      "configCode": "VIDEO_TOPIC",
+      "videoId": 10001,
+      "score": 0.9523,
+      "videoDetail": {
+        "topic": "美妆护肤",
+        "灵感点": "产品成分解读",
+        "灵感点-实质": [
+          { "word": "成分", "score": 0.92 },
+          { "word": "安全", "score": 0.88 }
+        ],
+        "关键点": "使用时需注意过敏",
+        "关键点-实质": [
+          { "word": "过敏", "score": 0.85 }
+        ],
+        "目的点": "种草转化",
+        "目的点-实质": [
+          { "word": "购买", "score": 0.91 }
+        ]
+      }
+    }
+  ]
+}
+```
+
+**VideoMatchResult 字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `configCode` | `String` | 命中时使用的向量配置编码 |
+| `videoId` | `Long` | 匹配到的视频ID |
+| `score` | `Double` | 余弦相似度分值(0~1,越大越相似) |
+| `videoDetail` | `Map<String, Object>` | 视频解构详情,来源 Redis `recall:vid_decode:{vid}`。包含 topic、灵感点、关键点、目的点及其"实质"分词信息 |
+
+---
+
+### 1.5 获取所有配置编码
+
+```
+GET /videoSearch/getAllConfigCodes
+```
+
+**用途**: 查询系统支持的所有向量配置编码及对应描述,供前端下拉选择。
+
+**请求参数**: 无。
+
+**示例请求**:
+
+```
+GET /videoSearch/getAllConfigCodes
+```
+
+**Response**: `data` 为 `Map<String, String>`,key 为 configCode,value 为中文描述:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": {
+    "VIDEO_TOPIC": "选题维度",
+    "VIDEO_INSPIRATION": "灵感点维度",
+    "VIDEO_KEYPOINT": "关键点维度",
+    "VIDEO_PURPOSE": "目的点维度"
+  }
+}
+```
+
+---
+
+## 二、VectorRecallTestController (`/recallTest`)
+
+向量召回前端测试页面专用接口。MVP 阶段不加鉴权。
+
+### 支持的 configCode
+
+| configCode | 说明 |
+|---|---|
+| `VIDEO_TOPIC` | 选题维度检索(默认) |
+| `VIDEO_INSPIRATION` | 灵感点维度检索 |
+
+> 不传 `configCode` 时默认使用 `VIDEO_TOPIC`。
+
+---
+
+### 2.1 获取视频基础详情 (Tab1)
+
+```
+GET /recallTest/videoDetail
+```
+
+**用途**: 查询单个视频的基础信息,用于测试页面 Tab1 展示。
+
+**Query Parameters**:
+
+| 参数 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `videoId` | `Long` | 是 | 视频ID |
+
+**示例请求**:
+
+```
+GET /recallTest/videoDetail?videoId=10086
+```
+
+**Response**: `data` 为 `VideoBasicVO`:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": {
+    "videoId": 10086,
+    "title": "2025春季妆容教程",
+    "videoUrl": "https://example.com/video/10086.mp4",
+    "cover": "https://example.com/cover/10086.jpg",
+    "playCount": "--"
+  }
+}
+```
+
+**VideoBasicVO 字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `videoId` | `Long` | 视频ID |
+| `title` | `String` | 视频标题 |
+| `videoUrl` | `String` | 视频播放地址 |
+| `cover` | `String` | 封面图URL |
+| `playCount` | `String` | 播放量。长视频API当前不返回播放量,值为 `"--"` |
+
+---
+
+### 2.2 文本召回 (Tab2)
+
+```
+POST /recallTest/matchByText
+```
+
+**用途**: 输入自由文本,在向量库中检索语义最相似的 Top-N 个视频/素材/长文。
+
+**Request Body** (`application/json`):
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `queryText` | `String` | 是 | 查询文本,将向量化后在指定维度检索 |
+| `configCode` | `String` | 否 | 向量配置编码,不传默认 `VIDEO_TOPIC`。支持: `VIDEO_TOPIC` / `VIDEO_INSPIRATION` |
+| `topN` | `Integer` | 否 | 返回 Top-N 结果,默认 `10` |
+
+**Request 示例**:
+
+```json
+{
+  "queryText": "美妆护肤教程",
+  "configCode": "VIDEO_TOPIC",
+  "topN": 20
+}
+```
+
+**Response**: `data` 为 `RecallResultVO`:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": {
+    "items": [
+      {
+        "id": 10001,
+        "modality": "VIDEO",
+        "configCode": "VIDEO_TOPIC",
+        "score": 0.9523,
+        "title": "春季护肤全攻略",
+        "cover": "https://example.com/cover/10001.jpg",
+        "videoUrl": "https://example.com/video/10001.mp4",
+        "imageList": null,
+        "bodyText": null,
+        "playCount": "--",
+        "exposure": "--",
+        "ctr": "--",
+        "readCount": "--",
+        "rov": "--",
+        "videoDetail": {
+          "topic": "护肤教程",
+          "灵感点": "产品成分解析",
+          "灵感点-实质": [
+            { "word": "成分", "score": 0.92 }
+          ],
+          "关键点": "敏感肌适用",
+          "关键点-实质": [
+            { "word": "敏感肌", "score": 0.89 }
+          ],
+          "目的点": "提升转化",
+          "目的点-实质": [
+            { "word": "购买", "score": 0.87 }
+          ]
+        }
+      },
+      {
+        "id": 20005,
+        "modality": "MATERIAL",
+        "configCode": "VIDEO_TOPIC",
+        "score": 0.8912,
+        "title": "护肤品种草图文",
+        "cover": "https://example.com/cover/20005.jpg",
+        "videoUrl": null,
+        "imageList": ["https://example.com/material/20005_1.jpg", "https://example.com/material/20005_2.jpg"],
+        "bodyText": null,
+        "playCount": "--",
+        "exposure": "--",
+        "ctr": "--",
+        "readCount": "--",
+        "rov": "--",
+        "videoDetail": { }
+      }
+    ],
+    "videoCount": 1,
+    "materialCount": 1,
+    "articleCount": 0,
+    "total": 2
+  }
+}
+```
+
+**RecallResultVO 字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `items` | `List<VideoMatchEnrichedVO>` | 召回结果列表(已 enrich,含模态信息) |
+| `videoCount` | `int` | 命中视频数量 |
+| `materialCount` | `int` | 命中素材(图文)数量 |
+| `articleCount` | `int` | 命中长文数量 |
+| `total` | `int` | 总召回条数 |
+
+**VideoMatchEnrichedVO 单条结果字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `id` | `Long` | 业务ID。视频时 = `wx_video.id`,素材时 = `channelContentId` 数值化 |
+| `modality` | `Modality` 枚举 | 模态类型: `VIDEO` / `MATERIAL` / `ARTICLE` |
+| `configCode` | `String` | 命中的向量配置编码 |
+| `score` | `Double` | 余弦相似度分值(0~1) |
+| `title` | `String` | 标题 |
+| `cover` | `String` | 封面/缩略图URL |
+| `videoUrl` | `String` | 视频播放地址(仅 `modality=VIDEO` 时有值,其余为 `null`) |
+| `imageList` | `List<String>` | 图片列表(仅 `modality=MATERIAL` 时有值,其余为 `null`) |
+| `bodyText` | `String` | 正文(仅 `modality=ARTICLE` 时有值,其余为 `null`) |
+| `playCount` | `String` | 播放量,默认 `"--"` |
+| `exposure` | `String` | 曝光量,默认 `"--"` |
+| `ctr` | `String` | CTR,默认 `"--"` |
+| `readCount` | `String` | 阅读数,默认 `"--"` |
+| `rov` | `String` | ROV,默认 `"--"` |
+| `videoDetail` | `Map<String, Object>` | 视频解构详情 KV(来源 Redis `recall:vid_decode:{vid}`),含 topic、灵感点、关键点、目的点及其"实质-分词" |
+
+**Modality 枚举映射**:
+
+| 枚举值 | 对应 content_type | 说明 |
+|---|---|---|
+| `VIDEO` | `3` | 视频 |
+| `MATERIAL` | `2` | 图文素材 |
+| `ARTICLE` | `1` | 长文 |
+
+> `content_type` 缺省或未知时默认按 `VIDEO` 处理。
+
+---
+
+### 2.3 获取视频解构层级 (Tab1 解构树)
+
+```
+GET /recallTest/deconstructPoints
+```
+
+**用途**: 获取视频的解构层级数据(选题 + 高价值点),用于前端解构树组件递归渲染。数据来源: Singapore RDS → Python 解析筛选 → 国内 Redis。
+
+**Query Parameters**:
+
+| 参数 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `videoId` | `Long` | 是 | 视频ID/素材ID |
+
+**示例请求**:
+
+```
+GET /recallTest/deconstructPoints?videoId=10086
+```
+
+**Response**: `data` 为 `DeconstructPointsVO`:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": {
+    "vid": 10086,
+    "title": "2025春季妆容教程",
+    "videoUrl": "https://example.com/video/10086.mp4",
+    "htmlUrl": "https://example.com/viz/10086.html",
+    "topic": "美妆教程-春季妆容",
+    "highValuePoints": [
+      {
+        "id": "inspiration_1",
+        "type": "灵感点",
+        "name": "春季流行色系运用",
+        "essences": [
+          { "word": "春季", "score": 0.95 },
+          { "word": "色系", "score": 0.91 },
+          { "word": "流行", "score": 0.86 }
+        ]
+      },
+      {
+        "id": "kp_abc123",
+        "type": "关键点",
+        "name": "底妆持久度关键步骤",
+        "essences": [
+          { "word": "持久", "score": 0.93 },
+          { "word": "底妆", "score": 0.88 }
+        ]
+      },
+      {
+        "id": "purpose_1",
+        "type": "目的点",
+        "name": "提升完播率与互动",
+        "essences": [
+          { "word": "完播", "score": 0.82 },
+          { "word": "互动", "score": 0.80 }
+        ]
+      }
+    ]
+  }
+}
+```
+
+**DeconstructPointsVO 字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `vid` | `Long` | 视频ID |
+| `title` | `String` | 视频标题 |
+| `videoUrl` | `String` | 视频播放地址 |
+| `htmlUrl` | `String` | 带权重的可视化页面URL |
+| `topic` | `String` | 最终选题(格式: `最终选题.选题`) |
+| `highValuePoints` | `List<HighValuePoint>` | 实质≥0.8 的高价值解构点列表 |
+
+**HighValuePoint 子结构**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `id` | `String` | 业务侧ID,格式如 `inspiration_1` / `purpose_1` / `kp_xxxxxx` |
+| `type` | `String` | 点类型: `灵感点` / `目的点` / `关键点` |
+| `name` | `String` | 该点的描述名称 |
+| `essences` | `List<EssenceWord>` | 拆解出的"实质"分词列表(score ≥ 0.8) |
+
+**EssenceWord 子结构**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `word` | `String` | 实质词 |
+| `score` | `Double` | 词级贡献度(0~1,越大越重要) |
+
+---
+
+### 2.4 按视频ID召回相似内容 (Tab1 解构节点点击触发)
+
+```
+POST /recallTest/matchByVideoId
+```
+
+**用途**: 以指定视频/素材为参考,召回与之在指定维度(选题/灵感点)上最相似的内容。典型场景: 用户在解构树上看到某个节点,点击后查询与该节点对应的高价值点相似的内容。
+
+**Request Body** (`application/json`):
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `videoId` | `Long` | 是 | 视频ID 或 `channelContentId` 数值化后的值 |
+| `configCode` | `String` | 否 | 向量配置编码,不传默认 `VIDEO_TOPIC`。支持: `VIDEO_TOPIC` / `VIDEO_INSPIRATION` |
+| `topN` | `Integer` | 否 | 返回 Top-N 结果,默认 `10` |
+
+**Request 示例**:
+
+```json
+{
+  "videoId": 10086,
+  "configCode": "VIDEO_INSPIRATION",
+  "topN": 15
+}
+```
+
+**Response**: `data` 为 `RecallResultVO`,结构与 2.2 接口完全一致。
+
+---
+
+### 2.5 获取视频AI理解结果 (Tab1)
+
+```
+GET /recallTest/aiUnderstanding
+```
+
+**用途**: 获取视频的 AI 理解结果(选题、主题、关键词、口播文案等)。数据来源: ODPS `loghubods.result_log` → DataWorks 同步 Job → 本地表 `video_ai_understanding`。
+
+> **重要**: MVP 阶段本地表可能为空,此时 `data` 的 field 全部为 `null`,前端需展示"AI理解数据未就绪,等待同步Job"。**严禁后端伪造任何字段返回值。**
+
+**Query Parameters**:
+
+| 参数 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `videoId` | `Long` | 是 | 视频ID |
+
+**示例请求**:
+
+```
+GET /recallTest/aiUnderstanding?videoId=10086
+```
+
+**Response (数据就绪时)**: `data` 为 `AIUnderstandingVO`:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": {
+    "videoId": 10086,
+    "contentTopic": "美妆护肤",
+    "videoTheme": "春季妆容教程",
+    "videoKeywords": "春季,妆容,护肤,教程",
+    "videoNarration": "大家好,今天我们来分享一款春季日常妆容...",
+    "dt": "2026050712"
+  }
+}
+```
+
+**Response (数据未就绪时)**:
+
+```json
+{
+  "code": 0,
+  "msg": "success",
+  "data": null
+}
+```
+
+> 前端在收到 `data == null` 时应展示"AI理解数据未就绪"占位状态,不可报错。
+
+**AIUnderstandingVO 字段说明**:
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `videoId` | `Long` | 视频ID |
+| `contentTopic` | `String` | 内容选题(AI识别) |
+| `videoTheme` | `String` | 视频主题 |
+| `videoKeywords` | `String` | 视频关键词(逗号分隔) |
+| `videoNarration` | `String` | 视频口播文案(ASR识别文本) |
+| `dt` | `String` | 数据所属分区,格式 `yyyyMMddHH`(如 `2026050712` 表示 2026-05-07 12时) |
+
+---
+
+## 三、数据流与架构说明
+
+### 解构数据管线的来源链路
+
+```
+Singapore RDS (aigc_topic_decode_task_result)
+  → Python 解析脚本(筛选实质 ≥ 0.8 的高价值点)
+    → 国内 Redis (key: recall:vid_decode:{vid})
+      → Java 后端 (DeconstructPointsVO / VideoMatchResult)
+        → 前端渲染
+```
+
+### 向量召回流程
+
+```
+输入(queryText / videoId / channelContentId)
+  → 获取或计算查询向量
+    → Milvus/ES 向量检索 (按 configCode 指定维度)
+      → 返回 Top-N 候选 vid
+        → enrich: 从 Redis recall:vid_decode:{vid} 读取解构详情
+          → 组装 RecallResultVO / VideoMatchResult 返回
+```
+
+### AI理解数据链路
+
+```
+ODPS (loghubods.result_log)
+  → DataWorks 同步 Job
+    → 本地 MySQL (video_ai_understanding 表)
+      → Java 后端查询
+        → 前端渲染
+```
+
+---
+
+## 四、接口总览表
+
+| 接口 | Method | 路径 | 说明 |
+|---|---|---|---|
+| 触发内容解构 | `POST` | `/videoSearch/deconstruct` | 发起AI解构任务,返回 taskId |
+| 查询解构结果 | `POST` | `/videoSearch/getDeconstructResult` | 按 taskId/channelContentId 查结果 |
+| 按内容ID查解构 | `GET` | `/videoSearch/getDeconstructResultByChannelContentId` | 简化查询,仅需 channelContentId |
+| 相似视频匹配 | `POST` | `/videoSearch/matchTopNVideo` | Top-N 向量召回相似视频 |
+| 获取所有配置码 | `GET` | `/videoSearch/getAllConfigCodes` | 查询支持的向量配置维度 |
+| 视频基础详情 | `GET` | `/recallTest/videoDetail` | 查单视频基础信息(Tab1) |
+| 文本召回 | `POST` | `/recallTest/matchByText` | 自由文本语义检索(Tab2) |
+| 解构层级 | `GET` | `/recallTest/deconstructPoints` | 视频解构树数据(Tab1) |
+| 按视频ID召回 | `POST` | `/recallTest/matchByVideoId` | 相似视频召回(Tab1节点触发) |
+| AI理解结果 | `GET` | `/recallTest/aiUnderstanding` | 视频AI理解数据(Tab1) |
+
+---
+
+## 五、错误状态码
+
+| code | 含义 | 典型场景 |
+|---|---|---|
+| `0` | 成功 | — |
+| 非 `0` | 业务异常 | 参数校验失败、向量库不可用、查询超时等,具体 `msg` 字段会描述原因 |

app/domains/recommend/demand_recommend/entrance.py

@@ -0,0 +1,148 @@
+import traceback
+from typing import Dict, List
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+from app.infra.external import feishu_robot
+from app.infra.shared import run_tasks_with_asyncio_task_group
+
+from ._const import DemandRecommendConst
+from ._mapper import DemandRecommendMapper
+from ._utils import DemandRecord, MatchResult
+from .match_engine import DemandVideoMatchEngine
+
+
+class DemandRecommendTask(DemandRecommendConst):
+    """需求-视频匹配任务入口"""
+
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.mapper = DemandRecommendMapper(pool)
+        self.engine = DemandVideoMatchEngine(base_url=self.BASE_URL)
+        self.log_service = log_service
+
+    # ── 单条需求匹配 handler（供 worker group 调用） ──
+
+    async def _match_and_save(self, demand: DemandRecord) -> List[MatchResult]:
+        """匹配单条需求 + 写入结果"""
+        results = await self.engine.match(demand)
+        if results:
+            await self.mapper.save_match_results(results)
+        return results
+
+    async def _handle_one_demand(self, demand: DemandRecord) -> None:
+        """worker group 的 handler 签名适配"""
+        await self._match_and_save(demand)
+
+    # ── 飞书通知 ──
+
+    async def _notify(
+        self,
+        title: str,
+        detail: Dict,
+        mention: bool = False,
+    ):
+        try:
+            await feishu_robot.bot(
+                title=title,
+                detail=detail,
+                mention=mention,
+            )
+        except Exception:
+            pass  # 通知失败不影响主流程
+
+    # ── 主入口 ──
+
+    async def deal(
+        self,
+        demands: List[DemandRecord] = None,
+        dt: str = None,
+    ):
+        """
+        主流程:
+          1. 获取需求列表（从参数传入 或 按 dt 从 DB 读取）
+          2. 并发调用匹配引擎
+          3. 写入匹配结果
+          4. 统计 & 通知
+
+        Args:
+            demands: 直接传入的需求列表（用于测试/手动触发）
+            dt: 按日期从 DB 读取需求（当 demands 为空时使用）
+        """
+        if not demands:
+            if dt:
+                demands = await self.mapper.fetch_demands_by_dt(dt)
+            if not demands:
+                await self.log_service.log(
+                    contents={
+                        "task": "demand_video_match",
+                        "status": "skip",
+                        "message": "无需求记录待匹配",
+                        "dt": dt,
+                    }
+                )
+                return {
+                    "total_demands": 0,
+                    "matched_demands": 0,
+                    "total_videos": 0,
+                    "errors": [],
+                }
+
+        total_demands = len(demands)
+        await self.log_service.log(
+            contents={
+                "task": "demand_video_match",
+                "status": "start",
+                "total_demands": total_demands,
+                "dt": dt,
+            }
+        )
+
+        # 并发匹配
+        result = await run_tasks_with_asyncio_task_group(
+            task_list=demands,
+            handler=self._handle_one_demand,
+            description="需求-视频匹配",
+            max_concurrency=self.MAX_WORKERS,
+            unit="demand",
+        )
+
+        processed = result["processed_task"]
+        errors = result["errors"]
+
+        if errors:
+            await self.log_service.log(
+                contents={
+                    "task": "demand_video_match",
+                    "status": "partial",
+                    "processed": processed,
+                    "errors_count": len(errors),
+                    "errors_detail": [
+                        {"idx": idx, "demand_id": getattr(d, "demand_id", ""), "error": str(e)}
+                        for idx, d, e in errors[:10]
+                    ],
+                }
+            )
+
+        summary = {
+            "total_demands": total_demands,
+            "matched_demands": processed,
+            "errors_count": len(errors),
+        }
+
+        if errors:
+            await self._notify(
+                title=f"需求-视频匹配完成(有{len(errors)}条失败)",
+                detail=summary,
+                mention=True,
+            )
+        else:
+            await self._notify(
+                title="需求-视频匹配完成",
+                detail=summary,
+                mention=False,
+            )
+
+        return summary
+
+
+__all__ = ["DemandRecommendTask"]

app/domains/recommend/demand_recommend/match_engine.py

@@ -0,0 +1,126 @@
+import asyncio
+from typing import Any, Dict, List, Optional
+
+from app.infra.shared.http_client import AsyncHttpClient
+
+from ._const import (
+    ConfigCode,
+    DemandRecommendConst,
+    MatchMethod,
+)
+from ._utils import (
+    DemandRecord,
+    DemandStrategyParser,
+    MatchResult,
+    MatchStrategy,
+    merge_multi_recall,
+    parse_recall_items,
+)
+
+
+class DemandVideoMatchEngine:
+    """统一匹配引擎：读策略 → 调用 API → 返回 MatchResult 列表"""
+
+    def __init__(self, base_url: str = DemandRecommendConst.BASE_URL):
+        self.base_url = base_url.rstrip("/")
+
+    # ── 统一入口 ──
+
+    async def match(self, demand: DemandRecord) -> List[MatchResult]:
+        """匹配单条需求，返回去重/合并后的结果"""
+        strategy = DemandStrategyParser.parse(demand)
+        if not strategy.match_methods or not strategy.config_codes:
+            return []
+
+        # 构建所有 (match_method, config_code) 笛卡尔积任务
+        tasks = []
+        for method in strategy.match_methods:
+            for code in strategy.config_codes:
+                tasks.append(self._match_one(method, code, strategy))
+
+        # 串行执行同一需求的多个 API 调用（避免单需求打太多请求）
+        all_result_groups: List[List[MatchResult]] = []
+        for task in tasks:
+            try:
+                results = await task
+                if results:
+                    all_result_groups.append(results)
+            except Exception:
+                continue  # 单路失败不阻塞其他路
+
+        if not all_result_groups:
+            return []
+
+        if strategy.multi_recall_fusion and len(all_result_groups) > 1:
+            return merge_multi_recall(all_result_groups, strategy.top_n)
+
+        # 单路召回：取第一个非空组
+        return all_result_groups[0][:strategy.top_n]
+
+    async def _match_one(
+        self,
+        method: str,
+        config_code: str,
+        strategy: MatchStrategy,
+    ) -> List[MatchResult]:
+        """执行单路匹配"""
+        async with AsyncHttpClient(timeout=DemandRecommendConst.API_TIMEOUT) as client:
+            if method == MatchMethod.TEXT:
+                resp = await self._match_by_text(client, strategy.query_text, config_code, strategy.top_n)
+            elif method == MatchMethod.VIDEO_ID:
+                resp = await self._match_by_video_id(client, strategy.video_id, config_code, strategy.top_n)
+            elif method == MatchMethod.CONTENT_ID:
+                resp = await self._match_by_content_id(client, strategy.content_id, config_code, strategy.top_n)
+            else:
+                return []
+        return parse_recall_items(resp, strategy, method, config_code)
+
+    # ── API 调用 ──
+
+    async def _match_by_text(
+        self,
+        client: AsyncHttpClient,
+        query_text: str,
+        config_code: str,
+        top_n: int,
+    ) -> Dict[str, Any]:
+        url = f"{self.base_url}{DemandRecommendConst.ApiPath.MATCH_BY_TEXT}"
+        body = {
+            "queryText": query_text,
+            "configCode": config_code,
+            "topN": top_n,
+        }
+        resp = await client.post(url, json=body)
+        return resp if isinstance(resp, dict) else {}
+
+    async def _match_by_video_id(
+        self,
+        client: AsyncHttpClient,
+        video_id: int,
+        config_code: str,
+        top_n: int,
+    ) -> Dict[str, Any]:
+        url = f"{self.base_url}{DemandRecommendConst.ApiPath.MATCH_BY_VIDEO_ID}"
+        body = {
+            "videoId": video_id,
+            "configCode": config_code,
+            "topN": top_n,
+        }
+        resp = await client.post(url, json=body)
+        return resp if isinstance(resp, dict) else {}
+
+    async def _match_by_content_id(
+        self,
+        client: AsyncHttpClient,
+        content_id: str,
+        config_code: str,
+        top_n: int,
+    ) -> Dict[str, Any]:
+        url = f"{self.base_url}{DemandRecommendConst.ApiPath.MATCH_TOP_N_VIDEO}"
+        body = {
+            "channelContentId": content_id,
+            "configCode": config_code,
+            "topN": top_n,
+        }
+        resp = await client.post(url, json=body)
+        return resp if isinstance(resp, dict) else {}