feat(06-02): move pipeline core files to kb_workflow.pipeline
- Move pipeline.py, schemas.py, config.py, state.py, errors.py - Move retry.py, idempotency.py, link_parser.py - Create pipeline/__init__.py with updated imports - Update all internal imports to use kb_workflow.pipeline prefix Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,446 @@
|
||||
"""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
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
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: 视频文件路径（可能为空，用于 Whisper 降级）
|
||||
content_type: 内容类型，默认为 'video'
|
||||
content_dir: 内容目录路径，用于输出字幕文件
|
||||
"""
|
||||
video_path: str = Field(default="", description="视频文件路径（可能为空）")
|
||||
content_type: str = Field(default="video", description="内容类型")
|
||||
content_dir: str = Field(..., 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)
|
||||
source_url: 原始 URL（可选，用于知乎回答等需要完整 URL 的情况）
|
||||
"""
|
||||
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)")
|
||||
source_url: Optional[str] = Field(None, description="Original source URL")
|
||||
|
||||
|
||||
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")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Analyze Schemas (Phase 04)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class AnalyzeInput(StepInput):
|
||||
"""Analyze 步骤输入
|
||||
|
||||
继承 StepInput 的基础字段，添加分析特定字段。
|
||||
|
||||
Attributes:
|
||||
metadata_file: 元数据 JSON 文件路径
|
||||
content_file: 内容文件路径（字幕 JSON 或文章内容）
|
||||
content_type: 内容类型（video 或 article）
|
||||
output_file: 输出分析 Markdown 文件路径
|
||||
"""
|
||||
metadata_file: str = Field(..., min_length=1, description="Path to metadata JSON file")
|
||||
content_file: str = Field(..., min_length=1, description="Path to content file (subtitle JSON or article content)")
|
||||
content_type: str = Field(..., min_length=1, description="Content type (video or article)")
|
||||
output_file: str = Field(..., min_length=1, description="Path to output analysis Markdown file")
|
||||
|
||||
|
||||
class AnalyzeOutput(StepOutput):
|
||||
"""Analyze 步骤输出
|
||||
|
||||
继承 StepOutput 的基础字段，添加分析特定字段。
|
||||
|
||||
Attributes:
|
||||
analysis_file: 生成的分析报告文件路径
|
||||
summary: 内容摘要
|
||||
main_topics: 提取的主要主题列表
|
||||
key_insights: 关键见解列表
|
||||
tags: 推荐标签列表
|
||||
stats: 分析统计信息字典
|
||||
"""
|
||||
analysis_file: str = Field(..., description="Path to generated analysis report")
|
||||
summary: str = Field(default="", description="Content summary")
|
||||
main_topics: List[str] = Field(default_factory=list, description="Main topics extracted")
|
||||
key_insights: List[str] = Field(default_factory=list, description="Key insights")
|
||||
tags: List[str] = Field(default_factory=list, description="Recommended tags")
|
||||
stats: Dict[str, Any] = Field(default_factory=dict, description="Statistics about analysis")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Pipeline Schemas (Phase 04-04)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class PipelineInput(BaseModel):
|
||||
"""Pipeline input - URL to process
|
||||
|
||||
Attributes:
|
||||
url: URL to process (required)
|
||||
output_dir: Optional output directory for content
|
||||
skip_completed: Skip already completed steps (default True)
|
||||
"""
|
||||
url: str = Field(..., min_length=1, description="URL to process")
|
||||
output_dir: Optional[Path] = Field(None, description="Output directory for content")
|
||||
skip_completed: bool = Field(default=True, description="Skip already completed steps")
|
||||
|
||||
|
||||
class PipelineOutput(BaseModel):
|
||||
"""Pipeline output - results from all steps
|
||||
|
||||
Attributes:
|
||||
success: Whether the pipeline completed successfully
|
||||
error_message: Error message if pipeline failed
|
||||
results: Dictionary of results from each step
|
||||
content_dir: Path to the content directory
|
||||
"""
|
||||
success: bool = Field(..., description="Whether the pipeline completed successfully")
|
||||
error_message: Optional[str] = Field(None, description="Error message if pipeline failed")
|
||||
results: Dict[str, Any] = Field(default_factory=dict, description="Results from each step")
|
||||
content_dir: Optional[Path] = Field(None, description="Path to the content directory")
|
||||
|
||||
|
||||
class PipelineConfig(BaseModel):
|
||||
"""Pipeline configuration
|
||||
|
||||
Attributes:
|
||||
steps: Steps to execute in order (default: parse, download, asr, analyze, archive)
|
||||
output_dir: Base output directory
|
||||
skip_completed: Skip completed steps (default True)
|
||||
max_retries: Max retries per step (default 3)
|
||||
"""
|
||||
steps: List[str] = Field(
|
||||
default_factory=lambda: ["parse", "download", "asr", "analyze", "archive"],
|
||||
description="Steps to execute in order"
|
||||
)
|
||||
output_dir: Optional[Path] = Field(None, description="Base output directory")
|
||||
skip_completed: bool = Field(default=True, description="Skip completed steps")
|
||||
max_retries: int = Field(default=3, description="Max retries per step")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Archive Schemas (Phase 5)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class ArchiveInput(StepInput):
|
||||
"""Archive 步骤输入
|
||||
|
||||
继承 StepInput 的基础字段，添加归档特定字段。
|
||||
|
||||
Attributes:
|
||||
content_dir: 内容目录路径（中间目录）
|
||||
analysis_file: 分析报告 md 文件路径
|
||||
content_type: 内容类型
|
||||
"""
|
||||
content_dir: str = Field(..., min_length=1, description="Content directory path (intermediate)")
|
||||
analysis_file: str = Field(..., min_length=1, description="Path to analysis Markdown file")
|
||||
content_type: str = Field(..., min_length=1, description="Content type (video, article, etc.)")
|
||||
|
||||
|
||||
class ArchiveOutput(StepOutput):
|
||||
"""Archive 步骤输出
|
||||
|
||||
继承 StepOutput 的基础字段，添加归档特定字段。
|
||||
|
||||
Attributes:
|
||||
archive_file: 归档后的文件路径
|
||||
skipped: 是否跳过（已有相同文件）
|
||||
platform: 平台名称
|
||||
"""
|
||||
archive_file: str = Field(default="", description="Path to archived file")
|
||||
skipped: bool = Field(default=False, description="Whether archiving was skipped")
|
||||
platform: str = Field(default="", description="Platform name")
|
||||
Reference in New Issue
Block a user