apps/api/app/api/routes/ai.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 
 from typing import List, Literal, Optional
 
-from app.services.ai_service import TITLE_STYLES, generate_smart_titles
+from app.services.ai_service import TITLE_STYLES, generate_smart_titles, semantic_match_assets
 from fastapi import APIRouter
 from pydantic import BaseModel, Field
 
@@ -45,6 +45,42 @@ class TitleStyleInfo(BaseModel):
     description: str
 
 
+# ── 素材语义匹配 ────────────────────────────────────────────────────────────
+
+
+class AssetMatchItem(BaseModel):
+    """待匹配素材项."""
+
+    id: str = Field(..., description="素材ID")
+    name: str = Field(default="", description="素材名称")
+    tags: List[str] = Field(default_factory=list, description="标签列表")
+    description: str = Field(default="", description="素材描述")
+
+
+class SemanticMatchRequest(BaseModel):
+    """语义匹配请求."""
+
+    description: str = Field(..., min_length=1, max_length=500, description="目标视频内容描述")
+    assets: List[AssetMatchItem] = Field(..., min_length=1, max_length=100, description="待匹配素材列表")
+    top_k: int = Field(default=0, ge=0, le=100, description="返回前K个，0返回全部")
+
+
+class SemanticMatchResultItem(AssetMatchItem):
+    """匹配结果项."""
+
+    match_score: float = Field(..., description="匹配度评分 0-1")
+    match_reason: str = Field(..., description="匹配方式：doubao_semantic / fallback_keyword / fallback_default")
+
+
+class SemanticMatchResponse(BaseModel):
+    """语义匹配响应."""
+
+    matches: List[SemanticMatchResultItem] = Field(..., description="按匹配度降序排列的素材列表")
+    source: str = Field(..., description="来源：doubao / fallback")
+    description: str = Field(..., description="原始描述")
+    total: int = Field(..., description="输入素材总数")
+
+
 # ── 路由 ────────────────────────────────────────────────────────────────────
 
 
@@ -70,3 +106,31 @@ def list_title_styles():
         TitleStyleInfo(key=key, name=info["name"], description=info["description"])
         for key, info in TITLE_STYLES.items()
     ]
+
+
+@router.post("/assets/match", response_model=SemanticMatchResponse)
+def match_assets(request: SemanticMatchRequest):
+    """智能素材语义匹配.
+
+    根据用户描述，对素材列表做语义匹配并按匹配度排序。
+    未配置豆包 API Key 时自动降级为关键词匹配。
+
+    - 支持最多 100 个素材同时匹配
+    - 返回 match_score (0-1)，按降序排列
+    - top_k 可限制返回数量
+    """
+    # 转为 dict 传给服务层
+    assets_dict = [asset.model_dump() for asset in request.assets]
+
+    result = semantic_match_assets(
+        description=request.description,
+        assets=assets_dict,
+        top_k=request.top_k,
+    )
+
+    return SemanticMatchResponse(
+        matches=[SemanticMatchResultItem(**m) for m in result["matches"]],
+        source=result["source"],
+        description=result["description"],
+        total=result["total"],
+    )

apps/api/app/services/ai_service.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import json
 import logging
+import math
 import random
 import time
 from typing import Any, Dict, List, Optional
@@ -314,6 +315,266 @@ def generate_smart_titles(
     }
 
 
