starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(06-02): remove original kb_pipeline directory

Browse Source 
All code has been moved to kb_workflow.pipeline with updated imports.
278 tests pass (14 pre-existing mock-related failures unrelated to refactoring).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-17 00:17:19 +08:00
parent c9ca1b538d
commit 2df08c6e26
20 changed files with 0 additions and 4192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kb_pipeline/__init__.py
-64
View File
@@ -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"
kb_pipeline/config.py
-72
View File
@@ -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()
kb_pipeline/errors.py
-174
View File
@@ -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
kb_pipeline/idempotency.py
-80
View File
@@ -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)
kb_pipeline/link_parser.py
-181
View File
@@ -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
kb_pipeline/pipeline.py
-399
View File
@@ -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}")
kb_pipeline/platforms/__init__.py
-21
View File
@@ -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",
]
kb_pipeline/platforms/base.py
-122
View File
@@ -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
kb_pipeline/platforms/bilibili.py
-485
View File
@@ -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
kb_pipeline/platforms/zhihu.py
-263
View File
@@ -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}"
kb_pipeline/processors/__init__.py
-26
View File
@@ -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",
]
kb_pipeline/processors/analyze.py
-325
View File
@@ -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}")
kb_pipeline/processors/archive.py
-170
View File
@@ -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')
kb_pipeline/processors/asr.py
-685
View File
@@ -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
kb_pipeline/processors/base.py
-55
View File
@@ -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
kb_pipeline/processors/download.py
-338
View File
@@ -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
)
kb_pipeline/processors/parse.py
-83
View File
@@ -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
kb_pipeline/retry.py
-51
View File
@@ -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
"""
kb_pipeline/schemas.py
-446
View File
@@ -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")
kb_pipeline/state.py
-152
View File
@@ -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
)