feat(15-02): integrate KBSettings config for semantic segmentation

- Modified SRTProcessor to read enable_semantic_segmentation from KBSettings - Changed constructor parameter default from False to None (None = read from config) - Added config priority: constructor parameter > KBSettings > default (True) - Added logging for config status (INFO level) - Added comprehensive tests for config integration (TestConfigIntegration) - Added tests for output format compatibility (TestOutputFormatBenefits) - Fixed MockAIClient to accept disable_reasoning parameter - All 34 tests passing Requirements satisfied: - CONFIG-01: SRTProcessor reads from KBSettings - CONFIG-02: Disabled config uses rule-based segmentation - OUTPUT-01: LLM segmentation outputs H2 titles - OUTPUT-02: title=null skips title output - OUTPUT-03: Rule-based segmentation outputs "## 内容" Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-23 23:03:54 +08:00
parent 5debdd6f36
commit 5983e95352
2 changed files with 372 additions and 5 deletions
@@ -9,12 +9,13 @@ Tests for:

 import asyncio
 import json
+import os

 import pytest
 from pathlib import Path
 from dataclasses import dataclass
 from typing import List, Optional
-from unittest.mock import Mock
+from unittest.mock import Mock, patch


 # =============================================================================
@@ -323,7 +324,7 @@ class TestASRCorrector:
        call_count = 0

        class MockAIClient:
-            def generate(self, prompt: str, max_tokens: int = 16000) -> str:
+            def generate(self, prompt: str, max_tokens: int = 16000, disable_reasoning: bool = False) -> str:
                nonlocal call_count
                call_count += 1
                # 返回提示词中的文本部分
@@ -401,7 +402,11 @@ class TestSRTProcessor:
        from kb_workflow.pipeline.processors.srt import SRTProcessor
        from kb_workflow.pipeline.schemas import SRTInput

