tests/unit/domain/test_ai_parsing.py

@@ -0,0 +1,414 @@
+"""AI响应解析纯逻辑单测.
+
+覆盖：标题解析（多格式）、语义匹配解析、
+标题降级生成、关键词匹配降级。
+"""
+
+from __future__ import annotations
+
+import random
+from unittest.mock import patch
+
+from packages.domain.ai_parsing import (
+    generate_titles_fallback,
+    keyword_match_fallback,
+    parse_semantic_match_response,
+    parse_titles_from_response,
+)
+
+
+class TestParseTitlesFromResponse:
+    def test_empty_content(self):
+        assert parse_titles_from_response("") == []
+
+    def test_json_array(self):
+        content = '["标题一", "标题二", "标题三"]'
+        result = parse_titles_from_response(content)
+        assert result == ["标题一", "标题二", "标题三"]
+
+    def test_json_array_with_whitespace_items(self):
+        content = '["  标题一  ", "", "标题二"]'
+        result = parse_titles_from_response(content)
+        assert result == ["标题一", "标题二"]
+
+    def test_json_dict_with_titles_key(self):
+        content = '{"titles": ["爆款标题1", "爆款标题2"]}'
+        result = parse_titles_from_response(content)
+        assert result == ["爆款标题1", "爆款标题2"]
+
+    def test_json_code_block(self):
+        content = '```json\n["标题A", "标题B"]\n```'
+        result = parse_titles_from_response(content)
+        assert result == ["标题A", "标题B"]
+
+    def test_json_code_block_with_backticks_only(self):
+        content = '```\n["X", "Y"]\n```'
+        result = parse_titles_from_response(content)
+        assert result == ["X", "Y"]
+
+    def test_numbered_list_dot(self):
+        content = "1. 第一个标题\n2. 第二个标题\n3. 第三个标题"
+        result = parse_titles_from_response(content)
+        assert result == ["第一个标题", "第二个标题", "第三个标题"]
+
+    def test_numbered_list_chinese_comma(self):
+        content = "1、标题甲\n2、标题乙"
+        result = parse_titles_from_response(content)
+        assert result == ["标题甲", "标题乙"]
+
+    def test_numbered_list_parenthesis(self):
+        """右括号格式编号能被去掉，左括号保留（实际行为）."""
+        content = "1) 标题1\n2) 标题2"
+        result = parse_titles_from_response(content)
+        assert result == ["标题1", "标题2"]
+
+    def test_dash_prefix(self):
+        content = "- 标题A\n- 标题B\n- 标题C"
+        result = parse_titles_from_response(content)
+        assert result == ["标题A", "标题B", "标题C"]
+
+    def test_bullet_prefix(self):
+        content = "• 要点一\n• 要点二"
+        result = parse_titles_from_response(content)
+        assert result == ["要点一", "要点二"]
+
+    def test_newline_only(self):
+        content = "标题一\n标题二\n标题三"
+        result = parse_titles_from_response(content)
+        assert result == ["标题一", "标题二", "标题三"]
+
+    def test_quoted_titles(self):
+        content = "\"双引号标题\"\n'单引号标题'\n「中文引号」"
+        result = parse_titles_from_response(content)
+        assert result == ["双引号标题", "单引号标题", "中文引号"]
+
+    def test_skip_empty_lines(self):
+        content = "标题1\n\n标题2\n\n标题3"
+        result = parse_titles_from_response(content)
+        assert result == ["标题1", "标题2", "标题3"]
+
+    def test_filter_long_lines(self):
+        """超过100字符的行被过滤."""
+        long_title = "a" * 150
+        content = f"短标题\n{long_title}\n另一个短标题"
+        result = parse_titles_from_response(content)
+        assert len(result) == 2
+        assert "短标题" in result
+        assert "另一个短标题" in result
+
+    def test_invalid_json_falls_back_to_line_parse(self):
+        content = '["标题1", "标题2", invalid]'  # 非法JSON
+        result = parse_titles_from_response(content)
+        # 会走到按行解析
+        assert len(result) >= 1
+
+    def test_mixed_format_numbered_and_dash(self):
+        content = "1. 第一题\n- 第二题\n2. 第三题"
+        result = parse_titles_from_response(content)
+        assert "第一题" in result
+        assert "第二题" in result
+        assert "第三题" in result
+
+
+class TestParseSemanticMatchResponse:
+    def test_empty_content(self):
+        assert parse_semantic_match_response("", ["a1", "a2"]) is None
+
+    def test_dict_format_asset_id_score(self):
+        content = '{"asset_1": 0.85, "asset_2": 0.6}'
+        result = parse_semantic_match_response(content, ["asset_1", "asset_2"])
+        assert result is not None
+        assert result["asset_1"] == 0.85
+        assert result["asset_2"] == 0.6
+
+    def test_matches_array_format(self):
+        content = '{"matches": [{"asset_id": "a1", "score": 0.9}, {"asset_id": "a2", "score": 0.7}]}'
+        result = parse_semantic_match_response(content, ["a1", "a2"])
+        assert result is not None
+        assert result["a1"] == 0.9
+        assert result["a2"] == 0.7
+
+    def test_list_format(self):
+        content = '[{"asset_id": "x", "score": 0.5}, {"asset_id": "y", "score": 0.8}]'
+        result = parse_semantic_match_response(content, ["x", "y"])
+        assert result is not None
+        assert result["x"] == 0.5
+        assert result["y"] == 0.8
+
+    def test_id_alias_in_matches(self):
+        """matches中用id替代asset_id."""
+        content = '{"matches": [{"id": "a1", "score": 0.75}]}'
+        result = parse_semantic_match_response(content, ["a1", "a2"])
+        assert result is not None
+        assert result["a1"] == 0.75
+
+    def test_score_clamped_to_0_1(self):
+        """分数超出0-1范围会被截断."""
+        content = '{"a1": -0.5, "a2": 1.5, "a3": 0.5}'
+        result = parse_semantic_match_response(content, ["a1", "a2", "a3"])
+        assert result is not None
+        assert result["a1"] == 0.0
+        assert result["a2"] == 1.0
+        assert result["a3"] == 0.5
+
+    def test_score_int_converted_to_float(self):
+        content = '{"a1": 1, "a2": 0}'
+        result = parse_semantic_match_response(content, ["a1", "a2"])
+        assert result is not None
+        assert result["a1"] == 1.0
+        assert result["a2"] == 0.0
+
+    def test_json_code_block(self):
+        content = '```json\n{"a1": 0.9, "a2": 0.8}\n```'
+        result = parse_semantic_match_response(content, ["a1", "a2"])
+        assert result is not None
+        assert result["a1"] == 0.9
+
+    def test_half_threshold_with_asset_ids(self):
+        """提供asset_ids时，至少一半有评分才算成功."""
+        # 4个assets，只有1个有评分（<2）→ 失败
+        content = '{"a1": 0.9}'
+        result = parse_semantic_match_response(content, ["a1", "a2", "a3", "a4"])
+        assert result is None
+
+    def test_half_threshold_passes(self):
+        # 4个assets，2个有评分（=一半）→ 成功
+        content = '{"a1": 0.9, "a2": 0.8}'
+        result = parse_semantic_match_response(content, ["a1", "a2", "a3", "a4"])
+        assert result is not None
+
+    def test_no_asset_ids_returns_any_result(self):
+        content = '{"x1": 0.7}'
+        result = parse_semantic_match_response(content, [])
+        assert result is not None
+        assert result["x1"] == 0.7
+
+    def test_no_asset_ids_empty_result_returns_none(self):
+        content = "{}"
+        result = parse_semantic_match_response(content, [])
+        assert result is None
+
+    def test_invalid_json_returns_none(self):
+        content = "not json at all"
+        result = parse_semantic_match_response(content, ["a1"])
+        assert result is None
+
+    def test_non_numeric_values_ignored(self):
+        content = '{"a1": "high", "a2": 0.8}'
+        result = parse_semantic_match_response(content, ["a1", "a2"])
+        assert result is not None
+        assert "a1" not in result
+        assert result["a2"] == 0.8
+
+    def test_single_asset_id_needs_at_least_1(self):
+        """1个asset，需要至少max(1, 0)=1个评分."""
+        content = '{"a1": 0.5}'
+        result = parse_semantic_match_response(content, ["a1"])
+        assert result is not None
+        assert result["a1"] == 0.5
+
+
+class TestGenerateTitlesFallback:
+    def test_basic_generation(self):
+        with patch.object(random, "shuffle", lambda x: None):  # 禁用shuffle
+            result = generate_titles_fallback(
+                "美食 探店 川菜",
+                {"examples": ["必看攻略", "绝密技巧"]},
+                count=3,
+            )
+        assert len(result) == 3
+        assert all(isinstance(t, str) for t in result)
+        assert all(len(t) > 0 for t in result)
+
+    def test_count_limited_by_templates(self):
+        """最多10个模板."""
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "测试",
+                {"examples": ["例1", "例2"]},
+                count=20,
+            )
+        assert len(result) == 10  # 模板总数上限
+
+    def test_default_count(self):
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "科技 产品",
+                {"examples": ["测试标题", "另一个例子"]},
+            )
+        assert len(result) == 5
+
+    def test_empty_description_uses_default_keyword(self):
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "   ",
+                {"examples": ["例A", "例B"]},
+                count=1,
+            )
+        assert "精彩内容" in result[0]
+
+    def test_keyword_extracted_from_description(self):
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "Python编程入门教程",
+                {"examples": ["入门", "技巧"]},
+                count=5,
+            )
+        # 第一个关键词应该出现在某些标题中
+        assert any("Python编程入门教程" in t for t in result)
+
+    def test_examples_truncated(self):
+        """第一个example超过10字符会被截断."""
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "美食",
+                {"examples": ["这是一个非常长的例子超过十个字", "第二个例子"]},
+                count=1,
+            )
+        # 第一个标题应该包含截断的example + "..."
+        assert "..." in result[0]
+
+    def test_no_examples_uses_default(self):
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "健身",
+                {"examples": []},
+                count=2,
+            )
+        assert len(result) == 2
+        assert "必看" in result[0]  # 默认example_0
+
+    def test_second_example_default(self):
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "健身",
+                {"examples": ["只有一个"]},
+                count=3,
+            )
+        # 第二个标题应该包含默认的"你不知道的事"
+        assert any("你不知道的事" in t for t in result)
+
+    def test_single_word_keyword(self):
+        """单字会被过滤掉，使用默认关键词."""
+        with patch.object(random, "shuffle", lambda x: None):
+            result = generate_titles_fallback(
+                "a b c",
+                {"examples": ["例"]},
+                count=1,
+            )
+        # 所有词都是1个字符，应该用默认关键词
+        assert "精彩内容" in result[0]
+
+
+class TestKeywordMatchFallback:
+    def test_basic_matching(self):
+        assets = [
+            {"id": "a1", "name": "美食探店视频", "tags": ["美食", "探店"], "description": "成都美食"},
+            {"id": "a2", "name": "科技产品评测", "tags": ["科技"], "description": "手机评测"},
+            {"id": "a3", "name": "旅行Vlog", "tags": ["旅行"], "description": "日本旅行"},
+        ]
+        result = keyword_match_fallback("美食 探店 成都", assets)
+        assert len(result) == 3
+        # 美食相关的应该排第一
+        assert result[0]["id"] == "a1"
+        assert 0 < result[0]["match_score"] <= 1.0
+
+    def test_score_between_0_and_1(self):
+        assets = [{"id": "a1", "name": "测试素材", "tags": [], "description": ""}]
+        result = keyword_match_fallback("完全不相关的关键词", assets)
+        assert 0 <= result[0]["match_score"] <= 1
+
+    def test_no_keywords_default_score(self):
+        """描述中没有有效关键词时，所有素材0.5分."""
+        assets = [
+            {"id": "a1", "name": "素材1", "tags": [], "description": ""},
+            {"id": "a2", "name": "素材2", "tags": [], "description": ""},
+        ]
+        result = keyword_match_fallback("   ", assets)  # 空描述
+        assert len(result) == 2
+        assert result[0]["match_score"] == 0.5
+        assert result[0]["match_reason"] == "fallback_default"
+
+    def test_sorted_descending(self):
+        assets = [
+            {"id": "a_low", "name": "不相关", "tags": [], "description": ""},
+            {"id": "a_high", "name": "美食推荐", "tags": ["美食"], "description": "美食攻略"},
+        ]
+        result = keyword_match_fallback("美食 推荐", assets)
+        assert result[0]["id"] == "a_high"
+        assert result[0]["match_score"] > result[1]["match_score"]
+
+    def test_match_reason_keyword(self):
+        assets = [{"id": "a1", "name": "测试", "tags": [], "description": ""}]
+        result = keyword_match_fallback("测试关键词", assets)
+        assert result[0]["match_reason"] == "fallback_keyword"
+
+    def test_name_bonus(self):
+        """名称命中应该有额外加分."""
+        assets = [
+            {
+                "id": "a1",
+                "name": "完全不相关的名字",
+                "tags": [],
+                "description": "美食教程",  # 描述里有关键词
+            },
+            {
+                "id": "a2",
+                "name": "美食分享",  # 名称里有关键词
+                "tags": [],
+                "description": "",  # 描述里没有
+            },
+        ]
+        result = keyword_match_fallback("美食", assets)
+        # 名称命中的a2应该分数更高（name bonus）
+        assert result[0]["id"] == "a2"
+
+    def test_empty_assets(self):
+        result = keyword_match_fallback("美食", [])
+        assert result == []
+
+    def test_asset_dict_not_mutated(self):
+        """不修改原始asset字典."""
+        asset = {"id": "a1", "name": "测试", "tags": []}
+        original = dict(asset)
+        keyword_match_fallback("测试", [asset])
+        assert asset == original
+
+    def test_chinese_keywords_used(self):
+        """中文2-4字片段应该被用作关键词."""
+        assets = [
+            {"id": "a1", "name": "编程入门", "tags": [], "description": ""},
+            {"id": "a2", "name": "美食推荐", "tags": [], "description": ""},
+        ]
+        result = keyword_match_fallback("编程入门教程", assets)
+        assert result[0]["id"] == "a1"
+        assert result[0]["match_score"] > 0
+
+    def test_english_keywords_used(self):
+        """英文3字符以上单词应该被用作关键词."""
+        assets = [
+            {"id": "a1", "name": "Python tutorial", "tags": [], "description": ""},
+            {"id": "a2", "name": "Java course", "tags": [], "description": ""},
+        ]
+        result = keyword_match_fallback("python programming", assets)
+        assert result[0]["id"] == "a1"
+        assert result[0]["match_score"] > 0
+
+    def test_score_is_rounded_to_3_decimals(self):
+        assets = [{"id": "a1", "name": "测试素材", "tags": [], "description": ""}]
+        result = keyword_match_fallback("测试关键词", assets)
+        # 3位小数
+        assert len(str(result[0]["match_score"]).split(".")[-1]) <= 3
+
+    def test_perfect_match_score(self):
+        assets = [
+            {
+                "id": "a1",
+                "name": "美食探店推荐",
+                "tags": ["美食", "探店", "推荐"],
+                "description": "美食探店推荐视频",
+            }
+        ]
+        result = keyword_match_fallback("美食 探店 推荐", assets)
+        assert result[0]["match_score"] <= 1.0
+        assert result[0]["match_score"] > 0.5  # 应该有较高分数