"""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