From 279ae684eaa0c3894af4bb986ad0dd9c0bdf2e1e Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Tue, 17 Mar 2026 00:16:56 +0800 Subject: [PATCH] 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 --- kb_workflow/pipeline/__init__.py | 64 ++++ kb_workflow/pipeline/config.py | 72 +++++ kb_workflow/pipeline/errors.py | 174 +++++++++++ kb_workflow/pipeline/idempotency.py | 80 +++++ kb_workflow/pipeline/link_parser.py | 181 +++++++++++ kb_workflow/pipeline/pipeline.py | 399 +++++++++++++++++++++++++ kb_workflow/pipeline/retry.py | 51 ++++ kb_workflow/pipeline/schemas.py | 446 ++++++++++++++++++++++++++++ kb_workflow/pipeline/state.py | 152 ++++++++++ 9 files changed, 1619 insertions(+) create mode 100644 kb_workflow/pipeline/__init__.py create mode 100644 kb_workflow/pipeline/config.py create mode 100644 kb_workflow/pipeline/errors.py create mode 100644 kb_workflow/pipeline/idempotency.py create mode 100644 kb_workflow/pipeline/link_parser.py create mode 100644 kb_workflow/pipeline/pipeline.py create mode 100644 kb_workflow/pipeline/retry.py create mode 100644 kb_workflow/pipeline/schemas.py create mode 100644 kb_workflow/pipeline/state.py diff --git a/kb_workflow/pipeline/__init__.py b/kb_workflow/pipeline/__init__.py new file mode 100644 index 0000000..3b63b8c --- /dev/null +++ b/kb_workflow/pipeline/__init__.py @@ -0,0 +1,64 @@ +"""KB Pipeline - 工作流管道基础设施 + +提供内容处理工作流的核心 Schema 和工具。 +""" + +from kb_workflow.pipeline.schemas import ( + StepInput, + StepOutput, + StepResult, + StepStatus, + StepRecord, + WorkflowState, + StepDefinition, +) +from kb_workflow.pipeline.state import WorkflowStateManager +from kb_workflow.pipeline.errors import ( + KBError, + NetworkError, + ValidationError, + SystemError, + AuthError, + ErrorType, +) +from kb_workflow.pipeline.config import KBSettings, get_settings +from kb_workflow.pipeline.retry import retry_on_network_error +from kb_workflow.pipeline.link_parser import LinkParser, ParsedLink +from kb_workflow.pipeline.idempotency import IdempotencyChecker +from kb_workflow.pipeline.platforms import PlatformAdapter, BilibiliAdapter, ZhihuAdapter + +__all__ = [ + # Schemas + "StepInput", + "StepOutput", + "StepResult", + "StepStatus", + "StepRecord", + "WorkflowState", + "StepDefinition", + # State + "WorkflowStateManager", + # Errors + "KBError", + "NetworkError", + "ValidationError", + "SystemError", + "AuthError", + "ErrorType", + # Config + "KBSettings", + "get_settings", + # Retry + "retry_on_network_error", + # Link Parser + "LinkParser", + "ParsedLink", + # Idempotency + "IdempotencyChecker", + # Platform Adapters + "PlatformAdapter", + "BilibiliAdapter", + "ZhihuAdapter", +] + +__version__ = "0.1.0" diff --git a/kb_workflow/pipeline/config.py b/kb_workflow/pipeline/config.py new file mode 100644 index 0000000..855a113 --- /dev/null +++ b/kb_workflow/pipeline/config.py @@ -0,0 +1,72 @@ +"""Configuration management for KB Pipeline + +Provides centralized configuration using pydantic-settings. +Configuration is loaded from ~/.kb_env file and environment variables. +""" + +from functools import lru_cache +from pathlib import Path +from typing import Optional + +from pydantic import Field, SecretStr +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class KBSettings(BaseSettings): + """KB 工作流配置 + + 使用 pydantic-settings 从环境变量和 ~/.kb_env 文件加载配置。 + + Attributes: + base_path: 知识库根目录（必需） + archive_path: 归档目录（必需） + git_root: Git 仓库根目录（可选） + git_user: Git 用户名（可选） + git_email: Git 邮箱（可选） + hf_token: HuggingFace Token（敏感信息） + zhihu_cookie: 知乎 Cookie（敏感信息） + """ + + model_config = SettingsConfigDict( + env_file=str(Path.home() / ".kb_env"), + env_file_encoding="utf-8", + env_prefix="KB_", + extra="ignore", + ) + + # 必需配置 + base_path: str = Field(..., description="知识库根目录") + archive_path: str = Field(..., description="归档目录") + intermediate_path: str = Field( + default_factory=lambda: str(Path.home() / "work" / "intermediate" / "analysis"), + description="中间文件存储目录" + ) + + # 可选配置 + git_root: str = Field(default="", description="Git 仓库根目录") + git_user: str = Field(default="", description="Git 用户名") + git_email: str = Field(default="", description="Git 邮箱") + + # 敏感信息（从环境变量读取） + hf_token: SecretStr = Field( + default=SecretStr(""), description="HuggingFace Token" + ) + zhihu_cookie: SecretStr = Field( + default=SecretStr(""), description="知乎 Cookie" + ) + bilibili_cookie: SecretStr = Field( + default=SecretStr(""), description="B站 Cookie (for video download)" + ) + + +@lru_cache +def get_settings() -> KBSettings: + """获取配置单例 + + 使用 lru_cache 确保配置只加载一次。 + 如需重新加载配置，调用 get_settings.cache_clear()。 + + Returns: + KBSettings 实例 + """ + return KBSettings() diff --git a/kb_workflow/pipeline/errors.py b/kb_workflow/pipeline/errors.py new file mode 100644 index 0000000..f337fb7 --- /dev/null +++ b/kb_workflow/pipeline/errors.py @@ -0,0 +1,174 @@ +"""Structured error handling for KB Pipeline + +Provides structured error types with context information for debugging. +All errors include step_name, input_summary, error_type, message, and suggestion. +""" + +from enum import Enum +from typing import Optional, List, Any + + +class ErrorType(str, Enum): + """错误类型分类""" + NETWORK = "network" + VALIDATION = "validation" + SYSTEM = "system" + AUTH = "auth" + + +class KBError(Exception): + """KB 工作流基础错误 + + 所有工作流错误的基类，包含丰富的上下文信息便于调试。 + + Attributes: + message: 错误信息 + error_type: 错误类型 + step_name: 发生错误的步骤名称 + input_summary: 输入数据摘要 + suggestion: 可操作的建议 + cause: 原始异常（可选） + """ + + def __init__( + self, + message: str, + error_type: ErrorType, + step_name: str, + input_summary: str, + suggestion: str, + cause: Optional[Exception] = None + ): + super().__init__(message) + self.message = message + self.error_type = error_type + self.step_name = step_name + self.input_summary = input_summary + self.suggestion = suggestion + self.cause = cause + + def to_dict(self) -> dict: + """将错误转换为字典格式用于日志和状态记录 + + Returns: + 包含所有错误上下文信息的字典 + """ + return { + "error_type": self.error_type.value, + "step_name": self.step_name, + "input_summary": self.input_summary, + "message": self.message, + "suggestion": self.suggestion + } + + +class NetworkError(KBError): + """网络错误 (403/404/500/超时) + + 用于表示网络请求失败，包括 HTTP 状态码错误和连接超时。 + + Attributes: + status_code: HTTP 状态码（可选） + """ + + def __init__( + self, + step_name: str, + input_summary: str, + status_code: Optional[int] = None, + cause: Optional[Exception] = None + ): + if status_code: + suggestion = f"检查网络连接，状态码: {status_code}" + else: + suggestion = "检查网络连接" + super().__init__( + message=f"网络请求失败: {status_code or 'unknown'}", + error_type=ErrorType.NETWORK, + step_name=step_name, + input_summary=input_summary, + suggestion=suggestion, + cause=cause + ) + self.status_code = status_code + + +class ValidationError(KBError): + """验证错误 (Schema 验证失败) + + 用于表示数据验证失败，包含详细的验证错误列表。 + + Attributes: + validation_errors: 验证错误列表 + """ + + def __init__( + self, + step_name: str, + input_summary: str, + validation_errors: Optional[List[str]] = None, + cause: Optional[Exception] = None + ): + super().__init__( + message="Schema 验证失败", + error_type=ErrorType.VALIDATION, + step_name=step_name, + input_summary=input_summary, + suggestion="检查输入数据格式是否符合要求", + cause=cause + ) + self.validation_errors = validation_errors or [] + + +class SystemError(KBError): + """系统错误 (文件系统/内存/依赖缺失) + + 用于表示系统级错误，如文件不存在、内存不足、依赖缺失等。 + + Attributes: + system_message: 系统错误消息 + """ + + def __init__( + self, + step_name: str, + input_summary: str, + system_message: Optional[str] = None, + cause: Optional[Exception] = None + ): + super().__init__( + message=f"系统错误: {system_message or 'unknown'}", + error_type=ErrorType.SYSTEM, + step_name=step_name, + input_summary=input_summary, + suggestion="检查系统资源和依赖是否满足要求", + cause=cause + ) + self.system_message = system_message + + +class AuthError(KBError): + """认证/配额错误 (API配额/Cookie过期/Token失效) + + 用于表示认证失败或 API 配额超限。 + + Attributes: + auth_type: 认证类型（如 API Token, Cookie 等） + """ + + def __init__( + self, + step_name: str, + input_summary: str, + auth_type: Optional[str] = None, + cause: Optional[Exception] = None + ): + super().__init__( + message=f"认证失败: {auth_type or 'unknown'}", + error_type=ErrorType.AUTH, + step_name=step_name, + input_summary=input_summary, + suggestion="检查 API Token 或 Cookie 是否有效", + cause=cause + ) + self.auth_type = auth_type diff --git a/kb_workflow/pipeline/idempotency.py b/kb_workflow/pipeline/idempotency.py new file mode 100644 index 0000000..d1ce2fe --- /dev/null +++ b/kb_workflow/pipeline/idempotency.py @@ -0,0 +1,80 @@ +"""Idempotency Checker module + +Provides file existence checking to avoid reprocessing content. +Checks for metadata JSON and content Markdown files. +""" + +from pathlib import Path +from typing import Tuple, List + + +class IdempotencyChecker: + """Checks if content has already been processed + + Determines if the output files for a given content already exist, + enabling idempotent processing to avoid duplicate work. + + File naming convention: + - Metadata: {platform}_{content_id}_metadata.json + - Content: {platform}_{content_id}_content.md + """ + + def check_exists( + self, + output_dir: Path, + content_id: str, + platform: str + ) -> Tuple[bool, List[Path]]: + """Check if output files already exist for this content + + Args: + output_dir: Directory where output files are stored + content_id: Unique content identifier (e.g., BV1GJ411x7Qw) + platform: Platform name (e.g., bilibili, zhihu) + + Returns: + Tuple of (all_exist, existing_files): + - all_exist: True if both metadata and content files exist + - existing_files: List of Path objects for existing files + """ + existing_files: List[Path] = [] + + if not content_id or not platform: + return (False, existing_files) + + # Define expected file paths + metadata_file = output_dir / f"{platform}_{content_id}_metadata.json" + content_file = output_dir / f"{platform}_{content_id}_content.md" + + # Check if files exist + if metadata_file.exists(): + existing_files.append(metadata_file) + + if content_file.exists(): + existing_files.append(content_file) + + # Return True only if both files exist + all_exist = len(existing_files) == 2 + + return (all_exist, existing_files) + + def get_expected_files( + self, + output_dir: Path, + content_id: str, + platform: str + ) -> Tuple[Path, Path]: + """Get the expected file paths for a content + + Args: + output_dir: Directory where output files are stored + content_id: Unique content identifier + platform: Platform name + + Returns: + Tuple of (metadata_path, content_path) + """ + metadata_path = output_dir / f"{platform}_{content_id}_metadata.json" + content_path = output_dir / f"{platform}_{content_id}_content.md" + + return (metadata_path, content_path) diff --git a/kb_workflow/pipeline/link_parser.py b/kb_workflow/pipeline/link_parser.py new file mode 100644 index 0000000..f8514f5 --- /dev/null +++ b/kb_workflow/pipeline/link_parser.py @@ -0,0 +1,181 @@ +"""Link Parser module + +Provides unified URL parsing for various content platforms. +Supports Bilibili (video, article) and Zhihu (answer, zhuanlan) URLs. +""" + +import re +from dataclasses import dataclass, field +from typing import Optional + +import requests + +from kb_workflow.pipeline.retry import retry_on_network_error + + +@dataclass +class ParsedLink: + """Represents a parsed link with platform and content information + + Attributes: + platform: Platform name (bilibili, zhihu) + content_type: Content type (video, article, answer, zhuanlan) + content_id: Unique content identifier + original_url: Original URL provided by user + resolved_url: Resolved URL after short link expansion (defaults to original_url) + """ + platform: str + content_type: str + content_id: str + original_url: str + resolved_url: str = field(default="") + + def __post_init__(self): + """Set resolved_url to original_url if not provided""" + if not self.resolved_url: + self.resolved_url = self.original_url + + +class LinkParser: + """Unified URL parser for multiple platforms + + Parses URLs from Bilibili and Zhihu, extracting platform, content type, + and content ID. Supports short link resolution for b23.tv URLs. + """ + + # Bilibili patterns + BILIBILI_VIDEO_PATTERN = re.compile( + r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)', + re.IGNORECASE + ) + BILIBILI_ARTICLE_PATTERN = re.compile( + r'https?://(?:www\.)?bilibili\.com/(?:read|article)/(cv[a-zA-Z0-9]+)', + re.IGNORECASE + ) + + # Zhihu patterns + ZHIHU_ANSWER_PATTERN = re.compile( + r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)', + re.IGNORECASE + ) + ZHIHU_ZHUANLAN_PATTERN = re.compile( + r'https?://zhuanlan\.zhihu\.com/p/(\d+)', + re.IGNORECASE + ) + + # Short link patterns + B23_TV_PATTERN = re.compile( + r'https?://(?:www\.)?b23\.tv/([a-zA-Z0-9]+)', + re.IGNORECASE + ) + + # Timeout for short link resolution + SHORT_LINK_TIMEOUT = 10 + + @retry_on_network_error + def _resolve_shortlink(self, url: str) -> Optional[str]: + """Resolve short link to its final URL + + Uses HTTP HEAD request with redirects to get the final URL. + + Args: + url: Short link URL to resolve + + Returns: + Final URL after redirects, or None if resolution failed + """ + try: + response = requests.head( + url, + allow_redirects=True, + timeout=self.SHORT_LINK_TIMEOUT + ) + if response.status_code == 200: + return response.url + return None + except (requests.RequestException, OSError): + return None + + def parse(self, url: str) -> Optional[ParsedLink]: + """Parse a URL and extract platform, content type, and ID + + Args: + url: URL to parse + + Returns: + ParsedLink with extracted information, or None if URL is not recognized + """ + if not url or not isinstance(url, str): + return None + + url = url.strip() + + # Try b23.tv short link first + if self.B23_TV_PATTERN.match(url): + resolved_url = self._resolve_shortlink(url) + if resolved_url: + # Recursively parse the resolved URL + result = self._parse_standard_url(resolved_url) + if result: + result.original_url = url + result.resolved_url = resolved_url + return result + return None + + return self._parse_standard_url(url) + + def _parse_standard_url(self, url: str) -> Optional[ParsedLink]: + """Parse a standard (non-short) URL + + Args: + url: URL to parse + + Returns: + ParsedLink with extracted information, or None if URL is not recognized + """ + # Try Bilibili video + match = self.BILIBILI_VIDEO_PATTERN.match(url) + if match: + return ParsedLink( + platform="bilibili", + content_type="video", + content_id=match.group(1), + original_url=url, + resolved_url=url + ) + + # Try Bilibili article + match = self.BILIBILI_ARTICLE_PATTERN.match(url) + if match: + return ParsedLink( + platform="bilibili", + content_type="article", + content_id=match.group(1), + original_url=url, + resolved_url=url + ) + + # Try Zhihu answer + match = self.ZHIHU_ANSWER_PATTERN.match(url) + if match: + return ParsedLink( + platform="zhihu", + content_type="answer", + content_id=match.group(1), + original_url=url, + resolved_url=url + ) + + # Try Zhihu zhuanlan + match = self.ZHIHU_ZHUANLAN_PATTERN.match(url) + if match: + return ParsedLink( + platform="zhihu", + content_type="zhuanlan", + content_id=match.group(1), + original_url=url, + resolved_url=url + ) + + # URL not recognized + return None diff --git a/kb_workflow/pipeline/pipeline.py b/kb_workflow/pipeline/pipeline.py new file mode 100644 index 0000000..60d57fc --- /dev/null +++ b/kb_workflow/pipeline/pipeline.py @@ -0,0 +1,399 @@ +"""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_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 + +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.settings = get_settings() + self.output_dir = output_dir or Path(self.settings.intermediate_path) + + # 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 + + 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_workflow.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): + # Ask user whether to re-run the completed step + should_rerun = self._ask_rerun_step(step_name, content_dir) + + if should_rerun: + logger.info(f"Re-running completed step: {step_name}") + # Clear the step's state record so it can be re-executed + if step_name in state.steps: + del state.steps[step_name] + state_manager.save(state) + else: + logger.info(f"Skipping completed step: {step_name}") + # Scan existing files and populate results for skipped steps + self._populate_skipped_step_results(step_name, results, content_dir, parsed) + continue + + # Skip ASR for text-based content types + text_content_types = {"article", "answer", "zhuanlan", "question"} + if step_name == "asr" and parsed.content_type in text_content_types: + logger.info(f"Skipping ASR for text content type: {parsed.content_type}") + 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_workflow.pipeline.schemas import ( + ParseInput, DownloadInput, ASRInput, AnalyzeInput, ArchiveInput + ) + + 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, + source_url=url, + ) + 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, + content_dir=str(content_dir), + ) + 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"), + ) + elif step_name == "archive": + # Archive needs analysis file from analyze step + analyze_result = results.get("analyze", {}) + analysis_file = analyze_result.get("analysis_file", "") + + step_input = ArchiveInput( + content_id=parsed.content_id, + platform=parsed.platform, + content_dir=str(content_dir), + analysis_file=analysis_file, + content_type=parsed.content_type, + ) + else: + # Generic step input + from kb_workflow.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", + "archive_file", "skipped"]: + 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, + ) + + def _ask_rerun_step(self, step_name: str, content_dir: Path) -> bool: + """Ask user whether to re-run a completed step. + + Args: + step_name: Name of the step to potentially re-run + content_dir: Content directory path + + Returns: + True if user wants to re-run, False to skip + """ + try: + from rich.console import Console + from rich.prompt import Confirm + + console = Console() + return Confirm.ask( + f"Step '{step_name}' is already completed. Re-run it?", + default=False + ) + except ImportError: + # Fallback to basic input if rich is not available + response = input(f"Step '{step_name}' is already completed. Re-run it? [y/N] ") + return response.lower() in ('y', 'yes') + + def _populate_skipped_step_results( + self, + step_name: str, + results: Dict, + content_dir: Path, + parsed: ParsedLink + ) -> None: + """Populate results for skipped steps by scanning existing files. + + This ensures downstream steps can find the outputs they need, + even when a step is skipped. + + Args: + step_name: Name of the skipped step + results: Results dictionary to populate + content_dir: Content directory path + parsed: Parsed link with platform and content_id + """ + if step_name == "asr": + # Scan for subtitle files generated by ASR + # Try both new naming convention (with _subtitle suffix) and old convention (without suffix) + base_name_new = f"{parsed.platform}_{parsed.content_id}_subtitle" + base_name_old = f"{parsed.platform}_{parsed.content_id}" + + json_path = content_dir / f"{base_name_new}.json" + if not json_path.exists(): + json_path = content_dir / f"{base_name_old}.json" + + srt_path = content_dir / f"{base_name_new}.srt" + if not srt_path.exists(): + srt_path = content_dir / f"{base_name_old}.srt" + + output_files = {} + if json_path.exists(): + output_files["json"] = str(json_path) + if srt_path.exists(): + output_files["srt"] = str(srt_path) + + if output_files: + results["asr"] = { + "success": True, + "error_message": None, + "output_files": output_files, + } + logger.info(f"Found existing ASR outputs: {output_files}") + else: + logger.warning(f"No existing ASR outputs found in {content_dir}") + results["asr"] = { + "success": False, + "error_message": "No existing subtitle files found", + "output_files": {}, + } + + elif step_name == "download": + # Scan for metadata file + metadata_path = content_dir / "metadata.json" + if metadata_path.exists(): + results["download"] = { + "success": True, + "error_message": None, + "metadata_file": str(metadata_path), + } + logger.info(f"Found existing download metadata: {metadata_path}") + + elif step_name == "analyze": + # Scan for analysis output + analysis_path = content_dir / "analysis.md" + if analysis_path.exists(): + results["analyze"] = { + "success": True, + "error_message": None, + "analysis_file": str(analysis_path), + } + logger.info(f"Found existing analysis: {analysis_path}") diff --git a/kb_workflow/pipeline/retry.py b/kb_workflow/pipeline/retry.py new file mode 100644 index 0000000..ecd5c67 --- /dev/null +++ b/kb_workflow/pipeline/retry.py @@ -0,0 +1,51 @@ +"""Retry mechanism for KB Pipeline + +Provides automatic retry for network errors using tenacity library. +""" + +import logging +from tenacity import ( + retry, + stop_after_attempt, + wait_exponential, + retry_if_exception_type, + before_sleep_log, +) + +from .errors import NetworkError + +logger = logging.getLogger(__name__) + +retry_on_network_error = retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=10), + retry=retry_if_exception_type(NetworkError), + before_sleep=before_sleep_log(logger, logging.WARNING), + reraise=True, +) +"""Decorator for automatic retry on network errors + +Automatically retries functions that raise NetworkError up to 3 times +with exponential backoff (2s, 4s, up to max 10s). + +Example: + @retry_on_network_error + def fetch_data(url): + response = requests.get(url) + if response.status_code >= 500: + raise NetworkError( + step_name="fetch", + input_summary=f"url={url}", + status_code=response.status_code + ) + return response.json() + +Args: + func: Function to wrap with retry logic + +Returns: + Wrapped function with retry logic + +Raises: + NetworkError: After 3 failed attempts +""" diff --git a/kb_workflow/pipeline/schemas.py b/kb_workflow/pipeline/schemas.py new file mode 100644 index 0000000..83a3e02 --- /dev/null +++ b/kb_workflow/pipeline/schemas.py @@ -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") diff --git a/kb_workflow/pipeline/state.py b/kb_workflow/pipeline/state.py new file mode 100644 index 0000000..984cf42 --- /dev/null +++ b/kb_workflow/pipeline/state.py @@ -0,0 +1,152 @@ +""" +WorkflowStateManager - 步骤级状态持久化管理器 + +提供工作流状态的原子写入、备份轮转和崩溃恢复功能。 + +Purpose: 为工作流提供可靠的状态管理，支持断点续传和崩溃恢复 +""" + +from __future__ import annotations + +import json +import os +import tempfile +from pathlib import Path +from typing import Optional + +from kb_workflow.pipeline.schemas import WorkflowState + + +class WorkflowStateManager: + """工作流状态管理器 + + 管理工作流状态的持久化，包括： + - 原子写入（使用临时文件 + os.replace） + - 备份轮转（workflow_state.json -> workflow_state.json.bak） + - 主文件损坏时从备份恢复 + + Attributes: + STATE_FILENAME: 状态文件名 + content_dir: 内容目录路径 + state_path: 状态文件完整路径 + backup_path: 备份文件完整路径 + """ + + STATE_FILENAME = "workflow_state.json" + + def __init__(self, content_dir: Path): + """初始化状态管理器 + + Args: + content_dir: 内容目录路径，状态文件将保存在此目录下 + """ + self.content_dir = Path(content_dir) + self.state_path = self.content_dir / self.STATE_FILENAME + self.backup_path = self.content_dir / f"{self.STATE_FILENAME}.bak" + + def load(self) -> Optional[WorkflowState]: + """加载工作流状态 + + 加载策略： + 1. 尝试加载主文件 (workflow_state.json) + 2. 如果主文件损坏或不存在，尝试加载备份文件 + 3. 如果都不存在，返回 None + + Returns: + WorkflowState 实例，如果文件不存在则返回 None + + Raises: + ValueError: 如果文件格式无效且备份也无法恢复 + """ + def _try_load(path: Path) -> Optional[WorkflowState]: + """尝试从指定路径加载状态""" + if not path.exists(): + return None + + try: + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + return WorkflowState.model_validate(data) + except (json.JSONDecodeError, ValueError): + return None + + # 先尝试主文件 + state = _try_load(self.state_path) + if state is not None: + return state + + # 主文件失败，尝试备份 + state = _try_load(self.backup_path) + return state + + def save(self, state: WorkflowState) -> None: + """原子保存工作流状态 + + 保存策略（原子写入 + 备份轮转）： + 1. 创建父目录（如果不存在） + 2. 如果状态文件存在，将其重命名为备份文件 + 3. 写入临时文件 + 4. 原子替换（os.replace） + + 原子写入确保： + - 读取者永远不会看到部分写入的文件 + - 崩溃时要么是旧文件，要么是新文件 + + Args: + state: 要保存的工作流状态 + + Raises: + OSError: 文件操作失败（权限、磁盘空间等） + """ + # 1. 创建父目录 + self.content_dir.mkdir(parents=True, exist_ok=True) + + # 2. 备份轮转：如果状态文件存在，移动到备份 + if self.state_path.exists(): + # 删除旧备份 + if self.backup_path.exists(): + self.backup_path.unlink() + # 将当前状态移动到备份 + self.state_path.rename(self.backup_path) + + # 3. 写入临时文件（在相同目录下，确保原子替换有效） + fd, temp_path = tempfile.mkstemp(dir=self.content_dir, suffix='.tmp') + try: + with os.fdopen(fd, 'w', encoding='utf-8') as f: + f.write(state.model_dump_json(indent=2)) + # 4. 原子替换 + os.replace(temp_path, self.state_path) + except Exception: + # 清理临时文件 + if os.path.exists(temp_path): + os.unlink(temp_path) + raise + + def get_or_create( + self, + content_id: str, + platform: str, + source_url: str + ) -> WorkflowState: + """获取现有状态或创建新状态 + + 幂等操作：如果状态已存在，返回现有状态；否则创建新状态。 + + Args: + content_id: 内容唯一标识 + platform: 平台名称 + source_url: 原始链接 + + Returns: + WorkflowState 实例 + """ + existing = self.load() + if existing is not None: + return existing + + # 创建新状态 + return WorkflowState( + content_id=content_id, + platform=platform, + source_url=source_url + )