test(14-04): 添加长文本分块处理测试

- 添加 TestSplitIntoChunks 测试类：测试文本分块功能 - 添加 TestLongTextChunking 测试类：测试长文本自动分块分段 - 添加 TestChunkIndexCorrection 测试类：测试分块索引修正 - 添加 TestChunkFallback 测试类：测试单个分块失败时降级到规则分段测试覆盖： - 分块边界条件（空文本、短文本、长文本） - 句子边界分割（。！？\n） - 索引修正（start_index/end_index 相对于原文） - 分块失败降级 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-23 22:41:55 +08:00
parent 88302adb9d
commit ee5062ca9c
1 changed files with 333 additions and 0 deletions
@@ -657,3 +657,336 @@ class TestExtractJSON:
        response = '这不包含任何 JSON 内容。'
        with pytest.raises(ValueError, match="未找到 JSON 内容"):
            extract_json_from_llm_response(response)
+
+
+# ============================================================================
+# Test: 长文本分块处理 (Plan 14-04)
+# ============================================================================
+
+class TestSplitIntoChunks:
+    """测试文本分块功能（Plan 14-04 - Task 14-04-01）"""
+
+    def test_split_into_chunks_basic(self):
+        """测试基本的文本分块功能"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        # 创建一个包含多个句子的文本（约 6000 字符）
+        text = "这是第一句。" * 1000 + "这是第二句。" * 1000 + "这是第三句。" * 1000
+
+        chunks = strategy._split_into_chunks(text, max_chunk_size=5000)
+
+        # 验证分块数量
+        assert len(chunks) > 1
+
+        # 验证每个分块不超过最大字符数
+        for chunk_text, offset in chunks:
+            assert len(chunk_text) <= 5000
+
+        # 验证偏移量是递增的
+        offsets = [offset for _, offset in chunks]
+        assert offsets == sorted(offsets)
+
+        # 验证所有分块拼接后等于原文
+        reconstructed = ''.join(chunk_text for chunk_text, _ in chunks)
+        assert reconstructed == text
+
+    def test_split_into_chunks_short_text(self):
+        """测试短文本不分块"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        # 短文本（< 5000 字符）
+        text = "这是短文本。" * 100
+
+        chunks = strategy._split_into_chunks(text, max_chunk_size=5000)
+
+        # 验证只有一个分块
+        assert len(chunks) == 1
+
+        # 验证内容不变
+        chunk_text, offset = chunks[0]
+        assert chunk_text == text
+        assert offset == 0
+
+    def test_split_into_chunks_sentence_boundaries(self):
+        """测试分块在句子边界分割"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        # 创建包含不同标点符号的文本
+        text = "第一句。第二句！第三句？第四句\n" * 1000
+
+        chunks = strategy._split_into_chunks(text, max_chunk_size=5000)
+
+        # 验证每个分块在句子边界结束（以标点符号或换行符结尾）
+        for chunk_text, offset in chunks:
+            # 分块应该以句子结束字符结尾
+            assert chunk_text[-1] in '。！？\n'
+
+    def test_split_into_chunks_empty_text(self):
+        """测试空文本分块"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        chunks = strategy._split_into_chunks("", max_chunk_size=5000)
+
+        # 空文本应该返回空列表（没有句子可以分块）
+        assert len(chunks) == 0
+
+
+class TestLongTextChunking:
+    """测试长文本自动分块分段（Plan 14-04）"""
+
+    def test_long_text_triggers_chunking(self, caplog):
+        """测试长文本（>5000字符）自动触发分块处理"""
+        import logging
+        import json
+
+        # 创建模拟 AI 客户端
+        mock_client = Mock()
+        mock_client.api_key = "test-key"
+
+        # 创建长文本（>5000 字符）
+        long_text = "这是一个长句。" * 1000
+
+        # 计算会分成几个分块
+        expected_chunks = (len(long_text) + 4999) // 5000
+
+        # 为每个分块设置响应
+        responses = []
+        for i in range(expected_chunks):
+            chunk_len = min(5000, len(long_text) - i * 5000)
+            response = json.dumps({
+                "video_type": "知识类",
+                "segments": [
+                    {
+                        "title": "测试标题",
+                        "content": "分块内容" * (chunk_len // 4),
+                        "start_index": 0,
+                        "end_index": chunk_len
+                    }
+                ],
+                "actual_segment_count": 1
+            }, ensure_ascii=False)
+            responses.append(response)
+
+        mock_client.generate.side_effect = responses
+
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+        from kb_workflow.srt.parser import Segment
+
+        strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
+
+        # 创建长文本字幕
+        segments = [Segment(1, 0.0, 1.0, long_text)]
+
+        with caplog.at_level(logging.INFO):
+            result = strategy.segment(segments, {"title": "测试"})
+
+        # 验证调用了 LLM 多次（每个分块一次）
+        assert mock_client.generate.call_count == expected_chunks
+
+        # 验证日志包含分块信息
+        assert any("splitting into chunks" in record.message.lower()
+                   for record in caplog.records)
+
+        # 验证返回了分段结果
+        assert len(result) > 0
+
+    def test_short_text_no_chunking(self, caplog):
+        """测试短文本（<=5000字符）不分块直接处理"""
+        import logging
+        import json
+
+        mock_client = Mock()
+        mock_client.api_key = "test-key"
+
+        # 短文本（< 5000 字符）
+        short_text = "这是短文本。" * 100
+
+        response = json.dumps({
+            "video_type": "知识类",
+            "segments": [
+                {
+                    "title": "测试标题",
+                    "content": short_text,
+                    "start_index": 0,
+                    "end_index": len(short_text)
+                }
+            ],
+            "actual_segment_count": 1
+        }, ensure_ascii=False)
+
+        mock_client.generate.return_value = response
+
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+        from kb_workflow.srt.parser import Segment
+
+        strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
+
+        segments = [Segment(1, 0.0, 1.0, short_text)]
+
+        with caplog.at_level(logging.INFO):
+            result = strategy.segment(segments, {"title": "测试"})
+
+        # 验证只调用了一次 LLM（不分块）
+        assert mock_client.generate.call_count == 1
+
+        # 验证日志不包含分块信息
+        assert not any("splitting into chunks" in record.message.lower()
+                       for record in caplog.records)
+
+        # 验证返回了分段结果
+        assert len(result) == 1
+
+
+class TestChunkIndexCorrection:
+    """测试分块索引修正（Plan 14-04）"""
+
+    def test_chunk_index_correction(self):
+        """测试分块索引相对于原文位置"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        # 模拟分块结果
+        chunk_text = "这是第一块。"
+        offset = 100
+
+        # 创建模拟的分段结果（索引相对于分块）
+        from kb_workflow.srt.semantic_engine import VideoSegment
+
+        chunk_segments = [
+            VideoSegment(
+                title="测试",
+                content=chunk_text,
+                start_index=0,
+                end_index=len(chunk_text)
+            )
+        ]
+
+        # 合并分块结果
+        chunk_results = [(chunk_segments, offset)]
+        merged = strategy._merge_segment_results(chunk_results)
+
+        # 验证索引被修正为相对于原文
+        assert len(merged) == 1
+        assert merged[0].start_index == offset
+        assert merged[0].end_index == offset + len(chunk_text)
+
+    def test_multiple_chunks_index_correction(self):
+        """测试多个分块的索引修正"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy, VideoSegment
+
+        strategy = UnifiedSegmentationStrategy()
+
+        # 创建多个分块
+        chunk1_text = "第一块内容。" * 100
+        chunk2_text = "第二块内容。" * 100
+        chunk3_text = "第三块内容。" * 100
+
+        # 模拟分块结果
+        chunk1_segments = [VideoSegment(title="块1", content=chunk1_text, start_index=0, end_index=len(chunk1_text))]
+        chunk2_segments = [VideoSegment(title="块2", content=chunk2_text, start_index=0, end_index=len(chunk2_text))]
+        chunk3_segments = [VideoSegment(title="块3", content=chunk3_text, start_index=0, end_index=len(chunk3_text))]
+
+        chunk_results = [
+            (chunk1_segments, 0),  # 第一个分块偏移量为 0
+            (chunk2_segments, len(chunk1_text)),  # 第二个分块偏移量为第一块的长度
+            (chunk3_segments, len(chunk1_text) + len(chunk2_text))  # 第三个分块偏移量为前两块的长度之和
+        ]
+
+        # 合并分块结果
+        merged = strategy._merge_segment_results(chunk_results)
+
+        # 验证所有分块的索引都被正确修正
+        assert len(merged) == 3
+
+        assert merged[0].start_index == 0
+        assert merged[0].end_index == len(chunk1_text)
+
+        assert merged[1].start_index == len(chunk1_text)
+        assert merged[1].end_index == len(chunk1_text) + len(chunk2_text)
+
+        assert merged[2].start_index == len(chunk1_text) + len(chunk2_text)
+        assert merged[2].end_index == len(chunk1_text) + len(chunk2_text) + len(chunk3_text)
+
+
+class TestChunkFallback:
+    """测试分块失败时的降级处理（Plan 14-04）"""
+
+    def test_chunk_fallback_to_rule_based(self, caplog):
+        """测试单个分块失败时降级到规则分段"""
+        import logging
+        import json
+
+        mock_client = Mock()
+        mock_client.api_key = "test-key"
+
+        # 创建长文本（会分成多个分块）
+        long_text = "这是第一句。" * 1000 + "这是第二句。" * 1000
+
+        # 第一个分块成功，第二个分块失败
+        success_response = json.dumps({
+            "video_type": "知识类",
+            "segments": [
+                {
+                    "title": "成功分块",
+                    "content": "内容",
+                    "start_index": 0,
+                    "end_index": 10
+                }
+            ],
+            "actual_segment_count": 1
+        }, ensure_ascii=False)
+
+        # 第一个分块返回成功，第二个分块抛出异常
+        mock_client.generate.side_effect = [success_response, Exception("LLM error")]
+
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+        from kb_workflow.srt.parser import Segment
+
+        strategy = UnifiedSegmentationStrategy(ai_client=mock_client)
+
+        segments = [Segment(1, 0.0, 1.0, long_text)]
+
+        with caplog.at_level(logging.WARNING):
+            result = strategy.segment(segments, {"title": "测试"})
+
+        # 验证返回了分段结果（第一个分块的 LLM 结果 + 第二个分块的规则分段）
+        assert len(result) > 0
+
+        # 验证日志包含分块失败和降级信息
+        assert any("failed" in record.message.lower() and "chunk" in record.message.lower()
+                   for record in caplog.records)
+
+    def test_fallback_chunk_returns_valid_segments(self):
+        """测试 _fallback_chunk 返回有效的 VideoSegment"""
+        from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy
+
+        strategy = UnifiedSegmentationStrategy()
+
+        chunk_text = "这是降级分块的内容。" * 100
+        offset = 500
+
+        # 调用降级方法
+        segments = strategy._fallback_chunk(chunk_text, offset)
+
+        # 验证返回了分段列表
+        assert len(segments) > 0
+
+        # 验证所有分段都是 VideoSegment 类型
+        from kb_workflow.srt.semantic_engine import VideoSegment
+        assert all(isinstance(seg, VideoSegment) for seg in segments)
+
+        # 验证第一个分段的索引正确
+        assert segments[0].start_index == offset
+
+        # 验证降级分段没有标题
+        assert all(seg.title is None for seg in segments)
+