diff --git a/kb_pipeline/__init__.py b/kb_pipeline/__init__.py deleted file mode 100644 index f2d46c1..0000000 --- a/kb_pipeline/__init__.py +++ /dev/null @@ -1,64 +0,0 @@ -"""KB Pipeline - 工作流管道基础设施 - -提供内容处理工作流的核心 Schema 和工具。 -""" - -from .schemas import ( - StepInput, - StepOutput, - StepResult, - StepStatus, - StepRecord, - WorkflowState, - StepDefinition, -) -from .state import WorkflowStateManager -from .errors import ( - KBError, - NetworkError, - ValidationError, - SystemError, - AuthError, - ErrorType, -) -from .config import KBSettings, get_settings -from .retry import retry_on_network_error -from .link_parser import LinkParser, ParsedLink -from .idempotency import IdempotencyChecker -from .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_pipeline/config.py b/kb_pipeline/config.py deleted file mode 100644 index 855a113..0000000 --- a/kb_pipeline/config.py +++ /dev/null @@ -1,72 +0,0 @@ -"""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_pipeline/errors.py b/kb_pipeline/errors.py deleted file mode 100644 index f337fb7..0000000 --- a/kb_pipeline/errors.py +++ /dev/null @@ -1,174 +0,0 @@ -"""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_pipeline/idempotency.py b/kb_pipeline/idempotency.py deleted file mode 100644 index d1ce2fe..0000000 --- a/kb_pipeline/idempotency.py +++ /dev/null @@ -1,80 +0,0 @@ -"""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_pipeline/link_parser.py b/kb_pipeline/link_parser.py deleted file mode 100644 index cd5f779..0000000 --- a/kb_pipeline/link_parser.py +++ /dev/null @@ -1,181 +0,0 @@ -"""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_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_pipeline/pipeline.py b/kb_pipeline/pipeline.py deleted file mode 100644 index bd979e4..0000000 --- a/kb_pipeline/pipeline.py +++ /dev/null @@ -1,399 +0,0 @@ -"""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_pipeline.config import get_settings -from kb_pipeline.link_parser import LinkParser, ParsedLink -from kb_pipeline.processors.base import BaseProcessor -from kb_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_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_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_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_pipeline/platforms/__init__.py b/kb_pipeline/platforms/__init__.py deleted file mode 100644 index baa74f2..0000000 --- a/kb_pipeline/platforms/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -"""Platform Adapters for KB Pipeline - -This module provides platform-specific adapters for content extraction -from various platforms (Bilibili, Zhihu, etc.). - -Each adapter implements the PlatformAdapter interface: -- get_content_type: Identify content type from URL -- fetch_metadata: Fetch content metadata -- fetch_content: Fetch content body in Markdown format -- get_output_filename: Generate output filename -""" - -from .base import PlatformAdapter -from .bilibili import BilibiliAdapter -from .zhihu import ZhihuAdapter - -__all__ = [ - "PlatformAdapter", - "BilibiliAdapter", - "ZhihuAdapter", -] diff --git a/kb_pipeline/platforms/base.py b/kb_pipeline/platforms/base.py deleted file mode 100644 index 05f6782..0000000 --- a/kb_pipeline/platforms/base.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Platform Adapter Base Class - -Provides an abstract base class for platform-specific content adapters. -Each platform (Bilibili, Zhihu, etc.) should implement a concrete adapter -that handles metadata extraction and content fetching. - -Purpose: Define a unified interface for multi-platform content extraction -""" - -from abc import ABC, abstractmethod -from typing import Dict, Any - - -class PlatformAdapter(ABC): - """Abstract base class for platform-specific content adapters - - A platform adapter encapsulates the logic for: - - Identifying content type from URL (video, article, answer, etc.) - - Fetching metadata for a given content ID - - Fetching content body in Markdown format - - Generating output filenames - - Subclasses must implement all abstract methods to provide - platform-specific functionality. - - Example: - class BilibiliAdapter(PlatformAdapter): - def get_content_type(self, url: str) -> str: - if '/video/' in url: - return 'video' - elif '/read/' in url: - return 'article' - return 'unknown' - - async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: - # Platform-specific implementation - ... - """ - - @abstractmethod - def get_content_type(self, url: str) -> str: - """Identify content type from URL - - Args: - url: The content URL to analyze - - Returns: - Content type string (e.g., "video", "article", "answer", "zhuanlan") - - Example: - adapter.get_content_type("https://www.bilibili.com/video/BV1xx...") - -> "video" - """ - pass - - @abstractmethod - async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: - """Fetch metadata for the given content - - Args: - content_id: Platform-specific content identifier (e.g., BV号, CV号) - content_type: Type of content ("video", "article", etc.) - - Returns: - Dictionary containing metadata fields. Structure varies by platform - and content type, but typically includes: - - title: Content title - - author: Author name - - url: Original content URL - - Platform-specific fields - - Raises: - NetworkError: If network request fails - AuthError: If authentication fails or quota exceeded - """ - pass - - @abstractmethod - async def fetch_content(self, content_id: str, content_type: str) -> str: - """Fetch content body in Markdown format - - Args: - content_id: Platform-specific content identifier - content_type: Type of content - - Returns: - Markdown-formatted content string with YAML frontmatter: - --- - title: Article Title - author: Author Name - source_url: https://... - platform: bilibili - content_type: article - fetched_at: 2024-01-15T10:30:00 - --- - - # Article Content... - - Note: - For video content, this may return an empty string or - placeholder text, as video content extraction is handled - separately (e.g., ASR transcription). - """ - pass - - @abstractmethod - def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: - """Generate output filename for content - - Args: - content_id: Platform-specific content identifier - content_type: Type of content - suffix: Filename suffix (e.g., "metadata.json", "content.md") - - Returns: - Filename string without directory path - - Example: - adapter.get_output_filename("BV1xx...", "video", "metadata.json") - -> "bilibili_BV1xx..._metadata.json" - """ - pass diff --git a/kb_pipeline/platforms/bilibili.py b/kb_pipeline/platforms/bilibili.py deleted file mode 100644 index b698000..0000000 --- a/kb_pipeline/platforms/bilibili.py +++ /dev/null @@ -1,485 +0,0 @@ -"""Bilibili Platform Adapter - -Implements the PlatformAdapter interface for Bilibili platform content. -Supports video and article metadata extraction and content fetching. - -Key features: -- Video metadata extraction (BV号) -- Article metadata extraction (CV号) -- Video download using yt-dlp (for ASR/Whisper) -- Automatic retry on network errors (3 attempts with exponential backoff) -- State recording for failed steps -- Fallback from API to Playwright for 403 errors (handled by BilibiliExtractor) -""" - -from __future__ import annotations - -import asyncio -import logging -import os -import re -import tempfile -from pathlib import Path -from typing import Dict, Any, Optional - -from .base import PlatformAdapter -from ..errors import NetworkError, AuthError -from ..retry import retry_on_network_error -from ..state import WorkflowStateManager -from kb_workflow.metadata.bilibili_extractor import BilibiliExtractor - -logger = logging.getLogger(__name__) - - -class BilibiliAdapter(PlatformAdapter): - """Bilibili platform adapter for video and article content - - Implements the PlatformAdapter interface to provide: - - Content type identification from URL - - Video metadata extraction - - Article metadata extraction - - Article content fetching in Markdown format - - Automatic error retry and state recording - - Attributes: - PLATFORM: Platform identifier string ("bilibili") - _extractor: BilibiliExtractor instance for actual extraction - _state_manager: Optional WorkflowStateManager for recording step status - - Example: - adapter = BilibiliAdapter() - content_type = adapter.get_content_type("https://www.bilibili.com/video/BV1xx...") - # -> "video" - - metadata = await adapter.fetch_metadata("BV1xx...", "video") - # -> {"metadata": {...}, "stats": {...}, ...} - """ - - PLATFORM = "bilibili" - - # URL patterns for content type identification - VIDEO_PATTERN = re.compile(r'bilibili\.com/video/(BV[a-zA-Z0-9]+)') - ARTICLE_PATTERN = re.compile(r'bilibili\.com/read/(cv\d+)') - - def __init__( - self, - headless: bool = True, - timeout: int = 30, - state_manager: Optional[WorkflowStateManager] = None - ): - """Initialize BilibiliAdapter - - Args: - headless: Whether to use headless mode for Playwright (default True) - timeout: Request timeout in seconds (default 30) - state_manager: Optional WorkflowStateManager for recording step status - """ - self._extractor = BilibiliExtractor(headless=headless, timeout=timeout) - self._state_manager = state_manager - - def get_content_type(self, url: str) -> str: - """Identify content type from URL - - Args: - url: Bilibili URL to analyze - - Returns: - Content type string: "video", "article", or "unknown" - - Example: - adapter.get_content_type("https://www.bilibili.com/video/BV1xx...") - -> "video" - """ - if self.VIDEO_PATTERN.search(url): - return "video" - if self.ARTICLE_PATTERN.search(url): - return "article" - return "unknown" - - async def fetch_metadata( - self, - content_id: str, - content_type: str - ) -> Dict[str, Any]: - """Fetch metadata for the given content - - Args: - content_id: BV号 for videos, CV号 for articles - content_type: "video" or "article" - - Returns: - Dictionary containing metadata. Structure varies by content type: - - Video: {metadata: {...}, stats: {...}, description: str} - - Article: {cvid, title, author, publish_time, ...} - - Raises: - NetworkError: If network request fails after 3 retries - AuthError: If authentication fails (403/quota exceeded) - ValueError: If content_type is not supported - """ - try: - return await self._fetch_metadata_with_retry(content_id, content_type) - except (NetworkError, AuthError) as e: - # Record failure to state manager only after all retries exhausted - self._record_step_failed("metadata", str(e)) - raise - except Exception as e: - # Check if this is an API error that should be treated as network error - if self._is_network_related_error(e): - network_error = NetworkError( - step_name="metadata", - input_summary=f"content_id={content_id}, type={content_type}", - cause=e - ) - self._record_step_failed("metadata", str(network_error)) - raise network_error - raise - - @retry_on_network_error - async def _fetch_metadata_with_retry( - self, - content_id: str, - content_type: str - ) -> Dict[str, Any]: - """Internal method with retry logic for metadata fetching - - Args: - content_id: BV号 for videos, CV号 for articles - content_type: "video" or "article" - - Returns: - Dictionary containing metadata - - Raises: - NetworkError: If network request fails - AuthError: If authentication fails - ValueError: If content_type is not supported - """ - if content_type == "video": - # Call async method directly to avoid nested asyncio.run() - return await self._extractor._extract_video_async(content_id) - elif content_type == "article": - # Call async method directly to avoid nested asyncio.run() - return await self._extractor._extract_article_async(content_id) - else: - raise ValueError(f"Unsupported content type: {content_type}") - - def _is_network_related_error(self, error: Exception) -> bool: - """Check if an error is network-related and should trigger retry - - Args: - error: Exception to check - - Returns: - True if error is network-related - """ - error_str = str(error).lower() - network_indicators = [ - 'timeout', 'connection', 'network', '503', '502', '500', - 'connection refused', 'connection reset', 'eof', - ] - return any(indicator in error_str for indicator in network_indicators) - - async def fetch_content( - self, - content_id: str, - content_type: str - ) -> str: - """Fetch content body in Markdown format - - For videos, this returns an empty string as video content - extraction is handled separately (ASR transcription). - - For articles, this fetches the article content and formats - it with YAML frontmatter. - - Args: - content_id: CV号 for articles - content_type: "video" or "article" - - Returns: - Markdown-formatted content string with YAML frontmatter, - or empty string for videos - """ - if content_type == "video": - # Video content is handled by ASR in Phase 3 - return "" - - if content_type == "article": - # Fetch article metadata first - metadata = await self.fetch_metadata(content_id, "article") - - # Fetch article content (requires additional implementation) - content = await self._fetch_article_content(content_id) - - # Format with YAML frontmatter - return self._format_content_markdown(metadata, content) - - raise ValueError(f"Unsupported content type: {content_type}") - - async def _fetch_article_content(self, cvid: str) -> str: - """Fetch article body content - - Args: - cvid: Article CV号 - - Returns: - Raw article content string - - Note: - This method currently returns a placeholder. - Full implementation would use additional API calls - or Playwright to fetch the article body. - """ - # TODO: Implement full article content fetching - # This would require additional API endpoints or Playwright - logger.warning(f"Article content fetching not fully implemented for {cvid}") - return "" - - def _format_content_markdown( - self, - metadata: Dict[str, Any], - content: str, - content_type: str = "article" - ) -> str: - """Format content with YAML frontmatter - - Args: - metadata: Article metadata dictionary - content: Article body content - content_type: Content type string - - Returns: - Markdown string with YAML frontmatter - """ - import yaml - from datetime import datetime - - frontmatter = { - "title": metadata.get("title", ""), - "author": metadata.get("author", ""), - "source_url": metadata.get("url", ""), - "platform": self.PLATFORM, - "content_type": content_type, - "fetched_at": datetime.now().isoformat(), - } - - # Add additional metadata if available - if "cvid" in metadata: - frontmatter["cvid"] = metadata["cvid"] - if "mid" in metadata: - frontmatter["author_id"] = metadata["mid"] - if "word_count" in metadata: - frontmatter["word_count"] = metadata["word_count"] - - yaml_str = yaml.dump(frontmatter, allow_unicode=True, sort_keys=False) - return f"---\n{yaml_str}---\n\n{content}" - - def get_output_filename( - self, - content_id: str, - content_type: str, - suffix: str - ) -> str: - """Generate output filename for content - - Args: - content_id: BV号 or CV号 - content_type: "video" or "article" - suffix: Filename suffix (e.g., "metadata.json", "content.md") - - Returns: - Filename string in format "bilibili_{content_id}_{suffix}" - - Example: - adapter.get_output_filename("BV1xx...", "video", "metadata.json") - -> "bilibili_BV1xx..._metadata.json" - """ - return f"{self.PLATFORM}_{content_id}_{suffix}" - - def _record_step_failed(self, step_name: str, error_message: str) -> None: - """Record step failure to state manager - - Args: - step_name: Name of the failed step - error_message: Error message to record - """ - if self._state_manager is not None: - self._state_manager.mark_step_failed(step_name, error_message) - logger.warning(f"Step '{step_name}' failed: {error_message}") - - async def download_video( - self, - bvid: str, - output_dir: Path, - quality: int = 32 - ) -> Dict[str, Any]: - """Download Bilibili video using yt-dlp. - - Downloads only audio (sufficient for Whisper ASR) to save bandwidth. - - Args: - bvid: Bilibili video ID (BV号) - output_dir: Directory to save the video - quality: Video quality (16=360P, 32=480P, 64=720P, 80=1080P) - - Returns: - Dictionary with: - - success: bool - - filepath: str or None - - message: str - """ - result = {"success": False, "filepath": None, "message": ""} - - try: - from ..config import get_settings - - url = f"https://www.bilibili.com/video/{bvid}" - - # Get cookie from settings or environment - settings = get_settings() - cookie = settings.bilibili_cookie.get_secret_value() or os.environ.get('BILIBILI_COOKIE', '') - - # Create cookie file if needed - cookie_file = None - if cookie: - cookie_file = self._create_cookie_file(cookie) - - try: - # Get video title for filename - title = await self._get_video_title(url, cookie_file) - safe_title = self._sanitize_filename(title, bvid) - - # Build yt-dlp command - download best audio only for ASR - # This saves bandwidth since Whisper only needs audio - output_template = str(output_dir / f"{safe_title}.%(ext)s") - - # Use best audio format for ASR - cmd = [ - "yt-dlp", - "-f", "bestaudio/best", - "-o", output_template, - "--no-download-archive", - "-q", - url - ] - - if cookie_file: - cmd.extend(["--cookies", cookie_file]) - - logger.info(f"Downloading video: {bvid}") - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - stdout, stderr = await proc.communicate() - - # Find downloaded file - downloaded_files = list(output_dir.glob(f"{safe_title}.*")) - video_extensions = {'.mp4', '.mkv', '.webm', '.mp3', '.m4a', '.opus', '.wav'} - downloaded_files = [f for f in downloaded_files if f.suffix.lower() in video_extensions] - - if downloaded_files: - video_file = downloaded_files[0] - result["success"] = True - result["filepath"] = str(video_file) - result["message"] = f"Downloaded: {video_file.name}" - logger.info(f"Video downloaded: {video_file}") - else: - # Try to find any new media file in output_dir - all_media = list(output_dir.glob("*.*")) - all_media = [f for f in all_media if f.suffix.lower() in video_extensions] - if all_media: - # Get the most recently modified file - video_file = max(all_media, key=lambda f: f.stat().st_mtime) - result["success"] = True - result["filepath"] = str(video_file) - result["message"] = f"Downloaded: {video_file.name}" - logger.info(f"Video downloaded (fallback): {video_file}") - else: - result["message"] = "No downloaded file found" - logger.warning(f"No downloaded file found for {bvid}") - - finally: - # Clean up cookie file - if cookie_file and os.path.exists(cookie_file): - os.remove(cookie_file) - - except Exception as e: - result["message"] = f"Download error: {str(e)}" - logger.error(f"Video download failed for {bvid}: {e}") - - return result - - async def _get_video_title(self, url: str, cookie_file: Optional[str] = None) -> str: - """Get video title using yt-dlp. - - Args: - url: Bilibili video URL - cookie_file: Optional path to cookie file - - Returns: - Video title or fallback name - """ - try: - cmd = ["yt-dlp", "--get-title", url] - if cookie_file: - cmd.extend(["--cookies", cookie_file]) - - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - stdout, stderr = await proc.communicate() - - title = stdout.decode('utf-8').strip() - if title and 'error' not in title.lower(): - return title - except Exception as e: - logger.debug(f"Failed to get video title: {e}") - - return "video" - - def _sanitize_filename(self, title: str, bvid: str) -> str: - """Generate standardized filename for video download. - - Args: - title: Video title (unused, kept for API compatibility) - bvid: Video BV号 - - Returns: - Standardized filename in format "bilibili_{bvid}" - """ - # Use standardized naming convention: platform_bvid - return f"bilibili_{bvid}" - - def _create_cookie_file(self, cookie: str) -> str: - """Create Netscape format cookie file for yt-dlp. - - Args: - cookie: Cookie string (e.g., "SESSDATA=xxx; bili_jct=xxx") - - Returns: - Path to temporary cookie file - """ - # Parse cookie string - cookies = {} - for item in cookie.split(';'): - item = item.strip() - if '=' in item: - key, value = item.split('=', 1) - cookies[key.strip()] = value.strip() - - # Create Netscape format - cookie_content = "# Netscape HTTP Cookie File\n" - for name, value in cookies.items(): - cookie_content += f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n" - - # Write to temp file - fd, path = tempfile.mkstemp(suffix='.txt') - with os.fdopen(fd, 'w') as f: - f.write(cookie_content) - - return path diff --git a/kb_pipeline/platforms/zhihu.py b/kb_pipeline/platforms/zhihu.py deleted file mode 100644 index 609791f..0000000 --- a/kb_pipeline/platforms/zhihu.py +++ /dev/null @@ -1,263 +0,0 @@ -"""Zhihu Platform Adapter - -Provides content extraction for Zhihu platform: -- Answer extraction from question/answer URLs -- Zhuanlan (article) extraction from zhuanlan URLs -- Cookie-based authentication -- Login page detection for expired cookies -""" - -from typing import Dict, Any, Optional -from datetime import datetime -import re -import asyncio - -from kb_pipeline.platforms.base import PlatformAdapter -from kb_pipeline.errors import AuthError -from kb_pipeline.config import get_settings -from kb_workflow.metadata.zhihu_extractor import ZhihuPlaywrightExtractor - - -class ZhihuAdapter(PlatformAdapter): - """Zhihu platform adapter for answer and zhuanlan content extraction. - - Uses ZhihuPlaywrightExtractor (from kb-zhihu-metadata) for content extraction. - Requires KB_ZHIHU_COOKIE environment variable for authentication. - - Example: - adapter = ZhihuAdapter() - content_type = adapter.get_content_type("https://www.zhihu.com/question/123/answer/456") - # -> "answer" - - metadata = await adapter.fetch_metadata("456", "answer") - content = await adapter.fetch_content("456", "answer") - """ - - PLATFORM = "zhihu" - - # URL patterns for content type detection - ANSWER_PATTERN = re.compile(r'zhihu\.com/question/\d+/answer/(\d+)') - ZHUANLAN_PATTERN = re.compile(r'zhuanlan\.zhihu\.com/p/([\w\d]+)') - QUESTION_PATTERN = re.compile(r'zhihu\.com/question/(\d+)(?!/answer)') - - def __init__(self, headless: bool = True, timeout: int = 30): - """Initialize ZhihuAdapter. - - Args: - headless: Run browser in headless mode (default: True) - timeout: Page load timeout in seconds (default: 30) - """ - self.headless = headless - self.timeout = timeout - self._cookie: Optional[str] = None - self._extractor = None - - @property - def cookie(self) -> str: - """Get Zhihu cookie from settings. - - Returns: - Cookie string from KB_ZHIHU_COOKIE environment variable. - Returns empty string if not configured. - """ - if self._cookie is None: - settings = get_settings() - self._cookie = settings.zhihu_cookie.get_secret_value() - return self._cookie - - def get_content_type(self, url: str) -> str: - """Identify content type from URL. - - Args: - url: The content URL to analyze - - Returns: - Content type: 'answer', 'zhuanlan', or 'unknown' - """ - if self.ANSWER_PATTERN.search(url): - return "answer" - elif self.ZHUANLAN_PATTERN.search(url): - return "zhuanlan" - return "unknown" - - def _get_extractor(self): - """Get or create ZhihuPlaywrightExtractor instance. - - Returns: - ZhihuPlaywrightExtractor instance - """ - if self._extractor is None: - self._extractor = ZhihuPlaywrightExtractor( - headless=self.headless, - timeout=self.timeout, - cookie=self.cookie - ) - return self._extractor - - def _extract_with_playwright(self, url: str) -> Optional[Dict[str, Any]]: - """Extract content using Playwright-based extractor. - - Args: - url: Full URL to extract content from - - Returns: - Extraction result dictionary or None if extraction failed - """ - extractor = self._get_extractor() - return extractor.extract(url) - - def _detect_login_page(self, result: Dict[str, Any]) -> bool: - """Detect if the result indicates a login page redirect. - - This happens when the cookie has expired or is invalid. - - Args: - result: Extraction result dictionary - - Returns: - True if login page is detected, False otherwise - """ - if not result: - return False - - title = result.get("title", "") - content = result.get("content", "") - content_markdown = result.get("content_markdown", "") - - # Check for login-related indicators - login_indicators = ["登录", "登陆", "sign in", "login"] - title_lower = title.lower() - - # If title contains login keywords - if any(indicator in title_lower for indicator in login_indicators): - return True - - # If content is suspiciously short (login page has minimal content) - total_content = content + content_markdown - if len(total_content.strip()) < 100: - return True - - return False - - async def fetch_metadata(self, content_id: str, content_type: str, source_url: str = None) -> Dict[str, Any]: - """Fetch metadata for the content. - - Args: - content_id: Answer ID or article ID - content_type: 'answer' or 'zhuanlan' - source_url: Original source URL (optional, used for answers to get question ID) - - Returns: - Dictionary containing metadata (title, author, publish_time, etc.) - - Raises: - AuthError: If cookie is missing or expired - """ - # Check for missing cookie - if not self.cookie: - raise AuthError( - step_name="zhihu_metadata", - input_summary=f"content_id={content_id}, type={content_type}", - auth_type="zhihu_cookie", - ) - - # Build URL based on content type - if content_type == "answer": - # For answers, try to use source_url first, then fallback to constructed URL - if source_url: - url = source_url - else: - # Fallback: use answer ID only (may not work for all cases) - url = f"https://www.zhihu.com/question/0/answer/{content_id}" - elif content_type == "zhuanlan": - url = f"https://zhuanlan.zhihu.com/p/{content_id}" - else: - raise ValueError(f"Unsupported content type: {content_type}") - - # Extract content in a separate thread to avoid blocking asyncio - result = await asyncio.to_thread(self._extract_with_playwright, url) - - # Check for login page - if self._detect_login_page(result): - raise AuthError( - step_name="zhihu_metadata", - input_summary=f"content_id={content_id}, type={content_type}", - auth_type="zhihu_cookie", - cause=Exception("Cookie may have expired, please update KB_ZHIHU_COOKIE") - ) - - if not result: - raise AuthError( - step_name="zhihu_metadata", - input_summary=f"content_id={content_id}, type={content_type}", - auth_type="zhihu_cookie", - cause=Exception("Failed to extract content, check if content exists") - ) - - return result - - async def fetch_content(self, content_id: str, content_type: str, source_url: str = None) -> str: - """Fetch content in Markdown format. - - Args: - content_id: Answer ID or article ID - content_type: 'answer' or 'zhuanlan' - source_url: Original source URL (optional, used for answers) - - Returns: - Markdown-formatted content with YAML frontmatter - - Raises: - AuthError: If cookie is missing or expired - """ - # Get metadata (which also validates auth) - metadata = await self.fetch_metadata(content_id, content_type, source_url=source_url) - - # Get content markdown - content_markdown = metadata.get("content_markdown", "") or metadata.get("content", "") - - # Format with YAML frontmatter - return self._format_with_frontmatter(metadata, content_markdown, content_type) - - def _format_with_frontmatter( - self, - metadata: Dict[str, Any], - content: str, - content_type: str - ) -> str: - """Format content with YAML frontmatter. - - Args: - metadata: Content metadata dictionary - content: Markdown content - content_type: 'answer' or 'zhuanlan' - - Returns: - Formatted string with YAML frontmatter and content - """ - frontmatter_lines = [ - "---", - f"title: {metadata.get('title', '')}", - f"author: {metadata.get('author', '')}", - f"source_url: {metadata.get('url', '')}", - f"platform: {self.PLATFORM}", - f"content_type: {content_type}", - f"fetched_at: {datetime.now().isoformat()}", - "---", - "", - ] - - return "\n".join(frontmatter_lines) + content - - def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: - """Generate output filename for the content. - - Args: - content_id: Answer ID or article ID - content_type: 'answer' or 'zhuanlan' - suffix: File suffix (e.g., 'metadata.json', 'content.md') - - Returns: - Filename string (e.g., 'zhihu_789012_metadata.json') - """ - return f"zhihu_{content_id}_{suffix}" diff --git a/kb_pipeline/processors/__init__.py b/kb_pipeline/processors/__init__.py deleted file mode 100644 index e74f168..0000000 --- a/kb_pipeline/processors/__init__.py +++ /dev/null @@ -1,26 +0,0 @@ -"""Processors module for KB Pipeline - -This module contains processors for various content transformation tasks: -- BaseProcessor: Abstract base class for all processors -- ParseProcessor: URL parsing for various platforms -- DownloadProcessor: Metadata and content downloading -- ASRProcessor: Audio-to-text conversion (subtitle download + Whisper.cpp) -- AnalyzeProcessor: AI-powered content analysis using DeepAnalyzer -- ArchiveProcessor: Copy analysis reports to archive directory -""" - -from .base import BaseProcessor -from .parse import ParseProcessor -from .download import DownloadProcessor -from .asr import ASRProcessor -from .analyze import AnalyzeProcessor -from .archive import ArchiveProcessor - -__all__ = [ - "BaseProcessor", - "ParseProcessor", - "DownloadProcessor", - "ASRProcessor", - "AnalyzeProcessor", - "ArchiveProcessor", -] diff --git a/kb_pipeline/processors/analyze.py b/kb_pipeline/processors/analyze.py deleted file mode 100644 index 0cfd1d8..0000000 --- a/kb_pipeline/processors/analyze.py +++ /dev/null @@ -1,325 +0,0 @@ -"""AnalyzeProcessor - AI-powered content analysis - -Unified processor for generating AI analysis reports from content. -Wraps existing DeepAnalyzer for integration into the pipeline. -""" - -from __future__ import annotations - -import json -import logging -import os -import sys -from pathlib import Path -from typing import Optional, Dict, Any - -# Add kb-deep-analyzer to path for import -_kb_deep_analyzer_path = Path(__file__).parent.parent.parent / "kb-deep-analyzer" -if str(_kb_deep_analyzer_path) not in sys.path: - sys.path.insert(0, str(_kb_deep_analyzer_path)) - -from ..config import get_settings -from ..schemas import AnalyzeInput, AnalyzeOutput -from ..state import WorkflowStateManager -from .base import BaseProcessor - -logger = logging.getLogger(__name__) - - -class AnalyzeProcessor(BaseProcessor): - """Analyze 处理器 - AI 深度内容分析 - - 处理流程： - 1. 检查步骤是否已完成（幂等执行） - 2. 加载元数据和内容文件 - 3. 初始化 AI 客户端 - 4. 调用 DeepAnalyzer 生成分析报告 - 5. 生成 frontmatter 并保存 Markdown 文件 - 6. 记录处理状态到 workflow_state.json - - Attributes: - STEP_NAME: 步骤名称，用于状态记录 - settings: KB 配置 - state_manager: 状态管理器（可选） - """ - - STEP_NAME = "analyze" - - async def process(self, input_data: AnalyzeInput) -> AnalyzeOutput: - """执行分析处理 - - 策略： - 1. 检查步骤是否已完成（跳过已完成） - 2. 加载元数据和内容 - 3. 调用 DeepAnalyzer - 4. 保存分析报告 - 5. 记录状态 - - Args: - input_data: Analyze 输入数据 - - Returns: - Analyze 处理结果 - - Raises: - Exception: 处理失败时抛出异常 - """ - try: - # Step 1: 检查是否已完成（幂等执行） - if self.state_manager: - state = self.state_manager.load() - if state and state.is_step_completed(self.STEP_NAME): - logger.info(f"Step {self.STEP_NAME} already completed, skipping") - # 返回已有结果 - existing_file = Path(input_data.output_file) - if existing_file.exists(): - return AnalyzeOutput( - success=True, - analysis_file=str(existing_file), - summary="Skipped - already completed", - main_topics=[], - key_insights=[], - tags=[], - stats={"skipped": True} - ) - - # Step 2: 加载元数据和内容 - metadata = self._load_json(input_data.metadata_file) - content = self._load_json(input_data.content_file) - - # 处理 Whisper 字幕格式 - 提取完整文本 - if content and 'transcription' in content: - # Whisper JSON 格式，提取文本 - text_parts = [] - for item in content.get('transcription', []): - text_parts.append(item.get('text', '')) - content = {'text': ' '.join(text_parts)} - logger.info(f"Extracted {len(content['text'])} chars from Whisper transcription") - - # 对于文章类型，内容可能是纯文本文件 - if not content and input_data.content_type in ("article", "answer", "zhuanlan"): - content_path = Path(input_data.content_file) - if content_path.exists(): - text = content_path.read_text(encoding="utf-8") - # 去除 YAML frontmatter（如果存在） - text = self._strip_frontmatter(text) - content = {"text": text} - - # Step 3: 调用 DeepAnalyzer - result = await self._run_analysis( - metadata=metadata, - content=content, - content_type=input_data.content_type - ) - - # Step 4: 生成并保存分析报告 - output_path = Path(input_data.output_file) - output_path.parent.mkdir(parents=True, exist_ok=True) - - self._write_analysis_report( - output_path=output_path, - metadata=metadata, - result=result, - content_type=input_data.content_type - ) - - # Step 5: 记录成功状态 - if self.state_manager: - state = self.state_manager.load() - if state: - summary = f"topics={len(result.main_topics)}, insights={len(result.key_insights)}" - state.mark_step_completed(self.STEP_NAME, summary) - self.state_manager.save(state) - - return AnalyzeOutput( - success=True, - analysis_file=str(output_path), - summary=result.summary, - main_topics=result.main_topics, - key_insights=result.key_insights, - tags=result.recommended_tags, - stats={ - "topics_count": len(result.main_topics), - "insights_count": len(result.key_insights), - "tags_count": len(result.recommended_tags) - } - ) - - except Exception as e: - logger.error(f"Analyze failed: {e}") - - # 记录失败状态 - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_failed(self.STEP_NAME, str(e)) - self.state_manager.save(state) - - return AnalyzeOutput( - success=False, - error_message=str(e), - analysis_file="", - summary="", - main_topics=[], - key_insights=[], - tags=[], - stats={} - ) - - def _load_json(self, file_path: str) -> Dict[str, Any]: - """加载 JSON 文件 - - Args: - file_path: JSON 文件路径 - - Returns: - 解析后的字典，如果文件不存在或解析失败返回空字典 - """ - if not file_path: - return {} - - path = Path(file_path) - if not path.exists(): - logger.warning(f"File not found: {file_path}") - return {} - - try: - with open(path, 'r', encoding='utf-8') as f: - return json.load(f) - except json.JSONDecodeError as e: - logger.error(f"Failed to parse JSON: {file_path}, error: {e}") - return {} - - async def _run_analysis(self, metadata: Dict, content: Dict, content_type: str): - """调用 DeepAnalyzer 执行分析 - - Args: - metadata: 元数据字典 - content: 内容字典 - content_type: 内容类型 - - Returns: - AnalysisResult 对象 - """ - # Lazy import to avoid dependency issues - from deep_analyzer import DeepAnalyzer, SimpleAIClient - - # 初始化 AI 客户端 - ai_client = self._create_ai_client() - - # 创建分析器实例 - analyzer = DeepAnalyzer(ai_client=ai_client) - - # 执行分析 - return analyzer.analyze(metadata, content, content_type) - - def _create_ai_client(self): - """创建 AI 客户端 - - Returns: - SimpleAIClient 实例 - """ - from deep_analyzer import SimpleAIClient - - # 从环境变量获取配置 - api_key = os.environ.get("AI_API_KEY", "") - provider = os.environ.get("AI_PROVIDER", "") - model = os.environ.get("AI_MODEL", "glm-5") - base_url = os.environ.get("AI_BASE_URL", "") - - return SimpleAIClient( - api_key=api_key, - provider=provider, - model=model, - base_url=base_url - ) - - def _normalize_metadata(self, metadata: Dict) -> Dict: - """标准化元数据结构（扁平化嵌套结构） - - Args: - metadata: 原始元数据字典，可能是嵌套结构 - - Returns: - 扁平化的元数据字典 - """ - if not metadata: - return {} - - # 如果已经是扁平结构（顶层有 title） - if 'title' in metadata: - return metadata - - # 处理嵌套结构 {metadata: {...}, stats: {...}} - if 'metadata' in metadata: - return metadata['metadata'].copy() - - return metadata - - def _strip_frontmatter(self, text: str) -> str: - """去除 Markdown 文件中的 YAML frontmatter - - Args: - text: 可能包含 frontmatter 的文本 - - Returns: - 去除 frontmatter 后的文本 - """ - if not text: - return text - - # 检查是否以 --- 开头 - if not text.startswith('---'): - return text - - # 找到第二个 --- - lines = text.split('\n') - end_index = -1 - for i in range(1, len(lines)): - if lines[i].strip() == '---': - end_index = i - break - - if end_index == -1: - return text - - # 返回 frontmatter 之后的内容（跳过空行） - remaining = '\n'.join(lines[end_index + 1:]) - return remaining.lstrip('\n') - - def _write_analysis_report( - self, - output_path: Path, - metadata: Dict, - result, - content_type: str - ) -> None: - """写入分析报告 Markdown 文件 - - Args: - output_path: 输出文件路径 - metadata: 元数据字典 - result: AnalysisResult 对象 - content_type: 内容类型 - """ - from deep_analyzer import generate_frontmatter - - # 标准化元数据（扁平化嵌套结构） - normalized_metadata = self._normalize_metadata(metadata) - - # 生成 frontmatter - frontmatter = generate_frontmatter(normalized_metadata, result, content_type) - - # 获取完整分析内容 - full_analysis = result.full_analysis - - # 文章/回答类型：追加原文 - if content_type in ("article", "answer", "zhuanlan") and result.original_text: - full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text - - # 写入文件 - with open(output_path, 'w', encoding='utf-8') as f: - f.write(frontmatter + "\n\n") - f.write(full_analysis) - - logger.info(f"Analysis report saved to: {output_path}") diff --git a/kb_pipeline/processors/archive.py b/kb_pipeline/processors/archive.py deleted file mode 100644 index eb38716..0000000 --- a/kb_pipeline/processors/archive.py +++ /dev/null @@ -1,170 +0,0 @@ -"""ArchiveProcessor - Copy analysis reports to archive directory - -Final step in pipeline that copies the generated analysis.md -to the archive directory organized by platform. -""" - -from __future__ import annotations - -import logging -import shutil -from datetime import datetime -from pathlib import Path -from typing import Optional - -from ..config import get_settings -from ..schemas import ArchiveInput, ArchiveOutput -from ..state import WorkflowStateManager -from .base import BaseProcessor - -logger = logging.getLogger(__name__) - - -class ArchiveProcessor(BaseProcessor): - """Archive 处理器 - 归档分析报告 - - 处理流程： - 1. 检查步骤是否已完成（幂等执行） - 2. 检查归档目录是否存在相同 content_id 的文件 - 3. 如果存在重复，询问用户是否覆盖 - 4. 复制 md 文件到归档目录 - 5. 记录处理状态到 workflow_state.json - - 归档目录结构： - {KB_ARCHIVE_PATH}/{platform}/{YYYYMMDD}_{content_id}_analysis.md - """ - - STEP_NAME = "archive" - - async def process(self, input_data: ArchiveInput) -> ArchiveOutput: - """执行归档处理 - - Args: - input_data: Archive 输入数据 - - Returns: - Archive 处理结果 - """ - try: - # Step 1: 检查是否已完成（幂等执行） - if self.state_manager: - state = self.state_manager.load() - if state and state.is_step_completed(self.STEP_NAME): - logger.info(f"Step {self.STEP_NAME} already completed, skipping") - return ArchiveOutput( - success=True, - skipped=True, - platform=input_data.platform, - archive_file="" - ) - - # Step 2: 准备归档路径 - archive_base = Path(self.settings.archive_path) - platform_dir = archive_base / input_data.platform - platform_dir.mkdir(parents=True, exist_ok=True) - - # 文件命名: {YYYYMMDD}_{content_id}_analysis.md - timestamp = datetime.now().strftime("%Y%m%d") - archive_filename = f"{timestamp}_{input_data.content_id}_analysis.md" - archive_path = platform_dir / archive_filename - - # Step 3: 检查重复 - existing_file = self._find_existing_archive( - platform_dir, input_data.content_id - ) - if existing_file: - should_overwrite = self._ask_overwrite(existing_file) - if not should_overwrite: - logger.info(f"User chose not to overwrite, skipping archive") - return ArchiveOutput( - success=True, - skipped=True, - platform=input_data.platform, - archive_file=str(existing_file) - ) - # 删除旧文件 - existing_file.unlink() - logger.info(f"Removed existing archive: {existing_file}") - - # Step 4: 复制文件 - source_path = Path(input_data.analysis_file) - if not source_path.exists(): - raise FileNotFoundError(f"Analysis file not found: {source_path}") - - shutil.copy2(source_path, archive_path) - logger.info(f"Archived: {source_path} -> {archive_path}") - - # Step 5: 记录成功状态 - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_completed( - self.STEP_NAME, - f"archived to {archive_path}" - ) - self.state_manager.save(state) - - return ArchiveOutput( - success=True, - skipped=False, - platform=input_data.platform, - archive_file=str(archive_path) - ) - - except Exception as e: - logger.error(f"Archive failed: {e}") - - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_failed(self.STEP_NAME, str(e)) - self.state_manager.save(state) - - return ArchiveOutput( - success=False, - error_message=str(e), - platform=input_data.platform, - archive_file="" - ) - - def _find_existing_archive(self, platform_dir: Path, content_id: str) -> Optional[Path]: - """查找已存在的归档文件 - - 使用 glob 模式 `*_{content_id}_analysis.md` 搜索 - - Args: - platform_dir: 平台目录路径 - content_id: 内容 ID - - Returns: - 已存在的文件路径，如果不存在返回 None - """ - pattern = f"*_{content_id}_analysis.md" - matches = list(platform_dir.glob(pattern)) - return matches[0] if matches else None - - def _ask_overwrite(self, existing_file: Path) -> bool: - """询问用户是否覆盖已存在的文件 - - 使用 rich.prompt.Confirm 模式（与 Pipeline._ask_rerun_step 相同） - - Args: - existing_file: 已存在的文件路径 - - Returns: - True 如果用户选择覆盖 - """ - try: - from rich.console import Console - from rich.prompt import Confirm - - console = Console() - return Confirm.ask( - f"Archive already exists: {existing_file.name}\nOverwrite?", - default=False - ) - except ImportError: - response = input( - f"Archive already exists: {existing_file.name}\nOverwrite? [y/N] " - ) - return response.lower() in ('y', 'yes') diff --git a/kb_pipeline/processors/asr.py b/kb_pipeline/processors/asr.py deleted file mode 100644 index b94bbaf..0000000 --- a/kb_pipeline/processors/asr.py +++ /dev/null @@ -1,685 +0,0 @@ -"""ASRProcessor - Audio-to-text conversion - -Unified processor for B站官方字幕下载 and Whisper.cpp 语音识别. -Follows the pipeline pattern established in Phase 1/2. - -Strategy: -1. Try official subtitle first (highest quality, no compute cost) -2. Fall back to Whisper.cpp if no subtitle available -3. Auto-enable speaker diarization if HF_TOKEN is configured -""" - -from __future__ import annotations - -import asyncio -import json -import logging -import os -import subprocess -from pathlib import Path -from typing import Optional, Dict, Any, List - -import aiohttp - -from ..config import get_settings -from ..schemas import ASRInput, ASROutput -from ..state import WorkflowStateManager - -logger = logging.getLogger(__name__) - - -class WhisperNotFoundError(RuntimeError): - """Whisper.cpp 路径未找到""" - pass - - -def get_whisper_cpp_path() -> str: - """获取 whisper.cpp 安装路径 - - 检查环境变量 WHISPER_CPP_PATH 或默认路径 ~/whisper.cpp - - Returns: - whisper.cpp 目录路径 - - Raises: - WhisperNotFoundError: 如果路径不存在 - """ - # 检查环境变量 - path = os.environ.get('WHISPER_CPP_PATH') - if path and os.path.exists(path): - return os.path.expanduser(path) - - # 检查默认路径 - default = os.path.expanduser('~/whisper.cpp') - if os.path.exists(default): - return default - - raise WhisperNotFoundError( - "未找到 whisper.cpp。请设置 WHISPER_CPP_PATH 环境变量或安装到 ~/whisper.cpp" - ) - - -def convert_to_wav(input_path: str, output_path: Optional[str] = None) -> str: - """使用 ffmpeg 将音频/视频转换为 16kHz WAV 格式 - - Args: - input_path: 输入文件路径 - output_path: 输出文件路径（可选，默认为 input_path_audio.wav） - - Returns: - 输出 WAV 文件路径 - """ - if output_path is None: - base = os.path.splitext(input_path)[0] - output_path = f"{base}_audio.wav" - - cmd = [ - 'ffmpeg', '-y', '-i', input_path, - '-ar', '16000', # 16kHz 采样率 - '-ac', '1', # 单声道 - '-c:a', 'pcm_s16le', # 16-bit PCM - output_path - ] - - subprocess.run(cmd, capture_output=True, check=True) - return output_path - - -class ASRProcessor: - """ASR 处理器 - 统一字幕下载和语音识别 - - 处理流程： - 1. 如果是 B站视频，尝试下载官方字幕 - 2. 无官方字幕则调用 Whisper.cpp - 3. HF_TOKEN 配置时自动启用说话者分离 - 4. 记录处理状态到 workflow_state.json - - Attributes: - STEP_NAME: 步骤名称，用于状态记录 - settings: KB 配置 - state_manager: 状态管理器（可选） - """ - - STEP_NAME = "asr" - - def __init__(self, state_manager: Optional[WorkflowStateManager] = None): - """初始化 ASR 处理器 - - Args: - state_manager: 状态管理器（可选，用于记录处理状态） - """ - self.settings = get_settings() - self.state_manager = state_manager - - async def process(self, input_data: ASRInput) -> ASROutput: - """执行 ASR 处理 - - 策略： - 0. 先检查是否已有字幕文件（幂等性） - 1. 先尝试官方字幕（仅 B站） - 2. 无官方字幕则调用 Whisper - 3. 记录状态 - - Args: - input_data: ASR 输入数据 - - Returns: - ASR 处理结果 - - Raises: - Exception: 处理失败时抛出异常 - """ - result = None - - try: - # Step 0: 检查是否已有字幕文件（幂等性） - existing_result = self._check_existing_subtitle_files(input_data) - if existing_result: - logger.info(f"Found existing subtitle files, skipping processing") - result = existing_result - else: - # Step 1: 尝试官方字幕（仅 B站） - if input_data.platform == "bilibili": - result = await self._try_official_subtitle(input_data) - - # Step 2: 无官方字幕则调用 Whisper - if result is None: - result = await self._run_whisper(input_data) - - # Step 3: 记录成功状态 - if self.state_manager: - state = self.state_manager.load() - if state: - summary = f"tool={result.tool}, segments={len(result.segments)}" - if result.speakers: - summary += f", speakers={len(result.speakers)}" - state.mark_step_completed(self.STEP_NAME, summary) - self.state_manager.save(state) - - return result - - except Exception as e: - # 记录失败状态（只在所有重试都失败后） - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_failed(self.STEP_NAME, str(e)) - self.state_manager.save(state) - raise - - def _check_existing_subtitle_files(self, input_data: ASRInput) -> Optional[ASROutput]: - """检查是否已存在字幕文件（幂等性检查） - - Args: - input_data: ASR 输入数据 - - Returns: - 如果找到现有字幕文件返回 ASROutput，否则返回 None - """ - output_dir = Path(input_data.content_dir) - - # 尝试两种命名格式：新格式（带 _subtitle 后缀）和旧格式 - base_names = [ - f"{input_data.platform}_{input_data.content_id}_subtitle", - f"{input_data.platform}_{input_data.content_id}", - ] - - for base_name in base_names: - json_path = output_dir / f"{base_name}.json" - srt_path = output_dir / f"{base_name}.srt" - txt_path = output_dir / f"{base_name}.txt" - - if json_path.exists(): - # 读取现有字幕数据 - segments = [] - text = "" - duration = 0.0 - - try: - with open(json_path, 'r', encoding='utf-8') as f: - data = json.load(f) - - # 解析字幕格式 - if 'transcription' in data: - # Whisper 格式 - for i, item in enumerate(data['transcription']): - offsets = item.get('offsets', {}) - if isinstance(offsets, dict): - # 新格式：{"from": 0, "to": 5320} - start = offsets.get('from', 0) / 1000.0 - end = offsets.get('to', 0) / 1000.0 - else: - # 旧格式：[0, 5320] - start = offsets[0] / 1000.0 if len(offsets) > 0 else 0 - end = offsets[1] / 1000.0 if len(offsets) > 1 else 0 - segments.append({ - 'id': i, - 'start': start, - 'end': end, - 'text': item.get('text', '') - }) - elif 'body' in data: - # B站官方字幕格式 - for i, item in enumerate(data['body']): - segments.append({ - 'id': i, - 'start': item.get('from', 0), - 'end': item.get('to', 0), - 'text': item.get('content', '') - }) - - text = ' '.join(s['text'] for s in segments) - if segments: - duration = segments[-1]['end'] - - output_files = {'json': str(json_path)} - if srt_path.exists(): - output_files['srt'] = str(srt_path) - if txt_path.exists(): - output_files['txt'] = str(txt_path) - - return ASROutput( - success=True, - tool="existing", - has_subtitle=True, - output_files=output_files, - segments=segments, - text=text, - duration_seconds=duration - ) - except Exception as e: - logger.warning(f"Failed to read existing subtitle file: {e}") - continue - - return None - - async def _try_official_subtitle(self, input_data: ASRInput) -> Optional[ASROutput]: - """尝试下载 B站官方字幕 - - Args: - input_data: ASR 输入数据 - - Returns: - 如果找到字幕返回 ASROutput，否则返回 None - - Note: - 无字幕返回 None（不是抛异常），让流程继续到 Whisper - """ - # 使用 content_dir 作为输出目录（而不是从 video_path 推断） - output_dir = Path(input_data.content_dir) - # 使用 content_id 作为基础文件名 - base_name = f"{input_data.platform}_{input_data.content_id}_subtitle" - - # 输出文件路径 - txt_path = output_dir / f"{base_name}.txt" - srt_path = output_dir / f"{base_name}.srt" - json_path = output_dir / f"{base_name}.json" - - async with aiohttp.ClientSession() as session: - try: - # Step 1: 获取 CID - bvid = input_data.content_id - pagelist_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}" - async with session.get(pagelist_url) as resp: - pagelist_data = await resp.json() - - pages = pagelist_data.get('data', []) - if not pages: - logger.info(f"No pagelist data for {bvid}") - return None - - cid = pages[0].get('cid') - if not cid: - logger.info(f"No cid found for {bvid}") - return None - - # Step 2: 获取字幕列表 - player_url = f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}" - async with session.get(player_url) as resp: - player_data = await resp.json() - - subtitle_info = player_data.get('data', {}).get('subtitle', {}) - subtitles = subtitle_info.get('list', []) - - if not subtitles: - logger.info(f"No official subtitle for {bvid}") - return None - - # Step 3: 下载第一个字幕（优先 AI 生成的中文字幕） - subtitle_url = None - for sub in subtitles: - if sub.get('lan', '').startswith('zh'): - subtitle_url = sub.get('subtitle_url') - break - - if not subtitle_url and subtitles: - subtitle_url = subtitles[0].get('subtitle_url') - - if not subtitle_url: - return None - - # 补全 URL - if subtitle_url.startswith('//'): - subtitle_url = f"https:{subtitle_url}" - - async with session.get(subtitle_url) as resp: - subtitle_content = await resp.json() - - # Step 4: 转换为输出格式 - segments = [] - text_parts = [] - body = subtitle_content.get('body', []) - - for i, item in enumerate(body): - start = item.get('from', 0) - end = item.get('to', 0) - content = item.get('content', '') - - segments.append({ - 'id': i, - 'start': start, - 'end': end, - 'text': content - }) - text_parts.append(content) - - full_text = ' '.join(text_parts) - duration = segments[-1]['end'] if segments else 0 - - # Step 5: 写入文件 - # TXT 格式 - txt_path.write_text(full_text, encoding='utf-8') - - # SRT 格式 - srt_content = self._convert_to_srt(segments) - srt_path.write_text(srt_content, encoding='utf-8') - - # JSON 格式 - json_data = { - 'success': True, - 'tool': 'subtitle', - 'segments': segments, - 'text': full_text, - 'duration_seconds': duration - } - json_path.write_text(json.dumps(json_data, ensure_ascii=False, indent=2), encoding='utf-8') - - return ASROutput( - success=True, - tool="subtitle", - has_subtitle=True, - output_files={ - 'txt': str(txt_path), - 'srt': str(srt_path), - 'json': str(json_path) - }, - segments=segments, - text=full_text, - duration_seconds=duration - ) - - except aiohttp.ClientError as e: - logger.warning(f"Failed to fetch subtitle: {e}") - return None - - def _convert_to_srt(self, segments: List[Dict]) -> str: - """将字幕分段转换为 SRT 格式 - - Args: - segments: 字幕分段列表，每个包含 start, end, text - - Returns: - SRT 格式字符串 - """ - lines = [] - for i, seg in enumerate(segments, 1): - start_time = self._seconds_to_srt_time(seg['start']) - end_time = self._seconds_to_srt_time(seg['end']) - lines.append(f"{i}") - lines.append(f"{start_time} --> {end_time}") - lines.append(seg['text']) - lines.append("") # 空行 - - return '\n'.join(lines) - - def _seconds_to_srt_time(self, seconds: float) -> str: - """将秒数转换为 SRT 时间格式 (HH:MM:SS,mmm)""" - hours = int(seconds // 3600) - minutes = int((seconds % 3600) // 60) - secs = int(seconds % 60) - millis = int((seconds % 1) * 1000) - return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" - - async def _run_whisper(self, input_data: ASRInput) -> ASROutput: - """执行 Whisper 语音识别 - - 根据 HF_TOKEN 配置自动选择普通模式或说话者分离模式。 - - Args: - input_data: ASR 输入数据 - - Returns: - ASR 处理结果 - """ - hf_token = self.settings.hf_token.get_secret_value() - if hf_token: - return await self._run_whisper_with_diarization(input_data) - else: - return await self._run_whisper_basic(input_data) - - async def _run_whisper_basic(self, input_data: ASRInput) -> ASROutput: - """执行普通 Whisper.cpp 语音识别 - - Args: - input_data: ASR 输入数据 - - Returns: - ASR 处理结果 - - Raises: - RuntimeError: Whisper.cpp 路径不存在或执行失败，或 video_path 为空 - """ - # 检查 video_path 是否有效 - if not input_data.video_path: - raise RuntimeError( - "video_path 为空，无法执行 Whisper 语音识别。" - "请确保视频文件已下载（对于没有官方字幕的视频，需要先下载视频文件）。" - ) - - video_path = Path(input_data.video_path) - output_dir = Path(input_data.content_dir) - base_name = f"{input_data.platform}_{input_data.content_id}_whisper" - - # 获取 Whisper.cpp 路径 - try: - whisper_path = get_whisper_cpp_path() - except WhisperNotFoundError as e: - raise RuntimeError(str(e)) - - # 模型路径 - model_path = os.path.join(whisper_path, 'models', 'ggml-base.bin') - if not os.path.exists(model_path): - raise RuntimeError(f"Whisper model not found: {model_path}") - - # 转换音频格式 - audio_path = convert_to_wav(str(video_path)) - temp_audio_created = audio_path != str(video_path) - - try: - # 构建 Whisper 命令 - output_base = os.path.join(output_dir, base_name) - cmd = [ - os.path.join(whisper_path, 'whisper-cli'), - '-m', model_path, - '-f', audio_path, - '-of', output_base, - '-l', 'zh', - '--output-txt', '--output-srt', '--output-json', - ] - - # 执行 Whisper - result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) - - if result.returncode != 0: - raise RuntimeError(f"Whisper failed: {result.stderr}") - - # 解析输出 - return self._parse_whisper_output(output_dir, base_name) - - finally: - # 清理临时音频文件 - if temp_audio_created and os.path.exists(audio_path): - try: - os.remove(audio_path) - except Exception: - pass # 静默失败，不阻塞主流程 - - def _parse_whisper_output(self, output_dir: Path, base_name: str) -> ASROutput: - """解析 Whisper 输出文件 - - Args: - output_dir: 输出目录 - base_name: 文件基础名 - - Returns: - ASR 处理结果 - """ - txt_path = output_dir / f"{base_name}.txt" - srt_path = output_dir / f"{base_name}.srt" - json_path = output_dir / f"{base_name}.json" - - # 读取 JSON 输出 - segments = [] - text = "" - duration = 0.0 - - if json_path.exists(): - with open(json_path, 'r', encoding='utf-8') as f: - data = json.load(f) - # Whisper.cpp JSON 格式 - if 'transcription' in data: - for i, item in enumerate(data['transcription']): - offsets = item.get('offsets', {}) - if isinstance(offsets, dict): - # 新格式：{"from": 0, "to": 5320} - start = offsets.get('from', 0) / 1000.0 - end = offsets.get('to', 0) / 1000.0 - else: - # 旧格式：[0, 5320] - start = offsets[0] / 1000.0 if len(offsets) > 0 else 0 - end = offsets[1] / 1000.0 if len(offsets) > 1 else 0 - segments.append({ - 'id': i, - 'start': start, - 'end': end, - 'text': item.get('text', '') - }) - text = ' '.join(s['text'] for s in segments) - if segments: - duration = segments[-1]['end'] - - # 读取文本 - if txt_path.exists(): - text = txt_path.read_text(encoding='utf-8').strip() - - return ASROutput( - success=True, - tool="whisper", - has_subtitle=False, - output_files={ - 'txt': str(txt_path), - 'srt': str(srt_path), - 'json': str(json_path) - }, - segments=segments, - text=text, - duration_seconds=duration - ) - - async def _run_whisper_with_diarization(self, input_data: ASRInput) -> ASROutput: - """执行带说话者分离的 Whisper - - Args: - input_data: ASR 输入数据 - - Returns: - ASR 处理结果，包含说话者信息 - """ - # 先运行普通 Whisper - basic_result = await self._run_whisper_basic(input_data) - - try: - # 运行说话者分离 - diarization_result = run_diarization(input_data.video_path) - - # 合并结果 - merged_segments = assign_speakers_to_segments( - basic_result.segments, - diarization_result - ) - - # 提取说话者列表 - speakers = list(set( - seg.get('speaker') for seg in merged_segments if seg.get('speaker') - )) - - return ASROutput( - success=True, - tool="whisper-diarization", - has_subtitle=False, - output_files=basic_result.output_files, - segments=merged_segments, - text=basic_result.text, - duration_seconds=basic_result.duration_seconds, - speakers=sorted(speakers) - ) - - except ImportError: - logger.warning("pyannote.audio not available, falling back to basic whisper") - return basic_result - except Exception as e: - logger.warning(f"Diarization failed: {e}, falling back to basic whisper") - return basic_result - - -def run_diarization(audio_path: str, num_speakers: Optional[int] = None) -> Dict[str, List[Dict]]: - """运行 pyannote.audio 说话者分离 - - Args: - audio_path: 音频文件路径 - num_speakers: 预期说话者数量（可选） - - Returns: - 说话者时间线字典 {speaker_name: [{start, end}, ...]} - - Raises: - ImportError: 如果 pyannote.audio 未安装 - RuntimeError: 如果 HF_TOKEN 未配置 - """ - hf_token = os.environ.get('HF_TOKEN') - if not hf_token: - raise RuntimeError("需要设置 HF_TOKEN 环境变量以使用说话者分离功能") - - from huggingface_hub import login - from pyannote.audio import Pipeline - import torch - - login(token=hf_token) - - pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1") - - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - pipeline.to(device) - - diarization = pipeline(audio_path, num_speakers=num_speakers) - - # 解析结果 - speakers = {} - for turn, _, speaker in diarization.itertracks(yield_label=True): - if speaker not in speakers: - speakers[speaker] = [] - speakers[speaker].append({ - "start": round(turn.start, 2), - "end": round(turn.end, 2) - }) - - return speakers - - -def assign_speakers_to_segments( - segments: List[Dict], - diarization: Dict[str, List[Dict]] -) -> List[Dict]: - """为每个段落分配说话者 - - 根据时间戳匹配说话者分离结果到 Whisper 段落。 - - Args: - segments: Whisper 段落列表 - diarization: 说话者分离结果 - - Returns: - 带有 speaker 字段的段落列表 - """ - result = [] - - for seg in segments: - seg_start = seg.get('start', 0) - seg_end = seg.get('end', 0) - seg_mid = (seg_start + seg_end) / 2 - - # 找到包含中点时间的说话者 - assigned_speaker = None - for speaker, turns in diarization.items(): - for turn in turns: - if turn['start'] <= seg_mid <= turn['end']: - assigned_speaker = speaker - break - if assigned_speaker: - break - - new_seg = dict(seg) - new_seg['speaker'] = assigned_speaker - result.append(new_seg) - - return result diff --git a/kb_pipeline/processors/base.py b/kb_pipeline/processors/base.py deleted file mode 100644 index 8adc259..0000000 --- a/kb_pipeline/processors/base.py +++ /dev/null @@ -1,55 +0,0 @@ -"""BaseProcessor - Abstract base class for all pipeline processors - -Provides a unified interface that all processors (parse, download, asr, analyze) -must follow, enabling consistent state management and error handling. -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import Optional, TYPE_CHECKING - -from kb_pipeline.config import get_settings -from kb_pipeline.schemas import StepInput, StepOutput - -if TYPE_CHECKING: - from kb_pipeline.state import WorkflowStateManager - - -class BaseProcessor(ABC): - """Abstract base class for all pipeline processors. - - All processors (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor) - must inherit from this class and implement the process() method. - - Attributes: - STEP_NAME: Class attribute identifying the processor step name. - settings: KBSettings instance from get_settings(). - state_manager: Optional WorkflowStateManager for state recording. - """ - - STEP_NAME: str - - def __init__(self, state_manager: Optional[WorkflowStateManager] = None): - """Initialize the processor. - - Args: - state_manager: Optional WorkflowStateManager for recording step state. - """ - self.settings = get_settings() - self.state_manager = state_manager - - @abstractmethod - async def process(self, input_data: StepInput) -> StepOutput: - """Process the input data and return the result. - - Args: - input_data: Input data for the processor step. - - Returns: - StepOutput with processing results. - - Raises: - Exception: If processing fails. - """ - pass diff --git a/kb_pipeline/processors/download.py b/kb_pipeline/processors/download.py deleted file mode 100644 index 82241c8..0000000 --- a/kb_pipeline/processors/download.py +++ /dev/null @@ -1,338 +0,0 @@ -"""DownloadProcessor - Content download processor - -Handles metadata extraction and content downloading for all platforms. -Works with PlatformAdapter implementations (BilibiliAdapter, ZhihuAdapter). - -Purpose: Unify metadata and content fetching into a single processor -""" - -from __future__ import annotations - -import json -import logging -from pathlib import Path -from typing import Optional, TYPE_CHECKING, Dict, Any - -from kb_pipeline.schemas import DownloadInput, DownloadOutput -from .base import BaseProcessor - -if TYPE_CHECKING: - from kb_pipeline.state import WorkflowStateManager - from kb_pipeline.platforms.base import PlatformAdapter - -logger = logging.getLogger(__name__) - - -class DownloadProcessor(BaseProcessor): - """Content download processor. - - Fetches metadata and content from platforms using PlatformAdapter. - Supports video, article, answer, and zhuanlan content types. - - Attributes: - STEP_NAME: "download" - identifies this processor step - """ - - STEP_NAME = "download" - - async def process(self, input_data: DownloadInput) -> DownloadOutput: - """Process the download request. - - Args: - input_data: DownloadInput with content_id, platform, content_dir, content_type - - Returns: - DownloadOutput with metadata_file, content_file, title, author, etc. - """ - content_dir = Path(input_data.content_dir) - content_dir.mkdir(parents=True, exist_ok=True) - - # Check if step already completed (idempotency) - if self.state_manager: - state = self.state_manager.load() - if state and state.is_step_completed(self.STEP_NAME): - logger.info(f"Step {self.STEP_NAME} already completed, skipping") - # Return a success output based on existing files - return self._get_existing_output(input_data, content_dir) - - # Get appropriate adapter - try: - adapter = self._get_adapter(input_data.platform) - except ValueError as e: - return self._record_failure(str(e)) - - try: - # Fetch metadata (pass source_url for zhihu answers) - if input_data.platform == "zhihu": - metadata = await adapter.fetch_metadata( - input_data.content_id, - input_data.content_type, - source_url=input_data.source_url - ) - else: - metadata = await adapter.fetch_metadata( - input_data.content_id, - input_data.content_type - ) - - # Save metadata - metadata_filename = adapter.get_output_filename( - input_data.content_id, - input_data.content_type, - "metadata.json" - ) - metadata_path = content_dir / metadata_filename - self._save_metadata(metadata, metadata_path) - - # Extract title and author from metadata - title = self._extract_title(metadata, input_data.platform) - author = self._extract_author(metadata, input_data.platform) - duration = self._extract_duration(metadata, input_data.content_type) - - # Fetch content (for text-based content) - content_file = "" - if input_data.content_type in ["article", "answer", "zhuanlan"]: - if input_data.platform == "zhihu": - content = await adapter.fetch_content( - input_data.content_id, - input_data.content_type, - source_url=input_data.source_url - ) - else: - content = await adapter.fetch_content( - input_data.content_id, - input_data.content_type - ) - content_filename = adapter.get_output_filename( - input_data.content_id, - input_data.content_type, - "content.md" - ) - content_path = content_dir / content_filename - self._save_content(content, content_path) - content_file = str(content_path) - - # Download video for ASR (Whisper needs video file) - if input_data.content_type == "video" and input_data.platform == "bilibili": - download_result = await adapter.download_video( - input_data.content_id, - content_dir - ) - if download_result["success"]: - content_file = download_result["filepath"] - logger.info(f"Video downloaded for ASR: {content_file}") - else: - logger.warning(f"Video download failed: {download_result['message']}") - - # Record success - result = DownloadOutput( - success=True, - metadata_file=str(metadata_path), - content_file=content_file, - title=title, - author=author, - duration=duration - ) - - if self.state_manager: - state = self.state_manager.load() - if state: - summary = f"title={title}, type={input_data.content_type}" - state.mark_step_completed(self.STEP_NAME, summary) - self.state_manager.save(state) - - return result - - except Exception as e: - logger.error(f"Download failed: {e}") - return self._record_failure(str(e)) - - def _get_adapter(self, platform: str) -> PlatformAdapter: - """Get the appropriate PlatformAdapter for the platform. - - Args: - platform: Platform name ("bilibili" or "zhihu") - - Returns: - PlatformAdapter instance - - Raises: - ValueError: If platform is not supported - """ - if platform == "bilibili": - from kb_pipeline.platforms.bilibili import BilibiliAdapter - return BilibiliAdapter(state_manager=self.state_manager) - elif platform == "zhihu": - from kb_pipeline.platforms.zhihu import ZhihuAdapter - return ZhihuAdapter() - else: - raise ValueError(f"Unsupported platform: {platform}") - - def _save_metadata(self, metadata: Dict[str, Any], path: Path) -> None: - """Save metadata to JSON file. - - Args: - metadata: Metadata dictionary - path: Output file path - """ - with open(path, 'w', encoding='utf-8') as f: - json.dump(metadata, f, ensure_ascii=False, indent=2) - logger.info(f"Saved metadata to {path}") - - def _save_content(self, content: str, path: Path) -> None: - """Save content to Markdown file. - - Args: - content: Markdown content string - path: Output file path - """ - with open(path, 'w', encoding='utf-8') as f: - f.write(content) - logger.info(f"Saved content to {path}") - - def _extract_title(self, metadata: Dict[str, Any], platform: str) -> str: - """Extract title from metadata. - - Args: - metadata: Metadata dictionary - platform: Platform name - - Returns: - Title string - """ - if platform == "bilibili": - # Bilibili metadata structure: {metadata: {title: ...}, ...} - if "metadata" in metadata: - return metadata["metadata"].get("title", "") - return metadata.get("title", "") - else: - return metadata.get("title", "") - - def _extract_author(self, metadata: Dict[str, Any], platform: str) -> str: - """Extract author from metadata. - - Args: - metadata: Metadata dictionary - platform: Platform name - - Returns: - Author string - """ - if platform == "bilibili": - # Bilibili metadata structure: {metadata: {mid: ...}, ...} - if "metadata" in metadata: - return str(metadata["metadata"].get("mid", "")) - return str(metadata.get("mid", "")) - else: - return metadata.get("author", "") - - def _extract_duration(self, metadata: Dict[str, Any], content_type: str) -> Optional[int]: - """Extract duration from metadata for video content. - - Args: - metadata: Metadata dictionary - content_type: Content type - - Returns: - Duration in seconds for videos, None otherwise - """ - if content_type != "video": - return None - - # Bilibili video duration - if "metadata" in metadata and "duration" in metadata["metadata"]: - duration = metadata["metadata"]["duration"] - # Handle formatted duration string (e.g., "12:34" or "01:23:45") - if isinstance(duration, str): - if not duration.strip(): - return None - parts = duration.split(':') - try: - if len(parts) == 2: - return int(parts[0]) * 60 + int(parts[1]) - elif len(parts) == 3: - return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2]) - except (ValueError, IndexError): - return None - elif isinstance(duration, (int, float)): - return int(duration) - - return None - - def _record_failure(self, error_message: str) -> DownloadOutput: - """Record failure to state manager and return failure output. - - Args: - error_message: Error message - - Returns: - DownloadOutput with success=False - """ - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_failed(self.STEP_NAME, error_message) - self.state_manager.save(state) - - return DownloadOutput( - success=False, - error_message=error_message, - metadata_file="", - content_file="", - title="", - author="" - ) - - def _get_existing_output( - self, - input_data: DownloadInput, - content_dir: Path - ) -> DownloadOutput: - """Get output from existing files when step was already completed. - - Args: - input_data: DownloadInput - content_dir: Content directory path - - Returns: - DownloadOutput based on existing files - """ - adapter = self._get_adapter(input_data.platform) - - metadata_filename = adapter.get_output_filename( - input_data.content_id, - input_data.content_type, - "metadata.json" - ) - metadata_path = content_dir / metadata_filename - - # Try to read metadata for title/author - title = "" - author = "" - duration = None - if metadata_path.exists(): - try: - with open(metadata_path, 'r', encoding='utf-8') as f: - metadata = json.load(f) - title = self._extract_title(metadata, input_data.platform) - author = self._extract_author(metadata, input_data.platform) - duration = self._extract_duration(metadata, input_data.content_type) - except Exception: - pass - - content_filename = adapter.get_output_filename( - input_data.content_id, - input_data.content_type, - "content.md" - ) - content_path = content_dir / content_filename - content_file = str(content_path) if content_path.exists() else "" - - return DownloadOutput( - success=True, - metadata_file=str(metadata_path), - content_file=content_file, - title=title, - author=author, - duration=duration - ) diff --git a/kb_pipeline/processors/parse.py b/kb_pipeline/processors/parse.py deleted file mode 100644 index 6164a25..0000000 --- a/kb_pipeline/processors/parse.py +++ /dev/null @@ -1,83 +0,0 @@ -"""ParseProcessor - URL parsing processor - -Parses URLs from various platforms (Bilibili, Zhihu) and extracts -platform, content type, and content ID information. -""" - -from __future__ import annotations - -from typing import Optional, TYPE_CHECKING - -from kb_pipeline.config import get_settings -from kb_pipeline.link_parser import LinkParser -from kb_pipeline.schemas import ParseInput, ParseOutput -from .base import BaseProcessor - -if TYPE_CHECKING: - from kb_pipeline.state import WorkflowStateManager - - -class ParseProcessor(BaseProcessor): - """URL parsing processor. - - Parses URLs from Bilibili and Zhihu platforms, extracting - platform, content type, and content ID. - - Attributes: - STEP_NAME: "parse" - identifies this processor step - """ - - STEP_NAME = "parse" - - async def process(self, input_data: ParseInput) -> ParseOutput: - """Parse the URL and extract platform information. - - Args: - input_data: ParseInput with URL to parse - - Returns: - ParseOutput with parsed information or error message - """ - parser = LinkParser() - parsed = parser.parse(input_data.url) - - if parsed is None: - # Unknown URL format - result = ParseOutput( - success=False, - error_message="Unsupported URL format", - platform="", - content_type="", - content_id="", - resolved_url="", - original_url=input_data.url - ) - - # Record failure state - if self.state_manager: - state = self.state_manager.load() - if state: - state.mark_step_failed(self.STEP_NAME, result.error_message) - self.state_manager.save(state) - - return result - - # Successful parse - result = ParseOutput( - success=True, - platform=parsed.platform, - content_type=parsed.content_type, - content_id=parsed.content_id, - resolved_url=parsed.resolved_url, - original_url=parsed.original_url - ) - - # Record success state - if self.state_manager: - state = self.state_manager.load() - if state: - summary = f"platform={parsed.platform}, type={parsed.content_type}, id={parsed.content_id}" - state.mark_step_completed(self.STEP_NAME, summary) - self.state_manager.save(state) - - return result diff --git a/kb_pipeline/retry.py b/kb_pipeline/retry.py deleted file mode 100644 index ecd5c67..0000000 --- a/kb_pipeline/retry.py +++ /dev/null @@ -1,51 +0,0 @@ -"""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_pipeline/schemas.py b/kb_pipeline/schemas.py deleted file mode 100644 index 83a3e02..0000000 --- a/kb_pipeline/schemas.py +++ /dev/null @@ -1,446 +0,0 @@ -"""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_pipeline/state.py b/kb_pipeline/state.py deleted file mode 100644 index 598c6a5..0000000 --- a/kb_pipeline/state.py +++ /dev/null @@ -1,152 +0,0 @@ -""" -WorkflowStateManager - 步骤级状态持久化管理器 - -提供工作流状态的原子写入、备份轮转和崩溃恢复功能。 - -Purpose: 为工作流提供可靠的状态管理，支持断点续传和崩溃恢复 -""" - -from __future__ import annotations - -import json -import os -import tempfile -from pathlib import Path -from typing import Optional - -from kb_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 - )