From 6b8c07de1676dacb9e896f78a9c0b5faa73be95c Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Mon, 27 Jul 2026 00:26:55 +0800
Subject: [PATCH] =?UTF-8?q?refactor(wave108):=20=E6=8A=BD=E7=A6=BBai=5Fpar?=
 =?UTF-8?q?sing=E7=BA=AF=E9=80=BB=E8=BE=91=E6=A8=A1=E5=9D=97=20+=2044?=
 =?UTF-8?q?=E5=8D=95=E6=B5=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 从ai_service.py抽离4个解析/降级函数到packages/domain/ai_parsing.py
- ai_service保留薄包装函数，完全向后兼容
- 新增44个纯逻辑单测，覆盖标题解析、语义匹配解析、标题降级、关键词匹配
- ai_service.py: 536→351行 (-185行, -35%)
---
 apps/api/app/services/ai_service.py | 213 ++-----------------
 packages/domain/ai_parsing.py       | 249 ++++++++++++++++++++++
 tests/unit/test_ai_parsing.py       | 315 ++++++++++++++++++++++++++++
 3 files changed, 578 insertions(+), 199 deletions(-)
 create mode 100755 packages/domain/ai_parsing.py
 create mode 100755 tests/unit/test_ai_parsing.py

diff --git a/apps/api/app/services/ai_service.py b/apps/api/app/services/ai_service.py
index 0d1bca0a1..02650f709 100755
--- a/apps/api/app/services/ai_service.py
+++ b/apps/api/app/services/ai_service.py
@@ -14,10 +14,14 @@ from __future__ import annotations
 
 import json
 import logging
-import math
-import random
 from typing import Any, Dict, List, Optional
 
+from packages.domain.ai_parsing import (
+    generate_titles_fallback as _generate_titles_fallback_base,
+    keyword_match_fallback as _semantic_match_fallback_base,
+    parse_semantic_match_response as _parse_semantic_match_base,
+    parse_titles_from_response as _parse_titles_from_response,
+)
 from packages.shared.ai_client import get_doubao_client
 
 logger = logging.getLogger(__name__)
@@ -64,85 +68,9 @@ def _generate_titles_fallback(
     style: str = "viral",
     count: int = 5,
 ) -> List[str]:
-    """本地降级：基于模板规则生成标题.
-
-    当豆包 API 不可用或调用失败时使用，保证接口始终有返回。
-    """
+    """本地降级：基于模板规则生成标题（薄包装，转发到 ai_parsing 模块）."""
     style_info = TITLE_STYLES.get(style, TITLE_STYLES["viral"])
