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