diff --git a/kb_workflow/pipeline/pipeline.py b/kb_workflow/pipeline/pipeline.py index 60d57fc..f96f62d 100644 --- a/kb_workflow/pipeline/pipeline.py +++ b/kb_workflow/pipeline/pipeline.py @@ -21,7 +21,7 @@ from typing import Dict, List, Optional from kb_workflow.pipeline.config import get_settings from kb_workflow.pipeline.link_parser import LinkParser, ParsedLink from kb_workflow.pipeline.processors.base import BaseProcessor -from kb_workflow.pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput +from kb_workflow.pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput, SRTInput logger = logging.getLogger(__name__) @@ -174,6 +174,20 @@ class Pipeline: logger.info(f"Skipping ASR for text content type: {parsed.content_type}") continue + # B站视频:跳过 analyze 步骤,使用 srt 步骤替代 + is_bilibili_video = ( + parsed.platform == "bilibili" and + parsed.content_type == "video" + ) + if step_name == "analyze" and is_bilibili_video: + logger.info(f"Skipping analyze for Bilibili video, using SRT processor instead") + continue + + # B站文章/知乎等:跳过 srt 步骤 + if step_name == "srt" and not is_bilibili_video: + logger.info(f"Skipping SRT for non-Bilibili-video content: {parsed.platform}/{parsed.content_type}") + continue + processor = self.processors[step_name] # Mark step as started @@ -228,10 +242,26 @@ class Pipeline: content_type=parsed.content_type, output_file=str(content_dir / "analysis.md"), ) + elif step_name == "srt": + # SRT needs srt file from ASR step and metadata + download_result = results.get("download", {}) + asr_result = results.get("asr", {}) + metadata_file = download_result.get("metadata_file", "") + # SRT file from ASR output + srt_file = asr_result.get("output_files", {}).get("srt", "") + + step_input = SRTInput( + content_id=parsed.content_id, + platform=parsed.platform, + srt_file=srt_file, + metadata_file=metadata_file, + output_file=str(content_dir / "analysis.md"), + content_type=parsed.content_type, + ) elif step_name == "archive": - # Archive needs analysis file from analyze step - analyze_result = results.get("analyze", {}) - analysis_file = analyze_result.get("analysis_file", "") + # Archive needs analysis file from analyze step (or srt step for Bilibili video) + analyze_result = results.get("analyze", {}) or results.get("srt", {}) + analysis_file = analyze_result.get("analysis_file", "") or analyze_result.get("output_file", "") step_input = ArchiveInput( content_id=parsed.content_id, @@ -261,7 +291,7 @@ class Pipeline: # Add output fields to results for field_name in ["metadata_file", "content_file", "title", "author", "analysis_file", "summary", "output_files", - "archive_file", "skipped"]: + "archive_file", "skipped", "output_file", "paragraphs", "tags", "corrected"]: if hasattr(step_output, field_name): results[step_name][field_name] = getattr(step_output, field_name) @@ -397,3 +427,14 @@ class Pipeline: "analysis_file": str(analysis_path), } logger.info(f"Found existing analysis: {analysis_path}") + + elif step_name == "srt": + # Scan for SRT output (same file as analyze - analysis.md) + analysis_path = content_dir / "analysis.md" + if analysis_path.exists(): + results["srt"] = { + "success": True, + "error_message": None, + "output_file": str(analysis_path), + } + logger.info(f"Found existing SRT output: {analysis_path}")