From 0dfb51e9aa3d16255cc3c3275232bba0a941b249 Mon Sep 17 00:00:00 2001
From: starmen1265 <starmen1265@163.com>
Date: Fri, 13 Mar 2026 23:38:17 +0800
Subject: [PATCH] feat(04-04): create Pipeline class with processor
 initialization

- Pipeline class initializes with processors list
- Validates processor STEP_NAME matches config.steps
- _parse_url method parses URLs using LinkParser
- _get_content_dir creates content directory from parsed URL

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 kb_pipeline/pipeline.py | 104 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 104 insertions(+)
 create mode 100644 kb_pipeline/pipeline.py

diff --git a/kb_pipeline/pipeline.py b/kb_pipeline/pipeline.py
new file mode 100644
index 0000000..2fc54fd
--- /dev/null
+++ b/kb_pipeline/pipeline.py
@@ -0,0 +1,104 @@
+"""Pipeline engine - orchestrates multiple processors
+
+The Pipeline class coordinates 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
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from kb_pipeline.config import get_settings
+from kb_pipeline.link_parser import LinkParser, ParsedLink
+from kb_pipeline.processors.base import BaseProcessor
+from kb_pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput
+
+logger = logging.getLogger(__name__)
+
+
+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,
+    ):
+        """Initialize the Pipeline.
+
+        Args:
+            processors: List of processor instances to use
+            config: Optional PipelineConfig for pipeline settings
+            output_dir: Optional output directory for content
+
+        Raises:
+            ValueError: If a processor's STEP_NAME is not in config.steps
+        """
+        self.processors: Dict[str, BaseProcessor] = {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 - each processor's STEP_NAME must be in config.steps
+        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.
+
+        Args:
+            url: URL to parse
+
+        Returns:
+            ParsedLink with extracted information, or None if URL is not recognized
+        """
+        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
+
+    def _get_content_dir(self, parsed: ParsedLink) -> Path:
+        """Get content directory from parsed URL.
+
+        Creates the directory if it doesn't exist.
+
+        Args:
+            parsed: ParsedLink with platform and content_id
+
+        Returns:
+            Path to the content directory
+        """
+        content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
+        content_dir.mkdir(parents=True, exist_ok=True)
+        return content_dir