From 747bb799ca241fc32492b8c1c937c5c954140649 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:05:27 +0800 Subject: [PATCH] docs(04): complete phase planning --- .../04-analysis-orchestration/04-01-PLAN.md | 26 +- .../04-analysis-orchestration/04-02-PLAN.md | 31 +- .../04-analysis-orchestration/04-04-PLAN.md | 651 ++++-------------- .../04-VALIDATION.md | 26 +- 4 files changed, 180 insertions(+), 554 deletions(-) diff --git a/.planning/phases/04-analysis-orchestration/04-01-PLAN.md b/.planning/phases/04-analysis-orchestration/04-01-PLAN.md index 9f4cb96..d8fb58d 100644 --- a/.planning/phases/04-analysis-orchestration/04-01-PLAN.md +++ b/.planning/phases/04-analysis-orchestration/04-01-PLAN.md @@ -36,11 +36,11 @@ must_haves: - from: "kb_pipeline/processors/parse.py" to: "kb_pipeline/link_parser.py" via: "LinkParser.parse()" - pattern: "LinkParser\\(\\)\\.parse" + pattern: "LinkParser\(\)\.parse" - from: "kb_pipeline/processors/base.py" to: "kb_pipeline/state.py" via: "WorkflowStateManager integration" - pattern: "state_manager\\." + pattern: "state_manager\." --- @@ -198,13 +198,14 @@ Add tests to existing test_schemas.py file. - Task 3: Implement ParseProcessor - kb_pipeline/processors/parse.py, tests/test_processors/test_parse.py + Task 3: Implement ParseProcessor with __init__.py export + kb_pipeline/processors/parse.py, kb_pipeline/processors/__init__.py, tests/test_processors/test_parse.py - Test 1: ParseProcessor parses Bilibili video URL correctly - Test 2: ParseProcessor parses Zhihu answer URL correctly - Test 3: ParseProcessor returns ParseOutput with success=False for unknown URLs - Test 4: ParseProcessor records state when state_manager is provided + - Test 5: ParseProcessor can be imported from kb_pipeline.processors Create `kb_pipeline/processors/parse.py`: @@ -222,18 +223,8 @@ Create `kb_pipeline/processors/parse.py`: 4. Keep implementation simple - just wrap LinkParser 5. Handle both success and failure cases with proper state recording - - - pytest tests/test_processors/test_parse.py -v -x - - ParseProcessor correctly parses URLs and returns structured ParseOutput - - - Task 4: Update processors __init__.py - kb_pipeline/processors/__init__.py - -Update `kb_pipeline/processors/__init__.py` to export new classes: +Then update `kb_pipeline/processors/__init__.py` to export new classes: ```python from .base import BaseProcessor @@ -246,9 +237,9 @@ __all__ = ["BaseProcessor", "ParseProcessor", "ASRProcessor"] Keep existing ASRProcessor export and add new ones. - python -c "from kb_pipeline.processors import BaseProcessor, ParseProcessor, ASRProcessor; print('OK')" + pytest tests/test_processors/test_parse.py -v -x && python -c "from kb_pipeline.processors import BaseProcessor, ParseProcessor, ASRProcessor; print('OK')" - All processors can be imported from kb_pipeline.processors + ParseProcessor correctly parses URLs, returns structured ParseOutput, and is exported from processors module @@ -258,6 +249,7 @@ Keep existing ASRProcessor export and add new ones. - BaseProcessor is abstract and cannot be instantiated - ParseProcessor correctly wraps LinkParser - State recording works when state_manager is provided +- All processors can be imported from kb_pipeline.processors diff --git a/.planning/phases/04-analysis-orchestration/04-02-PLAN.md b/.planning/phases/04-analysis-orchestration/04-02-PLAN.md index f7f1685..91bf02c 100644 --- a/.planning/phases/04-analysis-orchestration/04-02-PLAN.md +++ b/.planning/phases/04-analysis-orchestration/04-02-PLAN.md @@ -6,6 +6,7 @@ wave: 2 depends_on: [04-01] files_modified: - kb_pipeline/processors/download.py + - kb_pipeline/processors/__init__.py - kb_pipeline/schemas.py - tests/test_processors/test_download.py autonomous: true @@ -37,7 +38,7 @@ must_haves: - from: "kb_pipeline/processors/download.py" to: "kb_pipeline/state.py" via: "WorkflowStateManager" - pattern: "state_manager\\." + pattern: "state_manager\." --- @@ -138,7 +139,7 @@ Follow StepInput/StepOutput pattern from Phase 1. Create test file with schema validation tests. - pytest tests/test_processors/test_download.py -v -x + pytest tests/test_processors/test_download.py -v -x -k "test_schema" DownloadInput/DownloadOutput schemas defined and validated with tests @@ -175,21 +176,22 @@ Create `kb_pipeline/processors/download.py` with DownloadProcessor class. Create comprehensive tests mocking adapters and state_manager. - pytest tests/test_processors/test_download.py -v -x + pytest tests/test_processors/test_download.py -v -x -k "test_metadata" DownloadProcessor fetches metadata with state recording - Task 3: Add content downloading to DownloadProcessor - kb_pipeline/processors/download.py, tests/test_processors/test_download.py + Task 3: Add content downloading and __init__.py export + kb_pipeline/processors/download.py, kb_pipeline/processors/__init__.py, tests/test_processors/test_download.py - Test 1: DownloadProcessor downloads Bilibili article content - Test 2: DownloadProcessor skips video download (handled by ASR) - Test 3: DownloadProcessor creates correct output file paths + - Test 4: DownloadProcessor can be imported from kb_pipeline.processors -Extend DownloadProcessor to download content. +Extend DownloadProcessor to download content and update __init__.py. **For articles/text content:** - Call adapter.fetch_content() to get text @@ -205,18 +207,8 @@ Extend DownloadProcessor to download content. - Content: `{platform}_{content_id}_content.md` (for articles) Update tests to verify content downloading behavior. - - - pytest tests/test_processors/test_download.py -v -x - - DownloadProcessor downloads content for text-based content types - - - Task 4: Update processors __init__.py - kb_pipeline/processors/__init__.py - -Update `kb_pipeline/processors/__init__.py` to export DownloadProcessor. +Then update `kb_pipeline/processors/__init__.py` to export DownloadProcessor: ```python from .base import BaseProcessor @@ -228,9 +220,9 @@ __all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor ``` - python -c "from kb_pipeline.processors import DownloadProcessor; print('OK')" + pytest tests/test_processors/test_download.py -v -x && python -c "from kb_pipeline.processors import DownloadProcessor; print('OK')" - DownloadProcessor exported from processors module + DownloadProcessor downloads content for text-based content types and is exported from processors module @@ -240,6 +232,7 @@ __all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor - DownloadProcessor correctly uses PlatformAdapter implementations - State recording works for both success and failure cases - Content files are saved to correct locations +- DownloadProcessor can be imported from kb_pipeline.processors diff --git a/.planning/phases/04-analysis-orchestration/04-04-PLAN.md b/.planning/phases/04-analysis-orchestration/04-04-PLAN.md index b1f288b..53b6455 100644 --- a/.planning/phases/04-analysis-orchestration/04-04-PLAN.md +++ b/.planning/phases/04-analysis-orchestration/04-04-PLAN.md @@ -5,51 +5,41 @@ type: execute wave: 4 depends_on: [04-01, 04-02, 04-03] files_modified: + - kb_pipeline/schemas.py - kb_pipeline/pipeline.py - - kb_pipeline/processors/__init__.py - - kb-workflow/kb_workflow.py - tests/test_pipeline/test_pipeline.py - - tests/test_cli/test_process_command.py autonomous: true requirements: [MNT-01] user_setup: [] must_haves: truths: - - "Pipeline orchestrestr executes full workflow from URL to analysis" - - "Single command triggers entire workflow" - - "Progress is visible through simple logging" - - "New developers can understand code in 30 minutes" + - "Pipeline schemas (PipelineInput/Output/Config) are defined" + - "Pipeline class initializes with processors list" + - "Pipeline orchestrates processor execution" artifacts: + - path: "kb_pipeline/schemas.py" + provides: "PipelineInput/Output/Config schemas" + contains: "class PipelineInput" - path: "kb_pipeline/pipeline.py" provides: "Pipeline engine class" - min_lines: 100 - - path: "kb-workflow/kb_workflow.py" - provides: "CLI entry point with process subcommand" + contains: "class Pipeline" min_lines: 50 - path: "tests/test_pipeline/test_pipeline.py" - provides: "Pipeline tests" - min_lines: 100 - - path: "tests/test_cli/test_process_command.py" - provides: "CLI tests" + provides: "Pipeline unit tests" + contains: "test_pipeline" key_links: - from: "kb_pipeline/pipeline.py" to: "kb_pipeline/processors" - via: "Pipeline.run()" - pattern: "Pipeline\\(" - - from: "kb-workflow/kb_workflow.py" - to: "kb_pipeline.processors" - via: "import" - pattern: "from.* import" - - from: "kb_pipeline/processors" import ParseProcessor, DownloadProcessor, ASrProcessor, AnalyzeProcessor - pattern: "from.* import.*" + via: "process() calls" + pattern: "processor\.process\(" --- -Create Pipeline engine to orchest all processors and execute the complete workflow from URL to analysis report. +Create Pipeline schemas and Pipeline class foundation to orchestrate processor execution. -Purpose: Enable single-command workflow execution with automatic state management and error recovery, and and new developer to easily understand and code结构. -Output: Pipeline engine class, process subcommand, and CLI entry point. +Purpose: Establish the core orchestration engine that coordinates all processors in a unified workflow. +Output: PipelineInput/Output/Config schemas, Pipeline class with processor initialization and step execution logic. @@ -61,24 +51,24 @@ Output: Pipeline engine class, process subcommand, and CLI entry point. @.planning/PROJECT.md @.planning/ROADMAP.md @.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-context.md +@.planning/phases/04-analysis-orchestration/04-research.md -@.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md -@.planning/phases/04-analysis-orchestration/04-02-SUMMARY.md -@.planning/phases/04-analysis-orchestration/04-03-SUMMARY.md +@.planning/phases/04-analysis-orchestration/04-01-summary.md +@.planning/phases/04-analysis-orchestration/04-02-summary.md +@.planning/phases/04-analysis-orchestration/04-03-summary.md - + From kb_pipeline/processors/base.py: ```python class BaseProcessor(ABC): STEP_NAME: str - async def process(self, input_data: StepInput) -> StepOutput: + async def process(self, input_data: StepInput) -> StepOutput ``` -From kb_pipeline/processors/parse.py: +from kb_pipeline/processors/parse.py: ```python class ParseProcessor(BaseProcessor): STEP_NAME = "parse" @@ -103,264 +93,45 @@ from kb_pipeline/processors/analyze.py: ```python class AnalyzeProcessor(BaseProcessor): STEP_NAME = "analyze" - async def process(self, input_data: AnalyInput) -> AnalyOutput + async def process(self, input_data: AnalyzeInput) -> AnalyzeOutput ``` from kb_pipeline/schemas.py: ```python class StepInput(BaseModel): - content_id: str = platform: Literal["bilibili", "zhihu"] + content_id: str = Field(..., min_length=1) + platform: Literal["bilibili", "zhihu"] class StepOutput(BaseModel): success: bool - error_message: Optional[str] - completed_at: datetime + error_message: Optional[str] = None + completed_at: datetime = Field(default_factory=datetime.now) class ParseInput(StepInput): - url: str + url: str = Field(..., min_length=1, description="URL to parse") class ParseOutput(StepOutput): - parsed: ParsedLink - content_dir: Path - metadata_file: Optional[str] = None - content_file: Optional[str] = None - content_type: str = "video" # video needs AS - - # article不需要 - - content_file: Optional[str] = None # 从 download步骤获取 - - content_type: str # "video" or "article" - download_input.content_dir - download_output( - success=True, - metadata_file=f"{content_dir}/{platform}_{content_id}_metadata.json", - content_file=content_file, - title=metadata.get('title', '未知'), - author=metadata.get('author', '未知'), - duration=metadata.get('duration') # seconds, ) - - return state.mark_step_completed( - self.STEP_NAME, f"title={title}, author={author}, type={content_type}" - ) - self.state_manager.save(state) - return output - + platform: str + content_type: str + content_id: str + resolved_url: str + original_url: str ``` -from kb_pipeline.state import WorkflowStateManager -from kb_pipeline.config import get_settings -from kb_pipeline.errors import PipelineError - -from kb_pipeline.processors.parse import ParseProcessor -from kb_pipeline.processors.download import DownloadProcessor -from kb_pipeline.processors.asr import ASRProcessor -from kb_pipeline.processors.analyze import AnalyzeProcessor - -from typing import List, Optional, Dict, Any, TYPE_CHECKING -from pathlib import Path -from datetime import datetime - -import logging - -logger = logging.getLogger(__name__) - - -class PipelineConfig: - """Pipeline configuration""" - steps: List[str] = Field(default_factory=lambda: ["parse", "download", "asr", "analyze"]) - output_dir: Path - skip_completed: bool = Field(default=True, description="Skip completed steps") - max_retries: int = Field(default=3, description="Max retries for each step") - - -class Pipeline: - """Pipeline engine - orchestrates multiple processors - - Orchestrates the workflow: - 1. Parse URL to extract platform and content ID - 2. Run processors in order (parse, download, asr, analyze) - 3. Handle errors and retry on failures - 4. Return results - - Key decisions: - - Sequential execution (not parallel - simpler to easier to debug - - State persistence after each step (via WorkflowStateManager) - - Progress logging to simple, clear - """ - - def __init__( - self, - processors: List[BaseProcessor], - config: Optional[PipelineConfig] = None, - output_dir: Optional[Path] = None, - skip_completed: bool = False - ): - self.processors = processors - self.config = config or PipelineConfig.default() - self.output_dir = output_dir - - self.skip_completed = skip_completed - - # Validate processors - for processor in self.processors: - if processor.STEP_NAME not in self.config.steps: - raise ValueError(f"Unknown processor: {processor.STEP_NAME}") - - async def run(self, url: str) -> Dict[str, Any]: - """Execute full workflow - - Args: - url: URL to process - - Returns: - Dict containing all step results - """ - results: Dict[str, Any] = {} - current_step = 0 - total_steps = len(self.config.steps) - - state: Optional[WorkflowState] = None - - try: - for step_name in self.config.steps: - current_step += 1 - logger.info(f"[Step {current_step}/{total_steps}] {step_name}...") - - - # Get processor - processor = self.processors.get(step_name) - if not processor: - raise ValueError(f"Processor not found: {step_name}") - - # Get content directory from parsed URL - parsed = await self._parse_url(url) - if not parsed: - results["error"] = "Failed to parse URL" - return results - - # Get or create content directory - content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}" - content_dir.mkdir(parents=True, exist_ok=True) - - # Initialize state manager for this content - state_manager = WorkflowStateManager(content_dir) - - # Run each step - previous_result = None - for step_name in self.config.steps: - # Check if already completed - if state and state.is_step_completed(step_name): - current_step += 1 - logger.info(f"[Step {current_step}/{total_steps}] {step_name} - skipping (already completed)") - results[step_name] = {"status": "skipped", "message": "Already completed"} - continue - - # Create step input - input_data = self._create_step_input( - step_name=step_name, - parsed=parsed, - state=state, - previous_result=previous_result - ) - - # Run processor - try: - output = await processor.process(input_data) - results[step_name] = { - "status": "success" if output.success else "failed", - "output": output.model_dump() if output.success else None, - "error": output.error_message - } - previous_result = output - except Exception as e: - results[step_name] = { - "status": "failed", - "error": str(e) - } - logger.error(f"[Step {current_step}/{total_steps}] {step_name} - failed: {e}") - # Don't stop pipeline on failure - raise - - return results - - def _parse_url(self, url: str) -> Optional[ParsedLink]: - """Parse URL using LinkParser""" - if not url: - return None - - parser = LinkParser() - parsed = parser.parse(url) - if not parsed: - logger.warning(f"Failed to parse URL: {url}") - return None - - return ParsedLink - - def _create_step_input( - self, - step_name: str, - parsed: ParsedLink, - state: Optional[WorkflowState], - previous_result: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Create step input based on step name and previous result""" - step_map = { - "parse": ParseProcessor, - "download": DownloadProcessor, - "asr": ASRProcessor, - "analyze": AnalyzeProcessor, - } - - # Get processor instance - processor = self.processors[step_name] - if not processor: - raise ValueError(f"Unknown processor: {step_name}") - - return step_map - - async def _run_step( - self, - step_name: str, - input_data: Dict[str, Any], - state: Optional[WorkflowState], - previous_result: Optional[Dict[str, Any]] - ) -> StepOutput: - """Run a single step""" - processor = self.processors[step_name] - if not processor: - raise ValueError(f"No processor for {step_name}") - - # Check if already completed - if state and state.is_step_completed(step_name): - logger.info(f"Step {step_name} already completed, skipping") - return self._create_skipped_output() - - - try: - # Run the - if previous_result and previous_result.get(step_name): - input_data = previous_result[step_name] - else: - raise ValueError(f"No previous result for {step_name}") - - return await processor.process(input_data) - except Exception as e: - logger.error(f"Step {step_name} failed: {e}") - # Record failure - if state: - state.mark_step_failed(step_name, str(e)) - self.state_manager.save(state) - raise - - # Update previous result - previous_result[step_name] = output.model_dump() - - return results - - def get_output_dir(self) -> Path: - return Path(self.output_dir) / Path(self.config.output_dir) +from kb_pipeline/state.py: +```python +class WorkflowStateManager: + STATE_FILENAME = "workflow_state.json" + def __init__(self, content_dir: Path): ... + def load(self) -> Optional[WorkflowState]: ... + def save(self, state: WorkflowState) -> None: ... + def get_or_create(self, content_id: str, platform: str, source_url: str) -> WorkflowState: ... +``` +from kb_pipeline/config.py: +```python +def get_settings() -> KBSettings: ... +``` @@ -371,29 +142,52 @@ class Pipeline: Task 1: Define Pipeline Schemas kb_pipeline/schemas.py, tests/test_pipeline/test_pipeline.py - - Test 1: PipelineInput validates required fields - - Test 2: PipelineOutput validates required fields - - Test 3: PipelineConfig default steps are correct + - Test 1: PipelineInput validates url field (required) + - Test 2: PipelineOutput contains results dict + - Test 3: PipelineConfig has default steps list + - Test 4: PipelineConfig allows custom output_dir Add PipelineInput, PipelineOutput, and PipelineConfig to kb_pipeline/schemas.py. -**PipelineInput fields (extends nothing - all fields inherited from StepInput):** -- content_dir: str - Directory to save downloaded content +**PipelineInput fields:** +```python +class PipelineInput(BaseModel): + """Pipeline input - URL to process""" + url: str = Field(..., min_length=1, description="URL to process") + output_dir: Optional[Path] = Field(None, description="Output directory for content") + skip_completed: bool = Field(default=True, description="Skip already completed steps") +``` -- content_type: str - Type of content (video, article, answer, zhuanlan) +**PipelineOutput fields:** +```python +class PipelineOutput(BaseModel): + """Pipeline output - results from all steps""" + success: bool + error_message: Optional[str] = None + results: Dict[str, Any] = Field(default_factory=dict) + content_dir: Optional[Path] = None +``` -- skip_completed: bool - Whether to skip already completed steps (default True, content_type not to ["video", "article", "answer", "zhuanlan", "download", "asr", "analyze"] +**PipelineConfig fields:** +```python +class PipelineConfig(BaseModel): + """Pipeline configuration""" + steps: List[str] = Field( + default_factory=lambda: ["parse", "download", "asr", "analyze"], + description="Steps to execute in order" + ) + output_dir: Optional[Path] = Field(None, description="Base output directory") + skip_completed: bool = Field(default=True, description="Skip completed steps") + max_retries: int = Field(default=3, description="Max retries per step") +``` -- steps: List[str] = Field(...) - - output_dir: Path - skip_completed: bool = True +Create test file tests/test_pipeline/test_pipeline.py with schema validation tests. - pytest tests/test_pipeline/test_pipeline.py -v -x -k"test_schema" in + pytest tests/test_pipeline/test_pipeline.py -v -x -k "test_schema" - PipelineInput/Output schemas defined and validated + PipelineInput/Output/Config schemas defined and validated @@ -402,257 +196,102 @@ Add PipelineInput, PipelineOutput, and PipelineConfig to kb_pipeline/schemas.py. - Test 1: Pipeline initializes with all processors - Test 2: Pipeline.parse_url creates correct content directory - - Test 3: Pipeline.run() executes all steps in correct order - - Test 4: Pipeline.run() handles step skipping correctly - - Test 5: Pipeline.run() handles errors and continues to next step - - Test 6: Pipeline.run() records state for each step + - Test 3: Pipeline validates processor STEP_NAME matches config.steps + - Test 4: Pipeline handles missing processor gracefully Create `kb_pipeline/pipeline.py` with Pipeline class. **Implementation:** -1. Initialize with processors list (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor) -2. Set config with PipelineConfig.default() -3. Set output_dir -4. Initialize state_manager (WorkflowStateManager) -5. Implement run() method that: - - Creates content directory - - Runs steps in order (parse, download, asr, analyze) - - Returns results dict +```python +from typing import List, Optional, Dict, Any +from pathlib import Path +import logging -6. Add helper methods: - - _parse_url(url) -> ParsedLink - - _get_content_dir(parsed) - - _run_step(step_name, parsed, state, previous_result) - - _record_state +from kb_pipeline.config import get_settings +from kb_pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.link_parser import LinkParser, ParsedLink +from kb_pipeline.processors.base import BaseProcessor -7. Add error handling with try/except - - logger.error for error message - if self.state_manager: - state.mark_step_failed(step_name, str(e)) - self.state_manager.save(state) - raise +logger = logging.getLogger(__name__) - def _parse_url(self, url: str) -> Optional[ParsedLink]: + +class Pipeline: + """Pipeline engine - orchestrates multiple processors + + Orchestrates the workflow: + 1. Parse URL to extract platform and content ID + 2. Run processors in order (parse, download, asr, analyze) + 3. Handle errors and state persistence + 4. Return results + + Key design decisions: + - Sequential execution (simpler, easier to debug) + - State persistence after each step (via WorkflowStateManager) + - Simple progress logging + """ + + def __init__( + self, + processors: List[BaseProcessor], + config: Optional[PipelineConfig] = None, + output_dir: Optional[Path] = None, + ): + self.processors = {p.STEP_NAME: p for p in processors} + self.config = config or PipelineConfig() + self.output_dir = output_dir or Path.cwd() + self.settings = get_settings() + + # Validate processors + for step_name in self.processors: + if step_name not in self.config.steps: + raise ValueError(f"Unknown processor: {step_name}") + + def _parse_url(self, url: str) -> Optional[ParsedLink]: """Parse URL using LinkParser""" if not url: return None - parser = LinkParser() parsed = parser.parse(url) if not parsed: + logger.warning(f"Failed to parse URL: {url}") return None - - return Parsed + return parsed def _get_content_dir(self, parsed: ParsedLink) -> Path: """Get content directory from parsed URL""" content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}" content_dir.mkdir(parents=True, exist_ok=True) return content_dir +``` - def _create_step_input( - self, - step_name: str, - parsed: ParsedLink, - state: Optional[WorkflowState], - previous_result: Optional[Dict[str, Any]] - ) -> Dict[str, Any]: - """Create step input based on step name and previous result""" - step_map = { - "parse": ParseProcessor, - "download": DownloadProcessor, - "asr": ASRProcessor, - "analyze": AnalyzeProcessor - } - - # Get processor instance - processor = self.processors[step_name] - if not processor: - raise ValueError(f"Unknown processor: {step_name}") - - return step_map - - async def _run_step( - self, - step_name: str, - input_data: Dict[str, Any], - state: Optional[WorkflowState], - previous_result: Optional[Dict[str, Any]], - ) -> StepOutput: - """Run a single step""" - processor = self.processors[step_name] - if not processor: - raise ValueError(f"No processor for {step_name}") - - # Check if already completed - if state and state.is_step_completed(step_name): - logger.info(f"Step {step_name} already completed, skipping") - return self._create_skipped_output() - - try: - # Run step - if previous_result and previous_result.get(step_name): - input_data = previous_result[step_name] - else: - raise ValueError(f"No previous result for {step_name}") - - return await processor.process(input_data) - except Exception as e: - logger.error(f"Step {step_name} failed: {e}") - if state: - state.mark_step_failed(step_name, str(e)) - self.state_manager.save(state) - raise - - # Update previous result - previous_result[step_name] = output.model_dump() - return results - - def _create_skipped_output(self) -> StepOutput: - return StepOutput(success=True, error_message="Skipped: already completed") - - - - - Task 3: Add step execution logic - kb_pipeline/pipeline.py - -Extend Pipeline.run() to implement step execution logic: - -1. Create content directory -2. Run steps in order -3. Handle step skipping (check state) -4. Handle errors (try/except, log and continue) -5. Record final state -6. Return results - -**Step execution logic:** -```python -async def _execute_steps( - self, - parsed: ParsedLink, - state: WorkflowState, - previous_result: Dict[str, Any] -) -> None: - for step_name in self.config.steps: - step_start = datetime.now() - state.mark_step_started(step_name) - - logger.info(f"[Step {step_num}/{total_steps}] {step_name}...") - - # Get or run processor - processor = self.processors[step_name] - if not processor: - raise ValueError(f"Unknown processor: {step_name}") - - # Get step input - input_data = self._create_step_input( - step_name=step_name, - parsed=parsed, - state=state, - previous_result=previous_result.get(step_name) - ) - - # Run processor - try: - output = await processor.process(input_data) - previous_result[step_name] = output.model_dump() - results[step_name] = { - "status": "success" if output.success else "failed", - "output": output.model_dump() if output.success else None, - "error": output.error_message - } - previous_result = output - except Exception as e: - results[step_name] = { - "status": "failed", - "error": str(e) - } - logger.error(f"[Step {step_num}/{total_steps}] {step_name} - failed: {e}") - # Record failure - if state: - state.mark_step_failed(step_name, str(e)) - self.state_manager.save(state) - # Stop on failure - raise - - return results +Create comprehensive tests with mocked processors. - pytest tests/test_pipeline/test_pipeline.py -v -x + pytest tests/test_pipeline/test_pipeline.py -v -x -k "test_init or test_parse" - Pipeline engine executes workflow with state persistence - - - - Task 4: Add CLI process command - kb-workflow/kb_workflow.py - -Extend kb-workflow/kb_workflow.py to add process subcommand. - -**CLI additions:** -- New argument: 'url' with help='URL to process' -- New argument: '--output-dir', '-o' with help='Output directory (default: KB_BASE_PATH/content) -- New argument: '--no-analyze', action='store_true', help='Skip analysis step') -- Set default to process function that handles the command - -- Call Pipeline().run(url, output_dir, skip_completed=no) -- Print success message -- If error: print error message and exit(1) - -**Update existing help text and kb_workflow.py to document the new command. - - - python kb-workflow/kb_workflow.py --help | grep process - - CLI process command added and functional - - - - Task 5: Create Pipeline tests - tests/test_pipeline/test_pipeline.py,files> - -Create comprehensive tests for Pipeline class. - -**Test cases:** -1. Test pipeline initialization -2. Test URL parsing and content dir creation -3. Test step execution order -4. Test step skipping -5. Test error handling and6. Test state persistence - -7. Test full workflow (mocked) - -Use fixtures for mocking processors, state_manager, settings. - - - pytest tests/test_pipeline/test_pipeline.py -v - - Pipeline class fully tested + Pipeline class initializes correctly and processor validation -- All tests pass: `pytest tests/test_pipeline/test_pipeline.py tests/test_cli/test_process_command.py -v` -- Pipeline.run() executes workflow correctly -- State persistence works -- Error handling works -- Step skipping works +- All tests pass: `pytest tests/test_pipeline/test_pipeline.py -v -x` +- PipelineInput/Output/Config schemas exist and validate +- Pipeline class initializes with processors +- URL parsing creates correct content directory +- Processor validation works -- Pipeline class orchestrates all processors -- Pipeline.run() executes workflow from URL to analysis -- Content directory created correctly -- Steps executed in order -- State persisted after each step -- Errors handled with retry -- Completed steps skipped -- CLI process command works +- PipelineInput/Output/Config schemas defined in schemas.py +- Pipeline class exists in pipeline.py +- Pipeline initializes with processor list +- URL parsing creates content directory +- All tests pass -After completion, create `.planning/phases/04-analysis-orchestration/04-04-SUMMARY.md` +After completion, create `.planning/phases/04-analysis-orchestration/04-04-SUMmary.md` diff --git a/.planning/phases/04-analysis-orchestration/04-VALIDATION.md b/.planning/phases/04-analysis-orchestration/04-VALIDATION.md index f51d1ad..dd33289 100644 --- a/.planning/phases/04-analysis-orchestration/04-VALIDATION.md +++ b/.planning/phases/04-analysis-orchestration/04-VALIDATION.md @@ -38,24 +38,26 @@ created: 2026-03-13 | Task ID | Plan | Wave | Requirement | Test Type | Automated Command | File Exists | Status | |---------|------|------|-------------|-----------|-------------------|-------------|--------| -| 04-01-01 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_base.py -v` | ⬜ W0 | ⬜ pending | -| 04-01-02 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_parse.py -v` | ⬜ W0 | ⬜ pending | -| 04-02-01 | 02 | 2 | MNT-01 | unit | `pytest tests/test_processors/test_download.py -v` | ⬜ W0 | ⬜ pending | -| 04-03-01 | 03 | 3 | MNT-01 | unit | `pytest tests/test_processors/test_analyze.py -v` | ⬜ W0 | ⬜ pending | -| 04-04-01 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | ⬜ W0 | ⬜ pending | -| 04-04-02 | 04 | 4 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | ⬜ W0 | ⬜ pending | -| 04-05-01 | 05 | 5 | MNT-01 | integration | `pytest tests/test_integration/test_e2e_workflow.py -v` | ⬜ W0 | ⬜ pending | - -*Status: ⬜ pending · ✅ green · ❌ red · ⚠️ flaky* +| 04-01-01 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_base.py -v` | W0 | pending | +| 04-01-02 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_parse.py -v` | W0 | pending | +| 04-02-01 | 02 | 2 | MNT-01 | unit | `pytest tests/test_processors/test_download.py -v` | W0 | pending | +| 04-03-01 | 03 | 3 | MNT-01 | unit | `pytest tests/test_processors/test_analyze.py -v` | W0 | pending | +| 04-04-01 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | W0 | pending | +| 04-04-02 | 04 | 4 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending | +| 04-04-03 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | W0 | pending | + | 04-04-04 | 04 | 5 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending | +| 04-05-01 | 05 | 5 | MNT-01 | integration | `pytest tests/test_integration/test_e2e_workflow.py -v` | W0 | pending | + 04-04-05 | 04 | 5 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending | +*Status: pending - green - red - flaky* --- ## Wave 0 Requirements Existing infrastructure covers all phase requirements: -- `pytest.ini` — configured with pythonpath -- `tests/conftest.py` — shared fixtures exist -- `tests/test_processors/` — directory exists for processor tests +- `pytest.ini` -- configured with pythonpath +- `tests/conftest.py` -- shared fixtures exist +- `tests/test_processors/` -- directory exists for processor tests ---