-    examples = style_info["examples"]
-
-    # 从描述中提取关键词（取前几个词）
-    keywords = [w for w in description.strip().split() if len(w) > 1][:3]
-    keyword = keywords[0] if keywords else "精彩内容"
-
-    # 基于模板生成
-    templates = [
-        f"「{keyword}」{examples[0][:10]}...",
-        f"{keyword}：{examples[1]}",
-        f"关于{keyword}，你不知道的3件事",
-        f"{keyword}入门指南，新手必看",
-        f"深度解析：{keyword}背后的秘密",
-        f"{keyword}怎么做？手把手教你",
-        f"干货分享 | {keyword}全攻略",
-        f"建议收藏：{keyword}实用技巧",
-        f"{keyword}避坑指南，别再踩雷了",
-        f"一分钟搞懂{keyword}",
-    ]
-
-    random.shuffle(templates)
-    return templates[: min(count, len(templates))]
-
-
-def _parse_titles_from_response(content: str) -> List[str]:
-    """从模型返回中解析标题列表.
-
-    支持多种返回格式：
-    - JSON 数组: ["标题1", "标题2"]
-    - 编号列表: 1. 标题1 / 2. 标题2
-    - 换行分隔: 标题1\n标题2
-    - 带破折号: - 标题1
-    """
-    if not content:
-        return []
-
-    # 尝试解析 JSON
-    try:
-        # 清理可能的 markdown 代码块标记
-        cleaned = content.strip()
-        if cleaned.startswith("```"):
-            cleaned = cleaned.strip("`")
-            if cleaned.lower().startswith("json"):
-                cleaned = cleaned[4:]
-            cleaned = cleaned.strip()
-
-        data = json.loads(cleaned)
-        if isinstance(data, list):
-            return [str(item).strip() for item in data if str(item).strip()]
-        if isinstance(data, dict) and "titles" in data:
-            titles = data["titles"]
-            if isinstance(titles, list):
-                return [str(t).strip() for t in titles if str(t).strip()]
-    except (json.JSONDecodeError, ValueError):
-        pass
-
-    # 尝试按行解析
-    titles: List[str] = []
-    for line in content.strip().split("\n"):
-        line = line.strip()
-        if not line:
-            continue
-        # 去掉编号前缀 "1. " "1、" "（1）"
-        import re
-
-        line = re.sub(r"^[\d]+[\.、\)）]\s*", "", line)
-        # 去掉破折号前缀 "- " "• "
-        line = re.sub(r"^[-•·]\s*", "", line)
-        # 去掉引号
-        line = line.strip('"').strip("'").strip("「」")
-        if line and len(line) < 100:  # 过滤过长的行
-            titles.append(line)
-
-    return titles
+    return _generate_titles_fallback_base(description, style_info, count)
 
 
 def generate_smart_titles(
@@ -241,132 +169,19 @@ 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
+    """本地降级：基于关键词的简单匹配（薄包装，转发到 ai_parsing 模块）."""
+    return _semantic_match_fallback_base(description, assets)
 
 
 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:
+    """从模型返回中解析素材匹配度（薄包装，转发到 ai_parsing 模块）."""
+    result = _parse_semantic_match_base(content, asset_ids)
+    if result is None:
         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
+    return dict(result)
 
 
 def semantic_match_assets(
diff --git a/packages/domain/ai_parsing.py b/packages/domain/ai_parsing.py
new file mode 100755
index 000000000..fd401a0e9
--- /dev/null
+++ b/packages/domain/ai_parsing.py
@@ -0,0 +1,249 @@
+"""AI 响应解析纯逻辑模块.
+
+抽离自 ai_service.py 的解析函数，方便单测覆盖，同时保持向后兼容。
+包括：
+- 标题列表解析（JSON/编号/换行/破折号格式）
+- 语义匹配结果解析（多种JSON格式）
+- 标题降级生成
+- 关键词匹配降级
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import random
+import re
+from typing import Any
+
+# ── 标题解析 ──────────────────────────────────────────────────────────────────
+
+
+def parse_titles_from_response(content: str) -> list[str]:
+    """从模型返回中解析标题列表.
+
+    支持多种返回格式：
+    - JSON 数组: ["标题1", "标题2"]
+    - 编号列表: 1. 标题1 / 2. 标题2
+    - 换行分隔: 标题1\n标题2
+    - 带破折号: - 标题1
+    """
+    if not content:
+        return []
+
+    # 尝试解析 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)
+        if isinstance(data, list):
+            return [str(item).strip() for item in data if str(item).strip()]
+        if isinstance(data, dict) and "titles" in data:
+            titles = data["titles"]
+            if isinstance(titles, list):
+                return [str(t).strip() for t in titles if str(t).strip()]
+    except (json.JSONDecodeError, ValueError):
+        pass
+
+    # 尝试按行解析
+    titles: list[str] = []
+    for line in content.strip().split("\n"):
+        line = line.strip()
+        if not line:
+            continue
+        # 去掉编号前缀 "1. " "1、" "（1）"
+        line = re.sub(r"^[\d]+[\.、\)）]\s*", "", line)
+        # 去掉破折号前缀 "- " "• "
+        line = re.sub(r"^[-•·]\s*", "", line)
+        # 去掉引号
+        line = line.strip('"').strip("'").strip("「」")
+        if line and len(line) < 100:  # 过滤过长的行
+            titles.append(line)
+
+    return titles
+
+
+# ── 语义匹配解析 ────────────────────────────────────────────────────────────
+
+
+def parse_semantic_match_response(
+    content: str,
+    asset_ids: list[str],
+) -> dict[str, float] | None:
+    """从模型返回中解析素材匹配度.
+
+    期望格式：JSON 对象 {asset_id: score} 或 {"matches": [{asset_id, score}]}
+    score 范围 0-1，自动截断到 [0, 1]。
+    """
+    if not content:
+        return None
+
+    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 asset_ids and len(result) >= max(1, len(asset_ids) // 2):
+            return result
+        # 没有 asset_ids 时，只要有结果就返回
+        if not asset_ids and result:
+            return result
+
+    except (json.JSONDecodeError, ValueError):
+        pass
+
+    return None
+
+
+# ── 标题降级生成 ─────────────────────────────────────────────────────────────
+
+
+def generate_titles_fallback(
+    description: str,
+    style_info: dict[str, Any],
+    count: int = 5,
+) -> list[str]:
+    """本地降级：基于模板规则生成标题.
+
+    Args:
+        description: 视频内容描述
+        style_info: 标题风格配置 {"name": ..., "examples": [...]}
+        count: 生成数量
+    """
+    examples = style_info.get("examples", [])
+
+    # 从描述中提取关键词（取前几个词）
+    keywords = [w for w in description.strip().split() if len(w) > 1][:3]
+    keyword = keywords[0] if keywords else "精彩内容"
+
+    # 基于模板生成
+    example_0 = examples[0][:10] + "..." if examples else "必看"
+    example_1 = examples[1] if len(examples) > 1 else "你不知道的事"
+
+    templates = [
+        f"「{keyword}」{example_0}",
+        f"{keyword}：{example_1}",
+        f"关于{keyword}，你不知道的3件事",
+        f"{keyword}入门指南，新手必看",
+        f"深度解析：{keyword}背后的秘密",
+        f"{keyword}怎么做？手把手教你",
+        f"干货分享 | {keyword}全攻略",
+        f"建议收藏：{keyword}实用技巧",
+        f"{keyword}避坑指南，别再踩雷了",
+        f"一分钟搞懂{keyword}",
+    ]
+
+    random.shuffle(templates)
+    return templates[: min(count, len(templates))]
+
+
+# ── 关键词匹配降级 ───────────────────────────────────────────────────────────
+
+
+def keyword_match_fallback(
+    description: str,
+    assets: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """本地降级：基于关键词的简单匹配.
+
+    计算描述中的关键词与素材名称/标签/描述的重叠度，
+    作为匹配度评分。0-1分。
+    """
+    # 提取关键词（中文按2字以上片段，英文按单词）
+    desc = description.lower()
+    keywords: set[str] = 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:
+        # 没有关键词时给所有素材中等分数
+        results = []
+        for asset in assets:
+            new_asset = dict(asset)
+            new_asset["match_score"] = 0.5
+            new_asset["match_reason"] = "fallback_default"
+            results.append(new_asset)
+        return results
+
+    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: list[str] = []
+        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)
+
+        new_asset = dict(asset)
+        new_asset["match_score"] = score
+        new_asset["match_reason"] = "fallback_keyword"
+        results.append(new_asset)
+
+    # 按匹配度降序
+    results.sort(key=lambda x: x["match_score"], reverse=True)
+    return results
diff --git a/tests/unit/test_ai_parsing.py b/tests/unit/test_ai_parsing.py
new file mode 100755
index 000000000..8f91f6f53
--- /dev/null
+++ b/tests/unit/test_ai_parsing.py
@@ -0,0 +1,315 @@
+"""ai_parsing 模块单测 — 纯逻辑."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.ai_parsing import (
+    generate_titles_fallback,
+    keyword_match_fallback,
+    parse_semantic_match_response,
+    parse_titles_from_response,
+)
+
+# ── parse_titles_from_response 测试 ──────────────────────────────────────────
+
+
+class TestParseTitlesJsonArray:
+    def test_simple_json_array(self):
+        result = parse_titles_from_response('["标题1", "标题2", "标题3"]')
+        assert result == ["标题1", "标题2", "标题3"]
+
+    def test_json_array_with_empty_strings_skipped(self):
+        result = parse_titles_from_response('["标题1", "", "标题2"]')
+        assert result == ["标题1", "标题2"]
+
+    def test_json_dict_with_titles_key(self):
+        result = parse_titles_from_response('{"titles": ["a", "b", "c"]}')
+        assert result == ["a", "b", "c"]
+
+    def test_json_with_markdown_code_block(self):
+        content = '```json\n["标题1", "标题2"]\n```'
+        result = parse_titles_from_response(content)
+        assert result == ["标题1", "标题2"]
+
+    def test_json_with_backticks_no_lang(self):
+        content = '```\n["标题1", "标题2"]\n```'
+        result = parse_titles_from_response(content)
+        assert result == ["标题1", "标题2"]
+
+    def test_none_returns_empty(self):
+        assert parse_titles_from_response(None) == []  # type: ignore
+
+    def test_empty_string_returns_empty(self):
+        assert parse_titles_from_response("") == []
+
+
+class TestParseTitlesNumberedList:
+    def test_dot_numbered(self):
+        content = "1. 第一个标题\n2. 第二个标题\n3. 第三个标题"
+        result = parse_titles_from_response(content)
+        assert len(result) == 3
+        assert result[0] == "第一个标题"
+        assert result[1] == "第二个标题"
+
+    def test_chinese_period_numbered(self):
+        content = "1、第一个标题\n2、第二个标题"
+        result = parse_titles_from_response(content)
+        assert result == ["第一个标题", "第二个标题"]
+
+    def test_parentheses_numbered(self):
+        content = "1) 第一个标题\n2) 第二个标题"
+        result = parse_titles_from_response(content)
+        assert result == ["第一个标题", "第二个标题"]
+
+    def test_chinese_paren_numbered(self):
+        # 原实现只支持半角括号，全角括号保留原样（不影响实际使用）
+        content = "（1）第一个标题\n（2）第二个标题"
+        result = parse_titles_from_response(content)
+        assert len(result) == 2
+
+
+class TestParseTitlesDash:
+    def test_dash_prefix(self):
+        content = "- 标题一\n- 标题二\n- 标题三"
+        result = parse_titles_from_response(content)
+        assert len(result) == 3
+        assert result[0] == "标题一"
+
+    def test_bullet_prefix(self):
+        content = "• 标题一\n• 标题二"
+        result = parse_titles_from_response(content)
+        assert len(result) == 2
+        assert result[0] == "标题一"
+
+
+class TestParseTitlesQuoted:
+    def test_strips_quotes(self):
+        content = '1. "带引号的标题"\n2. 正常标题'
+        result = parse_titles_from_response(content)
+        assert "带引号的标题" in result
+
+    def test_strips_chinese_quotes(self):
+        content = "1. 「中文引号标题」\n2. 正常标题"
+        result = parse_titles_from_response(content)
+        assert "中文引号标题" in result
+
+
+class TestParseTitlesEdgeCases:
+    def test_skips_empty_lines(self):
+        content = "标题一\n\n标题二\n\n标题三"
+        result = parse_titles_from_response(content)
+        assert len(result) == 3
+
+    def test_filters_long_lines(self):
+        long_title = "A" * 150
+        content = f"短标题\n{long_title}\n另一个短标题"
+        result = parse_titles_from_response(content)
+        assert len(result) == 2
+        assert long_title not in result
+
+    def test_invalid_json_falls_back_to_lines(self):
+        content = "标题1\n标题2\n标题3"
+        result = parse_titles_from_response(content)
+        assert result == ["标题1", "标题2", "标题3"]
+
+
+# ── parse_semantic_match_response 测试 ──────────────────────────────────────
+
+
+class TestParseSemanticMatchDictFormat:
+    def test_simple_dict(self):
+        asset_ids = ["a1", "a2", "a3"]
+        content = '{"a1": 0.8, "a2": 0.6, "a3": 0.9}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert result["a1"] == 0.8
+        assert result["a2"] == 0.6
+        assert result["a3"] == 0.9
+
+    def test_score_clamped_to_0_1(self):
+        asset_ids = ["a1", "a2"]
+        content = '{"a1": 1.5, "a2": -0.5}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert result["a1"] == 1.0
+        assert result["a2"] == 0.0
+
+
+class TestParseSemanticMatchMatchesFormat:
+    def test_matches_array(self):
+        asset_ids = ["a1", "a2"]
+        content = '{"matches": [{"asset_id": "a1", "score": 0.8}, {"asset_id": "a2", "score": 0.6}]}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert result["a1"] == 0.8
+        assert result["a2"] == 0.6
+
+    def test_matches_with_id_key(self):
+        asset_ids = ["a1", "a2"]
+        content = '{"matches": [{"id": "a1", "score": 0.7}, {"id": "a2", "score": 0.5}]}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert result["a1"] == 0.7
+
+
+class TestParseSemanticMatchArrayFormat:
+    def test_array_of_objects(self):
+        asset_ids = ["a1", "a2", "a3"]
+        content = (
+            '[{"asset_id": "a1", "score": 0.8}, {"asset_id": "a2", "score": 0.6}, {"asset_id": "a3", "score": 0.3}]'
+        )
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert len(result) == 3
+
+
+class TestParseSemanticMatchEdgeCases:
+    def test_empty_content_returns_none(self):
+        assert parse_semantic_match_response("", ["a1"]) is None
+
+    def test_invalid_json_returns_none(self):
+        assert parse_semantic_match_response("not json", ["a1"]) is None
+
+    def test_less_than_half_returns_none(self):
+        asset_ids = ["a1", "a2", "a3", "a4", "a5"]
+        # 只返回1个，少于 5//2=2，应该返回 None
+        content = '{"a1": 0.8}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is None
+
+    def test_at_least_half_returns_result(self):
+        asset_ids = ["a1", "a2", "a3", "a4", "a5"]
+        # 返回3个，>= 5//2=2
+        content = '{"a1": 0.8, "a2": 0.7, "a3": 0.6}'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert len(result) == 3
+
+    def test_markdown_code_block(self):
+        asset_ids = ["a1", "a2"]
+        content = '```json\n{"a1": 0.8, "a2": 0.6}\n```'
+        result = parse_semantic_match_response(content, asset_ids)
+        assert result is not None
+        assert result["a1"] == 0.8
+
+    def test_no_asset_ids_returns_result_if_any(self):
+        content = '{"a1": 0.8, "a2": 0.6}'
+        result = parse_semantic_match_response(content, [])
+        assert result is not None
+        assert len(result) == 2
+
+    def test_single_asset_id_needs_at_least_1(self):
+        # max(1, 1//2) = max(1, 0) = 1
+        content = '{"a1": 0.8}'
+        result = parse_semantic_match_response(content, ["a1"])
+        assert result is not None
+
+
+# ── generate_titles_fallback 测试 ────────────────────────────────────────────
+
+
+class TestGenerateTitlesFallback:
+    def test_returns_requested_count(self):
+        style = {"examples": ["例1", "例2", "例3"]}
+        result = generate_titles_fallback("测试描述 关键词", style, count=5)
+        assert len(result) == 5
+
+    def test_uses_keyword_from_description(self):
+        style = {"examples": ["例1", "例2"]}
+        result = generate_titles_fallback("美食 探店 打卡", style, count=3)
+        # 第一个关键词是"美食"
+        assert any("美食" in t for t in result)
+
+    def test_no_keywords_uses_default(self):
+        style = {"examples": ["例1", "例2"]}
+        result = generate_titles_fallback("", style, count=3)
+        assert any("精彩内容" in t for t in result)
+
+    def test_count_limited_by_templates(self):
+        style = {"examples": ["例1", "例2"]}
+        result = generate_titles_fallback("测试", style, count=100)
+        assert len(result) <= 10  # 模板只有10个
+
+    def test_all_titles_are_strings(self):
+        style = {"examples": ["例1", "例2"]}
+        result = generate_titles_fallback("测试", style, count=5)
+        assert all(isinstance(t, str) and t for t in result)
+
+    def test_empty_examples(self):
+        style = {"examples": []}
+        result = generate_titles_fallback("测试", style, count=3)
+        assert len(result) == 3
+        assert all(isinstance(t, str) for t in result)
+
+
+# ── keyword_match_fallback 测试 ──────────────────────────────────────────────
+
+
+class TestKeywordMatchFallback:
+    def test_basic_matching(self):
+        assets = [
+            {"id": "1", "name": "美食探店视频", "tags": ["美食", "探店"], "description": "好吃的"},
+            {"id": "2", "name": "旅行vlog", "tags": ["旅行", "风景"], "description": "出去玩"},
+        ]
+        result = keyword_match_fallback("美食探店 好吃的美食", assets)
+        assert len(result) == 2
+        # 第一个应该是美食相关的
+        assert result[0]["id"] == "1"
+        assert result[0]["match_score"] >= result[1]["match_score"]
+
+    def test_returns_match_score_and_reason(self):
+        assets = [{"id": "1", "name": "测试素材", "tags": [], "description": ""}]
+        result = keyword_match_fallback("美食", assets)
+        assert len(result) == 1
+        assert "match_score" in result[0]
+        assert "match_reason" in result[0]
+        assert 0.0 <= result[0]["match_score"] <= 1.0
+
+    def test_no_keywords_default_score(self):
+        assets = [
+            {"id": "1", "name": "素材1", "tags": [], "description": ""},
+            {"id": "2", "name": "素材2", "tags": [], "description": ""},
+        ]
+        # 单个字符不算关键词
+        result = keyword_match_fallback("a", assets)
+        assert len(result) == 2
+        assert all(r["match_score"] == 0.5 for r in result)
+        assert all(r["match_reason"] == "fallback_default" for r in result)
+
+    def test_sorted_by_score_descending(self):
+        assets = [
+            {"id": "low", "name": "无关素材", "tags": [], "description": ""},
+            {"id": "high", "name": "美食视频", "tags": ["美食"], "description": "美食分享"},
+        ]
+        result = keyword_match_fallback("美食分享", assets)
+        assert result[0]["id"] == "high"
+        assert result[0]["match_score"] > result[1]["match_score"]
+
+    def test_does_not_modify_original_assets(self):
+        original = {"id": "1", "name": "测试", "tags": [], "description": ""}
+        assets = [dict(original)]
+        keyword_match_fallback("美食", assets)
+        assert "match_score" not in assets[0]
+
+    def test_name_matches_higher_score(self):
+        assets = [
+            {"id": "name_match", "name": "美食教程", "tags": [], "description": "内容"},
+            {"id": "desc_match", "name": "视频1", "tags": [], "description": "美食教程内容"},
+        ]
+        result = keyword_match_fallback("美食教程", assets)
+        # 名称命中应该加分更多
+        name_idx = next(i for i, r in enumerate(result) if r["id"] == "name_match")
+        desc_idx = next(i for i, r in enumerate(result) if r["id"] == "desc_match")
+        assert name_idx < desc_idx
+
+    def test_empty_assets_returns_empty(self):
+        result = keyword_match_fallback("美食", [])
+        assert result == []
+
+    def test_score_is_rounded_to_3_decimals(self):
+        assets = [{"id": "1", "name": "测试素材", "tags": [], "description": "内容描述"}]
+        result = keyword_match_fallback("美食探店旅行", assets)
+        score = result[0]["match_score"]
+        # 验证是3位小数
+        assert round(score, 3) == score
-- 
2.54.0