From 5983e95352d71f17eb03c2156a485ef7b02ae91e Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 23 Mar 2026 23:03:54 +0800 Subject: [PATCH] feat(15-02): integrate KBSettings config for semantic segmentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- kb_workflow/pipeline/processors/srt.py | 18 +- tests/test_srt_processor.py | 359 ++++++++++++++++++++++++- 2 files changed, 372 insertions(+), 5 deletions(-) diff --git a/kb_workflow/pipeline/processors/srt.py b/kb_workflow/pipeline/processors/srt.py index 7632e4c..dcccf28 100644 --- a/kb_workflow/pipeline/processors/srt.py +++ b/kb_workflow/pipeline/processors/srt.py @@ -49,7 +49,7 @@ class SRTProcessor(BaseProcessor): self, state_manager: Optional[WorkflowStateManager] = None, enable_correction: bool = True, - enable_semantic_segmentation: bool = False, + enable_semantic_segmentation: Optional[bool] = None, enable_cache: bool = True ): """初始化 SRT 处理器 @@ -57,10 +57,24 @@ class SRTProcessor(BaseProcessor): Args: state_manager: 状态管理器 enable_correction: 是否启用 ASR 纠错（默认 True） - enable_semantic_segmentation: 是否启用 LLM 语义分段（默认 False） + enable_semantic_segmentation: 是否启用 LLM 语义分段（None 时从配置读取） enable_cache: 是否启用缓存（默认 True，仅在启用语义分段时有效） """ super().__init__(state_manager) + + # 如果未明确指定，从配置读取 + if enable_semantic_segmentation is None: + settings = get_settings() + enable_semantic_segmentation = settings.enable_semantic_segmentation + logger.info( + f"SRTProcessor using config: " + f"semantic_segmentation={'enabled' if enable_semantic_segmentation else 'disabled'}" + ) + elif enable_semantic_segmentation: + logger.info("SRTProcessor using explicit parameter: semantic_segmentation=enabled") + else: + logger.info("SRTProcessor using explicit parameter: semantic_segmentation=disabled") + self.enable_correction = enable_correction self.enable_semantic_segmentation = enable_semantic_segmentation self.segmenter = None diff --git a/tests/test_srt_processor.py b/tests/test_srt_processor.py index 7e5546c..9e4842f 100644 --- a/tests/test_srt_processor.py +++ b/tests/test_srt_processor.py @@ -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