starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(04-05): revise plan based on checker feedback

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-13 23:03:41 +08:00
parent bf44050392
commit 0fd243c8e1
1 changed files with 228 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.planning/phases/04-analysis-orchestration/04-05-PLAN.md
+228 -142
View File
@@ -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 <url> 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\("
---
<objective>
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 <url>` as specified in CONTEXT.md, and verify the complete workflow works.
Output: CLI process subcommand in kb_workflow.py, CLI tests, E2E integration tests.
</objective>
<execution_context>
@@ -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
<interfaces>
<!-- Key types and contracts from existing code. Executor should use these directly. -->
<!-- Key types and contracts from existing codebase. Executor should use these directly. -->
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
```
</interfaces>
</context>
<tasks>
<task type="auto" tdd="true">
<name>Task 1: Create README.md</name>
<files>README.md</files>
<name>Task 1: Add CLI 'process' subcommand</name>
<files>kb-workflow/kb_workflow.py, tests/test_cli/test_process_command.py</files>
<behavior>
- 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 <url>` calls Pipeline.run()
- Test 3: Invalid URL returns appropriate error
- Test 4: --output-dir flag sets custom output directory
</behavior>
<action>
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.
</action>
<verify>
<automated>python -c "import os; content = open('README.md', 'r', encoding='utf-8'); print(content[:100]) > 0"
<automated>pytest tests/test_cli/test_process_command.py -v -x</automated>
</verify>
<done>README.md created with project overview, installation, and, usage sections, docstrings</done>
<done>CLI process subcommand implemented and tested</done>
</task>
<task type="auto" tdd="true">
@@ -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)
</behavior>
<action>
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()
```
</action>
<verify>
<automated>pytest tests/test_integration/test_e2e_workflow.py -v</automated>
@@ -244,40 +339,31 @@ def temp_content_dir(tmp_path):
<task type="checkpoint:human-verify" gate="blocking">
<what-built>
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 <url>` command
- DeepAnalyzer integration for analysis reports saved as Markdown
- State persistence via workflow_state.json
</what-built>
DeepAnalyzer integration:
- B站视频/专栏链接 → parse, download, ASR, analyze
- Error recovery with retry
- Progress logging with simple format
- Documentation in README.md and docstrings
<how-to-verify>
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 <url>` with a real URL to test end-to-end
</how-to-verify>
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 <url>` to test
4. Run `kb-workflow process --help` to see full usage
</how-to-verify>
<resume-signal>Type "approved" or describe any issues found
release-signal>Type: "approved" to indicate you should proceed to Phase 4 for finalize the documentation
</how-to-verify>
<resume-signal>Type "approved" to indicate successful completion, or describe issues found</resume-signal>
</task>
</tasks>
<verification>
- 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
</verification>
<success_criteria>
- 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