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 <noreply@anthropic.com>

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",
+]

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

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

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}"

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",
+]

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}")

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')

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

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

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
+        )

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