+# ── 智能素材语义匹配 ───────────────────────────────────────────────────────────
+
+
+def _semantic_match_fallback(
+    description: str,
+    assets: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    """本地降级：基于关键词的简单匹配.
+
+    计算描述中的关键词与素材名称/标签/描述的重叠度，
+    作为匹配度评分。0-1分。
+    """
+    import re
+
+    # 提取关键词（中文按2字以上片段，英文按单词）
+    desc = description.lower()
+    # 简单分词：提取2字以上的中文字符串和英文单词
+    keywords = set()
+    # 英文单词
+    for word in re.findall(r"[a-zA-Z]{3,}", desc):
+        keywords.add(word)
+    # 中文2-4字片段
+    for i in range(len(desc)):
+        for j in range(i + 2, min(i + 5, len(desc) + 1)):
+            fragment = desc[i:j]
+            if all("\u4e00" <= c <= "\u9fff" for c in fragment):
+                keywords.add(fragment)
+
+    if not keywords:
+        # 没有关键词时给所有素材中等分数
+        for asset in assets:
+            asset["match_score"] = 0.5
+            asset["match_reason"] = "fallback_default"
+        return assets
+
+    results = []
+    for asset in assets:
+        # 组合素材的文本信息：名称 + 标签 + 描述
+        asset_text_parts = [
+            str(asset.get("name", "")).lower(),
+            " ".join(str(t) for t in asset.get("tags", [])).lower(),
+            str(asset.get("description", "")).lower(),
+        ]
+        asset_text = " | ".join(asset_text_parts)
+
+        # 计算匹配度：命中关键词占比 + 稀有关键词加权
+        hit_count = 0
+        hit_keywords = []
+        for kw in keywords:
+            if kw in asset_text:
+                hit_count += 1
+                hit_keywords.append(kw)
+
+        # 基础匹配度 = 命中关键词数 / 总关键词数（开根号平滑）
+        base_score = math.sqrt(hit_count / len(keywords)) if keywords else 0.5
+
+        # 名称命中加分（名称匹配更重要）
+        name = str(asset.get("name", "")).lower()
+        name_hits = sum(1 for kw in hit_keywords if kw in name)
+        name_bonus = min(0.2, name_hits * 0.05)
+
+        score = min(1.0, base_score * 0.8 + name_bonus)
+        score = round(score, 3)
+
+        results.append({
+            **asset,
+            "match_score": score,
+            "match_reason": "fallback_keyword",
+        })
+
+    # 按匹配度降序
+    results.sort(key=lambda x: x["match_score"], reverse=True)
+    return results
+
+
+def _parse_semantic_match_response(
+    content: str,
+    asset_ids: List[str],
+) -> Optional[Dict[str, float]]:
+    """从模型返回中解析素材匹配度.
+
+    期望格式：JSON 对象 {asset_id: score} 或 {"matches": [{asset_id, score}]}
+    score 范围 0-1。
+    """
+    if not content:
+        return None
+
+    # 尝试解析 JSON
+    try:
+        cleaned = content.strip()
+        if cleaned.startswith("```"):
+            cleaned = cleaned.strip("`")
+            if cleaned.lower().startswith("json"):
+                cleaned = cleaned[4:]
+            cleaned = cleaned.strip()
+
+        data = json.loads(cleaned)
+
+        result: Dict[str, float] = {}
+
+        # 格式1: {"asset_id1": 0.8, "asset_id2": 0.6}
+        if isinstance(data, dict):
+            if "matches" in data and isinstance(data["matches"], list):
+                # 格式2: {"matches": [{"asset_id": "...", "score": 0.8}]}
+                for item in data["matches"]:
+                    if isinstance(item, dict):
+                        aid = item.get("asset_id") or item.get("id")
+                        score = item.get("score", 0)
+                        if aid and isinstance(score, (int, float)):
+                            result[str(aid)] = max(0.0, min(1.0, float(score)))
+            else:
+                for key, value in data.items():
+                    if isinstance(value, (int, float)):
+                        result[str(key)] = max(0.0, min(1.0, float(value)))
+
+        # 格式3: [{"asset_id": "...", "score": 0.8}]
+        elif isinstance(data, list):
+            for item in data:
+                if isinstance(item, dict):
+                    aid = item.get("asset_id") or item.get("id")
+                    score = item.get("score", 0)
+                    if aid and isinstance(score, (int, float)):
+                        result[str(aid)] = max(0.0, min(1.0, float(score)))
+
+        if len(result) >= max(1, len(asset_ids) // 2):  # 至少一半素材有评分才算成功
+            return result
+
+    except (json.JSONDecodeError, ValueError):
+        pass
+
+    return None
+
+
+def semantic_match_assets(
+    description: str,
+    assets: List[Dict[str, Any]],
+    top_k: int = 0,
+) -> Dict[str, Any]:
+    """智能素材语义匹配.
+
+    根据用户描述，评估每个素材的语义匹配度并排序。
+
+    Args:
+        description: 用户描述的目标视频内容
+        assets: 素材列表，每个素材需含 id/name/tags/description 等字段
+        top_k: 返回前K个，0表示返回全部
+
+    Returns:
+        {
+            "matches": [{"asset_id": ..., "match_score": ..., ...}],
+            "source": "doubao" | "fallback",
+            "description": "...",
+            "total": 总数,
+        }
+    """
+    description = (description or "").strip()
+    if not assets:
+        return {"matches": [], "source": "fallback", "description": description, "total": 0}
+
+    client = DoubaoAIClient()
+    if not client.is_available:
+        logger.info("豆包API未配置，使用本地降级做素材语义匹配")
+        matched = _semantic_match_fallback(description, assets)
+        if top_k > 0:
+            matched = matched[:top_k]
+        return {
+            "matches": matched,
+            "source": "fallback",
+            "description": description,
+            "total": len(assets),
+        }
+
+    # 构建素材信息（控制 token 数量）
+    asset_summaries = []
+    for asset in assets[:50]:  # 最多传50个素材给模型
+        aid = asset.get("id", "")
+        name = asset.get("name", "")[:50]
+        tags = asset.get("tags", [])
+        tags_str = ",".join(str(t) for t in tags[:5])
+        desc = str(asset.get("description", ""))[:80]
+        asset_summaries.append(
+            f"ID:{aid} | 名称:{name} | 标签:[{tags_str}] | 描述:{desc}"
+        )
+
+    asset_ids = [str(a.get("id", "")) for a in assets[:50]]
+
+    system_prompt = (
+        "你是一个专业的视频素材匹配助手。"
+        "根据用户的视频目标描述，评估每个素材的匹配程度。\n"
+        "评分规则：\n"
+        "- 0.0-0.3: 完全不相关\n"
+        "- 0.3-0.6: 有一定关联但不够匹配\n"
+        "- 0.6-0.8: 比较匹配，适合使用\n"
+        "- 0.8-1.0: 高度匹配，非常适合\n"
+        "只返回JSON对象，key为素材ID，value为匹配分数（0-1之间的小数）。"
+        "不要其他文字说明。"
+    )
+
+    user_prompt = (
+        f"目标视频描述：{description}\n\n"
+        f"素材列表：\n" + "\n".join(asset_summaries) +
+        f"\n\n请返回每个素材的匹配分数JSON："
+    )
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    result = client._chat_completion(
+        messages=messages,
+        temperature=0.3,
+        max_tokens=1024,
+    )
+
+    if result:
+        scores = _parse_semantic_match_response(result, asset_ids)
+        if scores:
+            # 把评分填回素材
+            matched = []
+            for asset in assets:
+                aid = str(asset.get("id", ""))
+                score = scores.get(aid, 0.3)  # 没评分的给默认偏低分
+                matched.append({
+                    **asset,
+                    "match_score": round(score, 3),
+                    "match_reason": "doubao_semantic",
+                })
+            matched.sort(key=lambda x: x["match_score"], reverse=True)
+
+            logger.info(
+                "豆包语义匹配完成: assets=%d top_score=%.2f description=%s...",
+                len(matched),
+                matched[0]["match_score"] if matched else 0,
+                description[:20],
+            )
+
+            if top_k > 0:
+                matched = matched[:top_k]
+
+            return {
+                "matches": matched,
+                "source": "doubao",
+                "description": description,
+                "total": len(assets),
+            }
+        logger.warning("豆包语义匹配返回解析失败，降级到本地: %s", result[:100])
+
+    # 降级
+    matched = _semantic_match_fallback(description, assets)
+    if top_k > 0:
+        matched = matched[:top_k]
+    return {
+        "matches": matched,
+        "source": "fallback",
+        "description": description,
+        "total": len(assets),
+    }
+
+
 # ── 单例入口 ─────────────────────────────────────────────────────────────────
 
 
@@ -345,3 +606,11 @@ class AIService:
         count: int = 5,
     ) -> Dict[str, Any]:
         return generate_smart_titles(description, style, count)
+
+    def semantic_match(
+        self,
+        description: str,
+        assets: List[Dict[str, Any]],
+        top_k: int = 0,
+    ) -> Dict[str, Any]:
+        return semantic_match_assets(description, assets, top_k)

tests/unit/test_ai_service.py

@@ -22,7 +22,10 @@ from app.services.ai_service import (  # noqa: E402
     TITLE_STYLES,
     _generate_titles_fallback,
     _parse_titles_from_response,
+    _parse_semantic_match_response,
+    _semantic_match_fallback,
     generate_smart_titles,
+    semantic_match_assets,
 )
 
 
@@ -295,5 +298,230 @@ class TestTitleStyles(unittest.TestCase):
         self.assertIn("informative", TITLE_STYLES)
 
 
+# ── 语义匹配测试 ──────────────────────────────────────────────────────────────
+
+
+class TestSemanticMatchFallback(unittest.TestCase):
+    """降级关键词匹配测试."""
+
+    def _make_assets(self):
+        return [
+            {"id": "a1", "name": "海边日落风景", "tags": ["风景", "海边", "日落"], "description": "美丽的海边日落"},
+            {"id": "a2", "name": "城市夜景航拍", "tags": ["城市", "夜景", "航拍"], "description": "城市夜景航拍素材"},
+            {"id": "a3", "name": "美食制作过程", "tags": ["美食", "烹饪", "教程"], "description": "美食制作教程"},
+        ]
+
+    def test_fallback_returns_sorted_scores(self):
+        """返回按匹配度降序排列."""
+        assets = self._make_assets()
+        result = _semantic_match_fallback("海边日落风景视频", assets)
+        self.assertEqual(len(result), 3)
+        # 第一个应该是海边日落
+        self.assertEqual(result[0]["id"], "a1")
+        self.assertGreater(result[0]["match_score"], result[2]["match_score"])
+
+    def test_fallback_each_has_match_score(self):
+        """每个素材都有 match_score."""
+        assets = self._make_assets()
+        result = _semantic_match_fallback("测试", assets)
+        for item in result:
+            self.assertIn("match_score", item)
+            self.assertGreaterEqual(item["match_score"], 0.0)
+            self.assertLessEqual(item["match_score"], 1.0)
+            self.assertIn("match_reason", item)
+
+    def test_fallback_unrelated_desc_low_scores(self):
+        """完全不相关的描述得分低."""
+        assets = self._make_assets()
+        result = _semantic_match_fallback("篮球比赛运动", assets)
+        # 所有素材得分都应该较低
+        for item in result:
+            self.assertLess(item["match_score"], 0.8)
+
+    def test_fallback_empty_keywords_default_score(self):
+        """无有效关键词时给默认分."""
+        assets = self._make_assets()
+        result = _semantic_match_fallback("a", assets)  # 单字符无有效关键词
+        for item in result:
+            self.assertEqual(item["match_score"], 0.5)
+            self.assertEqual(item["match_reason"], "fallback_default")
+
+    def test_fallback_name_match_higher(self):
+        """名称命中得分更高."""
+        assets = [
+            {"id": "a1", "name": "美食探店vlog", "tags": [], "description": ""},
+            {"id": "a2", "name": "风景视频", "tags": ["美食"], "description": ""},
+        ]
+        result = _semantic_match_fallback("美食", assets)
+        # a1名称含美食，a2标签含美食，名称命中应有额外加分
+        self.assertEqual(result[0]["id"], "a1")
+        self.assertGreater(result[0]["match_score"], result[1]["match_score"])
+
+
+class TestSemanticMatchParsing(unittest.TestCase):
+    """语义匹配返回解析测试."""
+
+    def test_parse_dict_format(self):
+        """解析 {id: score} 格式."""
+        content = json.dumps({"asset1": 0.85, "asset2": 0.62, "asset3": 0.3})
+        result = _parse_semantic_match_response(content, ["asset1", "asset2", "asset3"])
+        self.assertIsNotNone(result)
+        self.assertEqual(len(result), 3)
+        self.assertAlmostEqual(result["asset1"], 0.85)
+
+    def test_parse_matches_list_format(self):
+        """解析 {matches: [...]} 格式."""
+        content = json.dumps({
+            "matches": [
+                {"asset_id": "a1", "score": 0.9},
+                {"asset_id": "a2", "score": 0.7},
+            ]
+        })
+        result = _parse_semantic_match_response(content, ["a1", "a2"])
+        self.assertIsNotNone(result)
+        self.assertAlmostEqual(result["a1"], 0.9)
+        self.assertAlmostEqual(result["a2"], 0.7)
+
+    def test_parse_array_format(self):
+        """解析数组格式."""
+        content = json.dumps([
+            {"id": "x1", "score": 0.5},
+            {"id": "x2", "score": 0.88},
+        ])
+        result = _parse_semantic_match_response(content, ["x1", "x2"])
+        self.assertIsNotNone(result)
+        self.assertAlmostEqual(result["x1"], 0.5)
+
+    def test_parse_score_clamped(self):
+        """分数被限制在0-1."""
+        content = json.dumps({"a1": 1.5, "a2": -0.2})
+        result = _parse_semantic_match_response(content, ["a1", "a2"])
+        self.assertIsNotNone(result)
+        self.assertAlmostEqual(result["a1"], 1.0)
+        self.assertAlmostEqual(result["a2"], 0.0)
+
+    def test_parse_markdown_code_block(self):
+        """解析markdown代码块."""
+        content = "```json\n{\"a1\": 0.7}\n```"
+        result = _parse_semantic_match_response(content, ["a1", "a2"])
+        # 只有1个素材评分，少于一半（需要至少1个，max(1, 2//2)=1）
+        self.assertIsNotNone(result)
+        self.assertAlmostEqual(result["a1"], 0.7)
+
+    def test_parse_empty_returns_none(self):
+        """空内容返回None."""
+        result = _parse_semantic_match_response("", ["a1"])
+        self.assertIsNone(result)
+
+    def test_parse_invalid_json_returns_none(self):
+        """无效JSON返回None."""
+        result = _parse_semantic_match_response("不是json", ["a1", "a2", "a3"])
+        self.assertIsNone(result)
+
+
+class TestSemanticMatchAssets(unittest.TestCase):
+    """semantic_match_assets 集成测试."""
+
+    def _make_assets(self):
+        return [
+            {"id": "a1", "name": "海边日落", "tags": ["风景"], "description": ""},
+            {"id": "a2", "name": "城市夜景", "tags": ["城市"], "description": ""},
+            {"id": "a3", "name": "美食制作", "tags": ["美食"], "description": ""},
+        ]
+
+    def test_fallback_mode_without_api_key(self):
+        """无API Key时走降级."""
+        with patch("app.services.ai_service.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock(
+                DOUBAO_API_KEY="",
+                DOUBAO_MODEL="test",
+                DOUBAO_BASE_URL="https://test.com",
+                DOUBAO_TIMEOUT=30,
+                DOUBAO_MAX_RETRIES=0,
+            )
+            result = semantic_match_assets("海边", self._make_assets())
+            self.assertEqual(result["source"], "fallback")
+            self.assertEqual(result["total"], 3)
+            self.assertEqual(len(result["matches"]), 3)
+
+    def test_empty_assets(self):
+        """空素材列表."""
+        result = semantic_match_assets("test", [])
+        self.assertEqual(result["total"], 0)
+        self.assertEqual(len(result["matches"]), 0)
+
+    def test_top_k_limit(self):
+        """top_k 限制返回数量."""
+        with patch("app.services.ai_service.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock(
+                DOUBAO_API_KEY="",
+                DOUBAO_MODEL="test",
+                DOUBAO_BASE_URL="https://test.com",
+                DOUBAO_TIMEOUT=30,
+                DOUBAO_MAX_RETRIES=0,
+            )
+            result = semantic_match_assets("测试", self._make_assets(), top_k=2)
+            self.assertEqual(len(result["matches"]), 2)
+
+    def test_with_doubao_success(self):
+        """豆包调用成功路径."""
+        with patch("app.services.ai_service.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock(
+                DOUBAO_API_KEY="sk-test-123",
+                DOUBAO_MODEL="test-model",
+                DOUBAO_BASE_URL="https://test.com",
+                DOUBAO_TIMEOUT=30,
+                DOUBAO_MAX_RETRIES=0,
+            )
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.json.return_value = {
+                "choices": [{
+                    "message": {
+                        "content": json.dumps({"a1": 0.9, "a2": 0.5, "a3": 0.2})
+                    }
+                }]
+            }
+            mock_response.raise_for_status = MagicMock()
+
+            with patch("httpx.post", return_value=mock_response):
+                result = semantic_match_assets("风景视频", self._make_assets())
+                self.assertEqual(result["source"], "doubao")
+                self.assertEqual(len(result["matches"]), 3)
+                # 按分数降序，a1最高
+                self.assertEqual(result["matches"][0]["id"], "a1")
+                self.assertAlmostEqual(result["matches"][0]["match_score"], 0.9)
+
+    def test_with_doubao_failure_fallback(self):
+        """豆包调用失败降级."""
+        with patch("app.services.ai_service.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock(
+                DOUBAO_API_KEY="sk-test-123",
+                DOUBAO_MODEL="test-model",
+                DOUBAO_BASE_URL="https://test.com",
+                DOUBAO_TIMEOUT=1,
+                DOUBAO_MAX_RETRIES=0,
+            )
+            with patch("httpx.post", side_effect=Exception("API Error")):
+                result = semantic_match_assets("测试", self._make_assets())
+                self.assertEqual(result["source"], "fallback")
+
+    def test_each_match_has_required_fields(self):
+        """每个匹配结果都有必要字段."""
+        with patch("app.services.ai_service.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock(
+                DOUBAO_API_KEY="",
+                DOUBAO_MODEL="test",
+                DOUBAO_BASE_URL="https://test.com",
+                DOUBAO_TIMEOUT=30,
+                DOUBAO_MAX_RETRIES=0,
+            )
+            result = semantic_match_assets("测试", self._make_assets())
+            for item in result["matches"]:
+                self.assertIn("id", item)
+                self.assertIn("match_score", item)
+                self.assertIn("match_reason", item)
+
+
 if __name__ == "__main__":
     unittest.main()