"""Core Pipeline Schemas 定义工作流管道的核心 Schema,包括 StepInput、StepOutput、StepResult、 StepStatus、StepRecord、WorkflowState 和 StepDefinition。 使用 Pydantic v2 确保类型安全和自动验证。 """ from pydantic import BaseModel, Field from typing import Optional, Dict, Literal, Any, List from datetime import datetime from enum import Enum class StepStatus(str, Enum): """步骤执行状态枚举""" PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" SKIPPED = "skipped" class StepInput(BaseModel): """所有步骤输入的基类 Attributes: content_id: 内容唯一标识,如 B站 BV号、知乎回答ID等 platform: 平台名称,限定为 bilibili 或 zhihu """ content_id: str = Field(..., min_length=1, description="内容唯一标识") platform: Literal["bilibili", "zhihu"] = Field(..., description="平台名称") class StepOutput(BaseModel): """所有步骤输出的基类 Attributes: success: 步骤是否成功执行 error_message: 错误信息,成功时为 None completed_at: 完成时间,默认为当前时间 """ success: bool = Field(..., description="步骤是否成功") error_message: Optional[str] = Field(None, description="错误信息") completed_at: datetime = Field(default_factory=datetime.now, description="完成时间") class StepResult(BaseModel): """步骤执行结果 包含步骤的输入、输出和执行状态。 Attributes: step_name: 步骤名称 input_data: 步骤输入数据 output_data: 步骤输出数据,可能为 None status: 步骤执行状态 """ step_name: str = Field(..., description="步骤名称") input_data: StepInput = Field(..., description="步骤输入数据") output_data: Optional[StepOutput] = Field(None, description="步骤输出数据") status: StepStatus = Field(default=StepStatus.PENDING, description="步骤执行状态") class StepRecord(BaseModel): """单个步骤的执行记录 记录步骤执行的状态、时间和结果摘要。 Attributes: step_name: 步骤名称 status: 步骤执行状态 started_at: 开始时间 completed_at: 完成时间 error_message: 错误信息 output_summary: 输出摘要 """ step_name: str = Field(..., description="步骤名称") status: StepStatus = Field(..., description="步骤执行状态") started_at: Optional[datetime] = Field(None, description="开始时间") completed_at: Optional[datetime] = Field(None, description="完成时间") error_message: Optional[str] = Field(None, description="错误信息") output_summary: Optional[str] = Field(None, description="输出摘要") class WorkflowState(BaseModel): """工作流状态 - 存储在 workflow_state.json 管理整个工作流的状态,包括内容信息、执行历史和步骤状态。 Attributes: schema_version: Schema 版本号 content_id: 内容唯一标识 platform: 平台名称 source_url: 原始链接 created_at: 创建时间 updated_at: 更新时间 steps: 步骤执行记录字典 """ schema_version: str = Field(default="1.0", description="Schema 版本号") content_id: str = Field(..., description="内容唯一标识") platform: str = Field(..., description="平台名称") source_url: str = Field(..., description="原始链接") created_at: datetime = Field(default_factory=datetime.now, description="创建时间") updated_at: datetime = Field(default_factory=datetime.now, description="更新时间") steps: Dict[str, StepRecord] = Field(default_factory=dict, description="步骤执行记录") def is_step_completed(self, step_name: str) -> bool: """检查指定步骤是否已完成 Args: step_name: 步骤名称 Returns: 如果步骤存在且状态为 COMPLETED 返回 True,否则返回 False """ record = self.steps.get(step_name) return record is not None and record.status == StepStatus.COMPLETED def mark_step_started(self, step_name: str) -> None: """标记步骤开始执行 Args: step_name: 步骤名称 """ self.steps[step_name] = StepRecord( step_name=step_name, status=StepStatus.RUNNING, started_at=datetime.now() ) self.updated_at = datetime.now() def mark_step_completed(self, step_name: str, output_summary: Optional[str] = None) -> None: """标记步骤完成 Args: step_name: 步骤名称 output_summary: 输出摘要(可选) """ if step_name in self.steps: self.steps[step_name].status = StepStatus.COMPLETED self.steps[step_name].completed_at = datetime.now() self.steps[step_name].output_summary = output_summary else: self.steps[step_name] = StepRecord( step_name=step_name, status=StepStatus.COMPLETED, completed_at=datetime.now(), output_summary=output_summary ) self.updated_at = datetime.now() def mark_step_failed(self, step_name: str, error_message: str) -> None: """标记步骤失败 Args: step_name: 步骤名称 error_message: 错误信息 """ if step_name in self.steps: self.steps[step_name].status = StepStatus.FAILED self.steps[step_name].error_message = error_message self.steps[step_name].completed_at = datetime.now() else: self.steps[step_name] = StepRecord( step_name=step_name, status=StepStatus.FAILED, error_message=error_message, completed_at=datetime.now() ) self.updated_at = datetime.now() class StepDefinition(BaseModel): """步骤定义 - 描述步骤的前置条件、输出格式、成功标准 Attributes: step_name: 步骤名称 description: 步骤描述 input_schema: 输入 Schema 类型 output_schema: 输出 Schema 类型 dependencies: 依赖的步骤列表 """ step_name: str = Field(..., description="步骤名称") description: str = Field(..., description="步骤描述") input_schema: type[StepInput] = Field(..., description="输入 Schema 类型") output_schema: type[StepOutput] = Field(..., description="输出 Schema 类型") dependencies: list[str] = Field(default_factory=list, description="依赖的步骤列表") # ============================================================================= # ASR Schemas (Phase 3) # ============================================================================= class ASRInput(StepInput): """ASR 步骤输入 继承 StepInput 的基础字段,添加视频路径和内容类型。 Attributes: video_path: 视频文件路径 content_type: 内容类型,默认为 'video' """ video_path: str = Field(..., description="视频文件路径") content_type: str = Field(default="video", description="内容类型") class ASROutput(StepOutput): """ASR 步骤输出 继承 StepOutput 的基础字段,添加 ASR 特定字段。 Attributes: tool: 使用的工具 (subtitle, whisper, whisper-diarization) has_subtitle: 是否有官方字幕 output_files: 输出文件路径字典 {txt, srt, json} segments: 带时间戳的分段列表 text: 完整文本 duration_seconds: 音频时长 speakers: 说话者列表(仅说话者分离时有值) """ tool: Literal["subtitle", "whisper", "whisper-diarization"] = Field( ..., description="使用的工具" ) has_subtitle: bool = Field(..., description="是否有官方字幕") output_files: Dict[str, str] = Field( default_factory=dict, description="输出文件路径 {txt, srt, json}" ) segments: List[Dict[str, Any]] = Field( default_factory=list, description="带时间戳的分段列表" ) text: str = Field(default="", description="完整文本") duration_seconds: float = Field(default=0.0, description="音频时长") speakers: Optional[List[str]] = Field( None, description="说话者列表(仅说话者分离时有值)" ) # ============================================================================= # Parse Schemas (Phase 04) # ============================================================================= class ParseInput(StepInput): """Parse 步骤输入 继承 StepInput 的基础字段,添加 URL 字段。 content_id 和 platform 在解析前是空的,解析后会被填充。 Attributes: url: 要解析的 URL content_id: 解析后填充的内容 ID(初始为空) platform: 解析后填充的平台名称(初始为空) """ url: str = Field(..., min_length=1, description="URL to parse") content_id: str = Field(default="", description="Will be populated after parsing") platform: str = Field(default="", description="Will be populated after parsing") class ParseOutput(StepOutput): """Parse 步骤输出 继承 StepOutput 的基础字段,添加解析特定字段。 Attributes: platform: 检测到的平台 (bilibili, zhihu) content_type: 内容类型 (video, article, answer, zhuanlan) content_id: 提取的内容 ID resolved_url: 短链解析后的完整 URL original_url: 用户提供的原始 URL """ platform: str = Field(..., description="Detected platform") content_type: str = Field(..., description="Content type (video, article, etc.)") content_id: str = Field(..., description="Extracted content ID") resolved_url: str = Field(..., description="Resolved URL after short link expansion") original_url: str = Field(..., description="Original URL provided by user") # ============================================================================= # Download Schemas (Phase 04) # ============================================================================= class DownloadInput(StepInput): """Download 步骤输入 继承 StepInput 的基础字段,添加下载特定字段。 Attributes: content_dir: 保存下载内容的目录路径 content_type: 内容类型 (video, article, answer, zhuanlan) """ content_dir: str = Field(..., min_length=1, description="Directory to save downloaded content") content_type: str = Field(..., min_length=1, description="Content type (video, article, answer, zhuanlan)") class DownloadOutput(StepOutput): """Download 步骤输出 继承 StepOutput 的基础字段,添加下载特定字段。 Attributes: metadata_file: 元数据 JSON 文件路径 content_file: 内容文件路径 (视频/音频或文本) title: 内容标题 author: 内容作者 duration: 视频内容的时长(秒),仅视频有值 """ metadata_file: str = Field(..., description="Path to metadata JSON file") content_file: str = Field(..., description="Path to content file (video/audio or text)") title: str = Field(..., description="Content title") author: str = Field(..., description="Content author") duration: Optional[int] = Field(None, description="Duration for video content in seconds")