From 0fd243c8e176672644cbe98fcd20048405bc0467 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:03:41 +0800 Subject: [PATCH] fix(04-05): revise plan based on checker feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issues addressed: 1. Remove non-existent 04-06 from depends_on → [04-04] only 2. Add CLI 'process' subcommand implementation task (was missing) 3. Fix verify command syntax (pytest instead of malformed python -c) 4. Rewrite Task 2 action with proper test code structure Co-Authored-By: Claude Opus 4.6 --- .../04-analysis-orchestration/04-05-PLAN.md | 370 +++++++++++------- 1 file changed, 228 insertions(+), 142 deletions(-) diff --git a/.planning/phases/04-analysis-orchestration/04-05-PLAN.md b/.planning/phases/04-analysis-orchestration/04-05-PLAN.md index e12da3a..a23bc92 100644 --- a/.planning/phases/04-analysis-orchestration/04-05-PLAN.md +++ b/.planning/phases/04-analysis-orchestration/04-05-PLAN.md @@ -5,7 +5,8 @@ type: execute wave: 5 depends_on: [04-04] files_modified: - - README.md + - kb-workflow/kb_workflow.py + - tests/test_cli/test_process_command.py - tests/test_integration/test_e2e_workflow.py autonomous: false requirements: [MNT-01] @@ -13,32 +14,37 @@ user_setup: [] must_haves: truths: + - "CLI 'process' subcommand is available" + - "kb-workflow process executes full pipeline" - "End-to-end workflow completes successfully" - "Analysis report is saved to content directory" - "workflow_state.json correctly records all steps" artifacts: - - path: "README.md" - provides: "Project documentation" - contains: "## Installation" + - path: "kb-workflow/kb_workflow.py" + provides: "CLI entry point with process subcommand" + contains: "def process_command" + - path: "tests/test_cli/test_process_command.py" + provides: "CLI command tests" + contains: "test_process_command" - path: "tests/test_integration/test_e2e_workflow.py" provides: "End-to-end integration test" contains: "test_e2e" key_links: - - from: "kb_pipeline/pipeline.py" - to: "kb_pipeline/processors" - via: "import processors" - pattern: "from.*processor import" - from: "kb-workflow/kb_workflow.py" to: "kb_pipeline" - via: "import pipeline" - pattern: "from.*pipeline import" + via: "import Pipeline" + pattern: "from kb_pipeline" + - from: "kb_pipeline/pipeline.py" + to: "kb_pipeline/processors" + via: "Pipeline.run()" + pattern: "Pipeline\(" --- -Create end-to-end integration test and documentation ( project overview, installation, and and usage. +Add CLI 'process' subcommand and create end-to-end integration tests. -Purpose: Verify the complete workflow works and provide clear project documentation for maintainability. -Output: README.md with project overview, tests/test_integration/test_e2e_workflow.py +Purpose: Provide the single-command entry point `kb-workflow process ` as specified in CONTEXT.md, and verify the complete workflow works. +Output: CLI process subcommand in kb_workflow.py, CLI tests, E2E integration tests. @@ -52,108 +58,121 @@ Output: README.md with project overview, tests/test_integration/test_e2e_workflo @.planning/STATE.md @.planning/phases/04-analysis-orchestration/04-CONTEXT.md @.planning/phases/04-analysis-orchestration/04-RESEARCH.md + @.planning/phases/04-analysis-orchestration/04-04-SUMMARY.md - + -From kb_pipeline/pipeline.py: +from kb_pipeline/pipeline.py: ```python class Pipeline: """Pipeline orchestration engine""" - def __init__(self, ...): # From 04-01 - ... + def __init__( + self, + processors: List[BaseProcessor], + config: Optional[PipelineConfig] = None, + output_dir: Optional[Path] = None, + ): ... + async def run(self, url: str) -> PipelineOutput: ... ``` -From kb_pipeline/processors/base.py: +from kb_pipeline/schemas.py: +```python +class PipelineInput(BaseModel): + url: str = Field(..., min_length=1) + output_dir: Optional[Path] = None + skip_completed: bool = Field(default=True) + +class PipelineOutput(BaseModel): + success: bool + error_message: Optional[str] = None + results: Dict[str, Any] = Field(default_factory=dict) + content_dir: Optional[Path] = None + +class PipelineConfig(BaseModel): + steps: List[str] = Field(default_factory=lambda: ["parse", "download", "asr", "analyze"]) + output_dir: Optional[Path] = None + skip_completed: bool = Field(default=True) + max_retries: int = Field(default=3) +``` + +from kb_pipeline/processors/base.py: ```python class BaseProcessor(ABC): STEP_NAME: str async def process(self, input_data: StepInput) -> StepOutput ``` -from kb_pipeline/processors/parse.py: +from kb-workflow/kb_workflow.py (existing CLI pattern): ```python -class ParseProcessor(BaseProcessor): - STEP_NAME = "parse" - async def process(self, input_data: ParseInput) -> ParseOutput +def main(): + parser = argparse.ArgumentParser(...) + parser.add_argument('--check', action='store_true', ...) + # ... existing arguments ... ``` -from kb_pipeline/processors/download.py: -```python -class DownloadProcessor(BaseProcessor): - STEP_NAME = "download" - async def process(self, input_data: DownloadInput) -> DownloadOutput -``` - -from kb_pipeline/processors/asr.py: -```python -class ASRProcessor: - STEP_NAME = "asr" - async def process(self, input_data: ASRInput) -> ASROutput -``` - -from kb_pipeline/processors/analyze.py -```python -class AnalyzeProcessor(BaseProcessor): - STEP_NAME = "analyze" - async def process(self, input_data: AnalyzeInput) -> AnalyOutput -``` - Task 1: Create README.md - README.md + Task 1: Add CLI 'process' subcommand + kb-workflow/kb_workflow.py, tests/test_cli/test_process_command.py - - Test 1: README.md exists and contains project overview - - Test 2: README.md contains installation instructions + - Test 1: `kb-workflow process --help` shows usage + - Test 2: `kb-workflow process ` calls Pipeline.run() + - Test 3: Invalid URL returns appropriate error + - Test 4: --output-dir flag sets custom output directory -Create `README.md` in project root directory with the following sections: +Add `process` subcommand to kb-workflow/kb_workflow.py using argparse subparsers. -**## kb-skills** +**Implementation requirements:** -知识库工作流工具，从链接到分析报告的端到端自动化处理系统。 +1. Add subparser for `process` command: +```python +subparsers = parser.add_subparsers(dest='command', help='Available commands') -### 格 Features - -- 统一 Processor 模式： Parse -> Download -> ASR -> Analyze -- 状态管理: workflow_state.json -- 配置管理: 通过 ~/.kb_env -- 输出: Markdown 格式分析报告 - -- 幂等性保证: 重复执行不会产生重复文件 - -- 断点续传: 中断后可继续 - -- 模块化设计: 新人能在30分钟内理解代码结构 -- 清晰的项目结构 - -- 统一的 CLI 入 - - entry - -### Installation - -```bash -# 安装依赖 -pip install kb-pipeline +# Process subcommand +process_parser = subparsers.add_parser('process', help='Process URL through full pipeline') +process_parser.add_argument('url', type=str, help='URL to process') +process_parser.add_argument('--output-dir', '-o', type=str, help='Output directory for content') +process_parser.add_argument('--skip-completed', action='store_true', default=True, help='Skip completed steps') +process_parser.add_argument('--no-skip', action='store_false', dest='skip_completed', help='Re-run all steps') ``` -验证安装: -```bash -# 测试 -pytest tests/test_pipeline/test_schemas.py tests/test_pipeline/test_state.py tests/test_pipeline/test_config.py tests/test_pipeline/test_errors.py tests/test_pipeline/test_retry.py +2. Add `process_command` function: +```python +def process_command(url: str, output_dir: Optional[str] = None, skip_completed: bool = True) -> Dict[str, Any]: + """Execute full pipeline for a URL""" + import asyncio + from kb_pipeline.pipeline import Pipeline + from kb_pipeline.processors import ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor + from kb_pipeline.schemas import PipelineConfig + from kb_pipeline.state import WorkflowStateManager + + # Parse URL first to get platform and content_id + # ... implementation ... + + # Create pipeline and run + async def run_pipeline(): + # Initialize processors and pipeline + # ... implementation ... + + result = asyncio.run(run_pipeline()) + return result ``` +3. Update `main()` to dispatch to `process_command` when `args.command == 'process'` + +4. Create tests/test_cli/test_process_command.py with mocked Pipeline tests. - python -c "import os; content = open('README.md', 'r', encoding='utf-8'); print(content[:100]) > 0" + pytest tests/test_cli/test_process_command.py -v -x - README.md created with project overview, installation, and, usage sections, docstrings + CLI process subcommand implemented and tested @@ -163,13 +182,14 @@ pytest tests/test_pipeline/test_schemas.py tests/test_pipeline/test_state.py tes - Test 1: Full workflow executes from URL to analysis - Test 2: Workflow handles errors gracefully - Test 3: State is persisted correctly - - Test 4: Workflow is resumable + - Test 4: Workflow is resumable (skip completed steps) Create `tests/test_integration/test_e2e_workflow.py` with comprehensive end-to-end tests. -**Test structure:** +**Test file structure:** ```python +"""End-to-end integration tests for the complete workflow.""" import pytest import pytest_asyncio from unittest.mock import AsyncMock, patch, MagicMock @@ -177,64 +197,139 @@ from pathlib import Path from typing import Dict, Any from kb_pipeline.pipeline import Pipeline -from kb_pipeline.config import get_settings -from kb_pipeline.processors import ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor +from kb_pipeline.schemas import PipelineConfig, WorkflowState +from kb_pipeline.state import WorkflowStateManager @pytest.fixture def temp_content_dir(tmp_path): - """Create temp content directory""" + """Create temp content directory for testing.""" + content_dir = tmp_path / "content" content_dir.mkdir(parents=True, exist_ok=True) yield content_dir - # Create test workflow state - state_path = content_dir / "workflow_state.json" - state = WorkflowState( + +@pytest.fixture +def mock_state_manager(temp_content_dir): + """Create mock workflow state manager.""" + return WorkflowStateManager(temp_content_dir) + + +@pytest.fixture +def mock_processors(): + """Create mock processors for testing.""" + processors = [] + + for step_name in ["parse", "download", "asr", "analyze"]: + processor = MagicMock() + processor.STEP_NAME = step_name + processor.process = AsyncMock(return_value=MagicMock( + success=True, + error_message=None, + )) + processors.append(processor) + + return processors + + +@pytest.mark.asyncio +async def test_full_workflow_executes_successfully(temp_content_dir, mock_processors): + """Test that full workflow executes from URL to analysis.""" + config = PipelineConfig( + steps=["parse", "download", "asr", "analyze"], + output_dir=temp_content_dir, + ) + + pipeline = Pipeline( + processors=mock_processors, + config=config, + output_dir=temp_content_dir, + ) + + result = await pipeline.run("https://www.bilibili.com/video/BV1test123") + + assert result.success + for processor in mock_processors: + processor.process.assert_called_once() + + +@pytest.mark.asyncio +async def test_workflow_handles_errors_gracefully(temp_content_dir): + """Test that workflow handles errors gracefully.""" + # Create processor that fails + failing_processor = MagicMock() + failing_processor.STEP_NAME = "parse" + failing_processor.process = AsyncMock(return_value=MagicMock( + success=False, + error_message="Network error", + )) + + config = PipelineConfig( + steps=["parse"], + output_dir=temp_content_dir, + ) + + pipeline = Pipeline( + processors=[failing_processor], + config=config, + output_dir=temp_content_dir, + ) + + result = await pipeline.run("https://invalid.url") + + assert not result.success + assert result.error_message is not None + + +@pytest.mark.asyncio +async def test_state_persisted_correctly(temp_content_dir, mock_state_manager): + """Test that state is persisted correctly after each step.""" + # ... test implementation + pass + + +@pytest.mark.asyncio +async def test_workflow_resumable(temp_content_dir): + """Test that workflow can resume (skip completed steps).""" + # Create state with parse already completed + state_manager = WorkflowStateManager(temp_content_dir) + state = state_manager.get_or_create( content_id="test123", platform="bilibili", source_url="https://test.url" ) - state_manager = WorkflowStateManager(content_dir) - yield state_manager + state.mark_step_completed("parse") + state_manager.save(state) - # Create real pipeline - settings = get_settings() - pipeline = Pipeline( - processors=[ - ParseProcessor(state_manager), - DownloadProcessor(state_manager), - ASRProcessor(state_manager), - AnalyzeProcessor(state_manager), - ], - output_dir=content_dir, - config=PipelineConfig( - steps=["parse", "download", "asr", "analyze"], - ) + # Create processors + parse_processor = MagicMock() + parse_processor.STEP_NAME = "parse" + parse_processor.process = AsyncMock() + + download_processor = MagicMock() + download_processor.STEP_NAME = "download" + download_processor.process = AsyncMock(return_value=MagicMock(success=True)) + + config = PipelineConfig( + steps=["parse", "download"], + output_dir=temp_content_dir, + skip_completed=True, ) - yield pipeline - # Mock workflow run - result = await pipeline.run("https://test.url") - # Verify success - assert result["parse"]["success"] - assert result["download"]["success"] - assert result["asr"]["success"] or result["asr"]["skipped"] # ASR not applicable for text content - assert result["analyze"]["success"] + pipeline = Pipeline( + processors=[parse_processor, download_processor], + config=config, + output_dir=temp_content_dir, + ) - # Verify state was saved - state = state_manager.load() - assert state is not None - assert state.is_step_completed("parse") - assert state.is_step_completed("download") - if content_type in ["video", "zvideo"]: - assert state.is_step_completed("asr") - assert not state.is_step_completed("asr") - assert state.is_step_completed("analyze") + # Run pipeline - parse should be skipped + result = await pipeline.run("https://www.bilibili.com/video/BV1test123") - # Verify output files exist - assert (content_dir / "bilibili_test123_metadata.json").exists() - assert (content_dir / "bilibili_test123_content.json").exists() or result["content_type"] == "video": - assert (content_dir / "bilibili_test123_analysis.md").exists() + # Parse should NOT be called (skipped) + parse_processor.process.assert_not_called() + # Download should be called + download_processor.process.assert_called_once() +``` pytest tests/test_integration/test_e2e_workflow.py -v @@ -244,40 +339,31 @@ def temp_content_dir(tmp_path): -End-to-end workflow (Phase 1-2/3): +End-to-end workflow (Phase 4): - Pipeline Engine with 4 processors (Parse, Download, ASR, Analyze) -- CLI entry point (kb-workflow process command) +- CLI entry point with `kb-workflow process ` command - DeepAnalyzer integration for analysis reports saved as Markdown - State persistence via workflow_state.json -DeepAnalyzer integration: - - B站视频/专栏链接 → parse, download, ASR, analyze -- Error recovery with retry -- Progress logging with simple format -- Documentation in README.md and docstrings + +1. Run `pytest tests/test_cli/test_process_command.py tests/test_integration/test_e2e_workflow.py -v -x` +2. Run `python -m kb_workflow.kb_workflow process --help` to see usage +3. (Optional) Run `kb-workflow process ` with a real URL to test end-to-end -1. Run `pytest tests/test_integration/test_e2e_workflow.py -v -x` -2. Visit https://github.com/your-org/kb-skills -3. Run `kb-workflow process ` to test -4. Run `kb-workflow process --help` to see full usage - - Type "approved" or describe any issues found -release-signal>Type: "approved" to indicate you should proceed to Phase 4 for finalize the documentation - + Type "approved" to indicate successful completion, or describe issues found -- All tests pass +- All tests pass: `pytest tests/test_cli/ tests/test_integration/ -v` +- CLI process subcommand available: `python -m kb_workflow.kb_workflow process --help` - End-to-end workflow tested with state persistence -- README.md exists with project overview -- tests pass -- README.md complete with project overview, installation, and - usage +- CLI `process` subcommand implemented in kb_workflow.py +- CLI tests pass - End-to-end integration test passes - workflow_state.json correctly records all steps - Documentation is clear and comprehensive