kb-skills/kb_pipeline/platforms/bilibili.py

"""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号)
- 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 logging
import re
import sys
from pathlib import Path
from typing import Dict, Any, Optional

from .base import PlatformAdapter
from ..errors import NetworkError, AuthError
from ..retry import retry_on_network_error
from ..state import WorkflowStateManager

logger = logging.getLogger(__name__)

# Add kb-bilibili-metadata to path if not installed as package
_BILIBILI_METADATA_PATH = Path(__file__).parent.parent.parent / "kb-bilibili-metadata"
if _BILIBILI_METADATA_PATH.exists():
    sys.path.insert(0, str(_BILIBILI_METADATA_PATH))

try:
    from bilibili_extractor import BilibiliExtractor
    BILIBILI_EXTRACTOR_AVAILABLE = True
except ImportError:
    BILIBILI_EXTRACTOR_AVAILABLE = False
    BilibiliExtractor = None


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
        """
        if not BILIBILI_EXTRACTOR_AVAILABLE:
            raise ImportError(
                "BilibiliExtractor not available. "
                "Ensure kb-bilibili-metadata/bilibili_extractor.py is accessible."
            )

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