-        processor = SRTProcessor(enable_correction=False)
+        # 明确禁用语义分段以使用规则分段器（测试规则分段输出）
+        processor = SRTProcessor(
+            enable_correction=False,
+            enable_semantic_segmentation=False
+        )

        input_data = SRTInput(
            content_id="BV1test123",
@@ -580,3 +585,351 @@ class TestSemanticSegmentationToggle:
        srt_file = tmp_path / "test.srt"
        srt_file.write_text(srt_content, encoding='utf-8')
        return srt_file
+
+
+# =============================================================================
+# Test 6: Configuration Integration (CONFIG-01, CONFIG-02) - Plan 15-02
+# =============================================================================
+
+class TestConfigIntegration:
+    """测试配置集成 (CONFIG-01, CONFIG-02, OUTPUT-01, OUTPUT-02, OUTPUT-03)"""
+
+    @pytest.fixture
+    def mock_env_base(self):
+        """基础环境变量"""
+        return {
+            "KB_BASE_PATH": "/tmp/kb",
+            "KB_ARCHIVE_PATH": "/tmp/archive"
+        }
+
+    def test_srt_processor_reads_from_kbsettings(self, mock_env_base):
+        """Test 1: enable_semantic_segmentation=None 时从 KBSettings 读取配置"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.pipeline.config import get_settings
+
+        # 清除缓存以确保使用新的环境变量
+        get_settings.cache_clear()
+
+        with patch.dict(os.environ, {
+            **mock_env_base,
+            "KB_ENABLE_SEMANTIC_SEGMENTATION": "true"
+        }, clear=False):
+            get_settings.cache_clear()
+            processor = SRTProcessor()
+
+            # 验证从配置读取
+            assert processor.enable_semantic_segmentation is True
+
+    def test_srt_processor_constructor_overrides_config(self, mock_env_base):
+        """Test 2: 显式传入参数时覆盖配置值（参数优先级高于配置）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.pipeline.config import get_settings
+
+        get_settings.cache_clear()
+
+        with patch.dict(os.environ, {
+            **mock_env_base,
+            "KB_ENABLE_SEMANTIC_SEGMENTATION": "true"
+        }, clear=False):
+            get_settings.cache_clear()
+            # 显式传入 False 应该覆盖配置的 True
+            processor = SRTProcessor(enable_semantic_segmentation=False)
+
+            assert processor.enable_semantic_segmentation is False
+
+    def test_config_disabled_uses_rule_based(self, mock_env_base):
+        """Test 3: 配置为 False 且无参数时使用规则分段"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.srt.segmenter import SemanticSegmenter
+        from kb_workflow.pipeline.config import get_settings
+
+        get_settings.cache_clear()
+
+        with patch.dict(os.environ, {
+            **mock_env_base,
+            "KB_ENABLE_SEMANTIC_SEGMENTATION": "false"
+        }, clear=False):
+            get_settings.cache_clear()
+            processor = SRTProcessor()
+
+            # 验证使用规则分段器
+            assert isinstance(processor.segmenter, SemanticSegmenter)
+            assert processor.enable_semantic_segmentation is False
+
+    def test_config_enabled_uses_llm_segmentation(self, mock_env_base):
+        """Test 4: 配置为 True 且无参数时使用 LLM 分段"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.srt.semantic_engine import SemanticSegmentationEngine
+        from kb_workflow.pipeline.config import get_settings
+
+        get_settings.cache_clear()
+
+        with patch.dict(os.environ, {
+            **mock_env_base,
+            "KB_ENABLE_SEMANTIC_SEGMENTATION": "true"
+        }, clear=False):
+            get_settings.cache_clear()
+
+            with patch('kb_workflow.deep_analyzer.SimpleAIClient') as mock_client:
+                mock_client.return_value = Mock(api_key="test_key")
+
+                processor = SRTProcessor()
+
+                # 验证使用 LLM 分段引擎
+                assert isinstance(processor.segmenter, SemanticSegmentationEngine)
+                assert processor.enable_semantic_segmentation is True
+
+    def test_initialization_logs_config_status(self, mock_env_base, caplog):
+        """Test 5: 初始化日志记录配置状态（INFO 级别）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.pipeline.config import get_settings
+        import logging
+
+        get_settings.cache_clear()
+
+        with patch.dict(os.environ, {
+            **mock_env_base,
+            "KB_ENABLE_SEMANTIC_SEGMENTATION": "false"
+        }, clear=False):
+            get_settings.cache_clear()
+
+            with caplog.at_level(logging.INFO):
+                processor = SRTProcessor()
+
+                # 验证日志记录了配置状态
+                assert any("semantic_segmentation=" in record.message for record in caplog.records)
+                assert any("disabled" in record.message for record in caplog.records)
+
+
+# =============================================================================
+# Test 7: Output Format Compatibility (OUTPUT-01, OUTPUT-02, OUTPUT-03)
+# =============================================================================
+
+class TestOutputFormatCompatibility:
+    """测试输出格式兼容性"""
+
+    @pytest.fixture
+    def srt_files_for_output(self, tmp_path):
+        """创建输出格式测试用的文件"""
+        # 创建 SRT 文件
+        srt_content = """1
+00:00:00,000 --> 00:00:05,000
+这是第一段内容
+
+2
+00:00:05,000 --> 00:00:10,000
+这是第二段内容
+"""
+        srt_file = tmp_path / "test.srt"
+        srt_file.write_text(srt_content, encoding="utf-8")
+
+        # 创建元数据文件
+        metadata = {
+            "title": "输出格式测试",
+            "author": "测试作者",
+            "platform": "bilibili"
+        }
+        metadata_file = tmp_path / "metadata.json"
+        metadata_file.write_text(json.dumps(metadata, ensure_ascii=False), encoding="utf-8")
+
+        output_file = tmp_path / "output.md"
+
+        return {
+            "srt_file": str(srt_file),
+            "metadata_file": str(metadata_file),
+            "output_file": str(output_file)
+        }
+
+    def test_output_format_with_semantic_disabled(self, srt_files_for_output):
+        """Test 3: 规则分段输出 "## 内容" 标题（OUTPUT-03）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.pipeline.schemas import SRTInput
+
+        processor = SRTProcessor(
+            enable_semantic_segmentation=False,
+            enable_correction=False
+        )
+
+        input_data = SRTInput(
+            content_id="test123",
+            platform="bilibili",
+            srt_file=srt_files_for_output["srt_file"],
+            metadata_file=srt_files_for_output["metadata_file"],
+            output_file=srt_files_for_output["output_file"]
+        )
+
+        result = asyncio.run(processor.process(input_data))
+
+        # 验证输出文件
+        content = Path(result.output_file).read_text(encoding="utf-8")
+
+        # 验证 "## 内容" 标题（OUTPUT-03）
+        assert "## 内容" in content
+        # 验证段落内容直接输出
+        assert "这是第一段内容" in content
+        assert "这是第二段内容" in content
+
+    def test_generate_markdown_with_video_segments(self):
+        """Test 1: LLM 分段输出带 H2 标题的 Markdown（OUTPUT-01）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.srt.semantic_engine import VideoSegment
+
+        processor = SRTProcessor(enable_correction=False)
+
+        # 创建 VideoSegment 对象模拟 LLM 分段结果
+        mock_segments = [
+            VideoSegment(title="第一段", content="这是第一段内容", start_index=0, end_index=10),
+            VideoSegment(title=None, content="这是第二段内容", start_index=10, end_index=20),
+            VideoSegment(title="第三段", content="这是第三段内容", start_index=20, end_index=30)
+        ]
+
+        # 创建临时输出文件
+        import tempfile
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f:
+            output_file = Path(f.name)
+
+        try:
+            # 调用 _generate_markdown 方法
+            processor._generate_markdown(
+                output_file=output_file,
+                metadata={"title": "测试"},
+                paragraphs=mock_segments,
+                tags=[]
+            )
+
+            # 验证输出文件
+            content = output_file.read_text(encoding="utf-8")
+
+            # 验证 H2 标题（OUTPUT-01）
+            assert "## 第一段" in content
+            assert "## 第三段" in content
+            # 验证 title=None 时不输出标题（OUTPUT-02）
+            assert "## 第二段" not in content
+            assert "这是第二段内容" in content
+            # 验证内容正确输出
+            assert "这是第一段内容" in content
+            assert "这是第三段内容" in content
+        finally:
+            # 清理临时文件
+            if output_file.exists():
+                output_file.unlink()
+
+    def test_generate_markdown_with_plain_text(self):
+        """Test 3b: 规则分段输出纯文本（带 "## 内容" 标题）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+
+        processor = SRTProcessor(enable_correction=False)
+
+        # 创建纯文本列表模拟规则分段结果
+        mock_paragraphs = [
+            "这是第一段内容",
+            "这是第二段内容"
+        ]
+
+        # 创建临时输出文件
+        import tempfile
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f:
+            output_file = Path(f.name)
+
+        try:
+            # 调用 _generate_markdown 方法
+            processor._generate_markdown(
+                output_file=output_file,
+                metadata={"title": "测试"},
+                paragraphs=mock_paragraphs,
+                tags=[]
+            )
+
+            # 验证输出文件
+            content = output_file.read_text(encoding="utf-8")
+
+            # 验证 "## 内容" 标题（OUTPUT-03）
+            assert "## 内容" in content
+            # 验证段落内容直接输出
+            assert "这是第一段内容" in content
+            assert "这是第二段内容" in content
+        finally:
+            # 清理临时文件
+            if output_file.exists():
+                output_file.unlink()
+
+    def test_generate_markdown_with_null_titles(self):
+        """Test 2: 所有 title=None 时不输出分段标题（OUTPUT-02）"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.srt.semantic_engine import VideoSegment
+
+        processor = SRTProcessor(enable_correction=False)
+
+        # 创建全部 title=None 的 VideoSegment
+        mock_segments = [
+            VideoSegment(title=None, content="第一段内容", start_index=0, end_index=10),
+            VideoSegment(title=None, content="第二段内容", start_index=10, end_index=20)
+        ]
+
+        # 创建临时输出文件
+        import tempfile
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f:
+            output_file = Path(f.name)
+
+        try:
+            # 调用 _generate_markdown 方法
+            processor._generate_markdown(
+                output_file=output_file,
+                metadata={"title": "测试"},
+                paragraphs=mock_segments,
+                tags=[]
+            )
+
+            # 验证输出文件
+            content = output_file.read_text(encoding="utf-8")
+
+            # 验证没有分段标题（只有 frontmatter 和内容）
+            lines = content.split('\n')
+            # 找到 frontmatter 结束后的内容
+            in_content = False
+            for line in lines:
+                if line == "---":
+                    if in_content:
+                        break
+                    in_content = True
+                    continue
+                if in_content and line.startswith("## "):
+                    # 只能有 "## 内容" 标题，不能有其他分段标题
+                    assert line == "## 内容"
+
+            # 验证内容仍然存在
+            assert "第一段内容" in content
+            assert "第二段内容" in content
+        finally:
+            # 清理临时文件
+            if output_file.exists():
+                output_file.unlink()
+
+    def test_both_modes_valid_output(self, srt_files_for_output):
+        """Test 4: 两种模式输出结构不同但都有效"""
+        from kb_workflow.pipeline.processors.srt import SRTProcessor
+        from kb_workflow.pipeline.schemas import SRTInput
+
+        # 测试规则分段模式
+        processor = SRTProcessor(
+            enable_semantic_segmentation=False,
+            enable_correction=False
+        )
+
+        input_data = SRTInput(
+            content_id="test123",
+            platform="bilibili",
+            srt_file=srt_files_for_output["srt_file"],
+            metadata_file=srt_files_for_output["metadata_file"],
+            output_file=srt_files_for_output["output_file"]
+        )
+
+        result = asyncio.run(processor.process(input_data))
+        assert result.success
+        assert Path(result.output_file).exists()
+
+        # 验证输出文件存在且非空
+        content = Path(result.output_file).read_text(encoding="utf-8")
+        assert len(content) > 0
+        assert "---" in content  # frontmatter
+        assert "## 内容" in content