From ec2d12a05b1433228d8f9902f64ddd0379da6e83 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:46:54 +0800 Subject: [PATCH] feat(04-05): implement Pipeline.run() method - Add async run() method to Pipeline class - Execute processors sequentially with state persistence - Handle step-specific input types (Parse, Download, ASR, Analyze) - Support skip_completed for resumable workflows - Return PipelineOutput with results and content_dir Co-Authored-By: Claude Opus 4.6 --- kb_pipeline/pipeline.py | 162 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) diff --git a/kb_pipeline/pipeline.py b/kb_pipeline/pipeline.py index 2fc54fd..111caf1 100644 --- a/kb_pipeline/pipeline.py +++ b/kb_pipeline/pipeline.py @@ -102,3 +102,165 @@ class Pipeline: content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}" content_dir.mkdir(parents=True, exist_ok=True) return content_dir + + async def run(self, url: str) -> PipelineOutput: + """Run the pipeline for a given URL. + + Executes all configured processors in order: + 1. Parse URL to get platform and content_id + 2. Create content directory + 3. Run each processor sequentially + 4. Handle errors and state persistence + 5. Return results + + Args: + url: URL to process + + Returns: + PipelineOutput with success status, results, and content_dir + """ + # 1. Parse URL + parsed = self._parse_url(url) + if not parsed: + return PipelineOutput( + success=False, + error_message=f"Failed to parse URL: {url}", + results={}, + content_dir=None, + ) + + # 2. Create content directory + content_dir = self._get_content_dir(parsed) + + # 3. Initialize state manager + from kb_pipeline.state import WorkflowStateManager + state_manager = WorkflowStateManager(content_dir) + + # Get or create workflow state + state = state_manager.get_or_create( + content_id=parsed.content_id, + platform=parsed.platform, + source_url=url, + ) + + # 4. Run processors in order + results: Dict = {} + + for step_name in self.config.steps: + if step_name not in self.processors: + logger.warning(f"Step {step_name} not found in processors, skipping") + continue + + # Check if step should be skipped + if self.config.skip_completed and state.is_step_completed(step_name): + logger.info(f"Skipping completed step: {step_name}") + continue + + processor = self.processors[step_name] + + # Mark step as started + state.mark_step_started(step_name) + state_manager.save(state) + + try: + # Create step input based on step type + from kb_pipeline.schemas import ( + ParseInput, DownloadInput, ASRInput, AnalyzeInput + ) + + if step_name == "parse": + step_input = ParseInput( + url=url, + content_id=parsed.content_id, + platform=parsed.platform, + ) + elif step_name == "download": + step_input = DownloadInput( + content_id=parsed.content_id, + platform=parsed.platform, + content_dir=str(content_dir), + content_type=parsed.content_type, + ) + elif step_name == "asr": + # ASR needs video path from download step + download_result = results.get("download", {}) + video_path = download_result.get("content_file", "") + step_input = ASRInput( + content_id=parsed.content_id, + platform=parsed.platform, + video_path=video_path, + content_type=parsed.content_type, + ) + elif step_name == "analyze": + # Analyze needs metadata and content files + download_result = results.get("download", {}) + asr_result = results.get("asr", {}) + metadata_file = download_result.get("metadata_file", "") + # Use subtitle JSON if available, otherwise content file + content_file = asr_result.get("output_files", {}).get("json", "") or download_result.get("content_file", "") + + step_input = AnalyzeInput( + content_id=parsed.content_id, + platform=parsed.platform, + metadata_file=metadata_file, + content_file=content_file, + content_type=parsed.content_type, + output_file=str(content_dir / "analysis.md"), + ) + else: + # Generic step input + from kb_pipeline.schemas import StepInput + step_input = StepInput( + content_id=parsed.content_id, + platform=parsed.platform, + ) + + # Run processor + step_output = await processor.process(step_input) + + # Record result + results[step_name] = { + "success": step_output.success, + "error_message": step_output.error_message, + } + + if step_output.success: + # Add output fields to results + for field_name in ["metadata_file", "content_file", "title", "author", + "analysis_file", "summary", "output_files"]: + if hasattr(step_output, field_name): + results[step_name][field_name] = getattr(step_output, field_name) + + state.mark_step_completed(step_name) + else: + state.mark_step_failed(step_name, step_output.error_message or "Unknown error") + # Stop pipeline on error + state_manager.save(state) + return PipelineOutput( + success=False, + error_message=f"Step {step_name} failed: {step_output.error_message}", + results=results, + content_dir=content_dir, + ) + + except Exception as e: + logger.error(f"Error in step {step_name}: {e}") + state.mark_step_failed(step_name, str(e)) + state_manager.save(state) + return PipelineOutput( + success=False, + error_message=f"Step {step_name} raised exception: {e}", + results=results, + content_dir=content_dir, + ) + + # Save state after each successful step + state_manager.save(state) + + # 5. Return success + return PipelineOutput( + success=True, + error_message=None, + results=results, + content_dir=content_dir, + )