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号)
- Video download using yt-dlp (for ASR/Whisper)
- Automatic retry on network errors (3 attempts with exponential backoff)
- State recording for failed steps
- Fallback from API to Playwright for 403 errors (handled by BilibiliExtractor)
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
import tempfile
from pathlib import Path
from typing import Dict, Any, Optional

from .base import PlatformAdapter
from ..errors import NetworkError, AuthError
from ..retry import retry_on_network_error
from ..state import WorkflowStateManager
from kb_workflow.metadata.bilibili_extractor import BilibiliExtractor

logger = logging.getLogger(__name__)


class BilibiliAdapter(PlatformAdapter):
    """Bilibili platform adapter for video and article content

    Implements the PlatformAdapter interface to provide:
    - Content type identification from URL
    - Video metadata extraction
    - Article metadata extraction
    - Article content fetching in Markdown format
    - Automatic error retry and state recording

    Attributes:
        PLATFORM: Platform identifier string ("bilibili")
        _extractor: BilibiliExtractor instance for actual extraction
        _state_manager: Optional WorkflowStateManager for recording step status

    Example:
        adapter = BilibiliAdapter()
        content_type = adapter.get_content_type("https://www.bilibili.com/video/BV1xx...")
        # -> "video"

        metadata = await adapter.fetch_metadata("BV1xx...", "video")
        # -> {"metadata": {...}, "stats": {...}, ...}
    """

    PLATFORM = "bilibili"

    # URL patterns for content type identification
    VIDEO_PATTERN = re.compile(r'bilibili\.com/video/(BV[a-zA-Z0-9]+)')
    ARTICLE_PATTERN = re.compile(r'bilibili\.com/read/(cv\d+)')

    def __init__(
        self,
        headless: bool = True,
        timeout: int = 30,
        state_manager: Optional[WorkflowStateManager] = None
    ):
        """Initialize BilibiliAdapter

        Args:
            headless: Whether to use headless mode for Playwright (default True)
            timeout: Request timeout in seconds (default 30)
            state_manager: Optional WorkflowStateManager for recording step status
        """
        self._extractor = BilibiliExtractor(headless=headless, timeout=timeout)
        self._state_manager = state_manager

    def get_content_type(self, url: str) -> str:
        """Identify content type from URL

        Args:
            url: Bilibili URL to analyze

        Returns:
            Content type string: "video", "article", or "unknown"

        Example:
            adapter.get_content_type("https://www.bilibili.com/video/BV1xx...")
            -> "video"
        """
        if self.VIDEO_PATTERN.search(url):
            return "video"
        if self.ARTICLE_PATTERN.search(url):
            return "article"
        return "unknown"

    async def fetch_metadata(
        self,
        content_id: str,
        content_type: str
    ) -> Dict[str, Any]:
        """Fetch metadata for the given content

        Args:
            content_id: BV号 for videos, CV号 for articles
            content_type: "video" or "article"

        Returns:
            Dictionary containing metadata. Structure varies by content type:
            - Video: {metadata: {...}, stats: {...}, description: str}
            - Article: {cvid, title, author, publish_time, ...}

        Raises:
            NetworkError: If network request fails after 3 retries
            AuthError: If authentication fails (403/quota exceeded)
            ValueError: If content_type is not supported
        """
        try:
            return await self._fetch_metadata_with_retry(content_id, content_type)
        except (NetworkError, AuthError) as e:
            # Record failure to state manager only after all retries exhausted
            self._record_step_failed("metadata", str(e))
            raise
        except Exception as e:
            # Check if this is an API error that should be treated as network error
            if self._is_network_related_error(e):
                network_error = NetworkError(
                    step_name="metadata",
                    input_summary=f"content_id={content_id}, type={content_type}",
                    cause=e
                )
                self._record_step_failed("metadata", str(network_error))
                raise network_error
            raise

    @retry_on_network_error
    async def _fetch_metadata_with_retry(
        self,
        content_id: str,
        content_type: str
    ) -> Dict[str, Any]:
        """Internal method with retry logic for metadata fetching

        Args:
            content_id: BV号 for videos, CV号 for articles
            content_type: "video" or "article"

        Returns:
            Dictionary containing metadata

        Raises:
            NetworkError: If network request fails
            AuthError: If authentication fails
            ValueError: If content_type is not supported
        """
        if content_type == "video":
            # Call async method directly to avoid nested asyncio.run()
            return await self._extractor._extract_video_async(content_id)
        elif content_type == "article":
            # Call async method directly to avoid nested asyncio.run()
            return await self._extractor._extract_article_async(content_id)
        else:
            raise ValueError(f"Unsupported content type: {content_type}")

    def _is_network_related_error(self, error: Exception) -> bool:
        """Check if an error is network-related and should trigger retry

        Args:
            error: Exception to check

        Returns:
            True if error is network-related
        """
        error_str = str(error).lower()
        network_indicators = [
            'timeout', 'connection', 'network', '503', '502', '500',
            'connection refused', 'connection reset', 'eof',
        ]
        return any(indicator in error_str for indicator in network_indicators)

    async def fetch_content(
        self,
        content_id: str,
        content_type: str
    ) -> str:
        """Fetch content body in Markdown format

        For videos, this returns an empty string as video content
        extraction is handled separately (ASR transcription).

        For articles, this fetches the article content and formats
        it with YAML frontmatter.

        Args:
            content_id: CV号 for articles
            content_type: "video" or "article"

        Returns:
            Markdown-formatted content string with YAML frontmatter,
            or empty string for videos
        """
        if content_type == "video":
            # Video content is handled by ASR in Phase 3
            return ""

        if content_type == "article":
            # Fetch article metadata first
            metadata = await self.fetch_metadata(content_id, "article")

            # Fetch article content (requires additional implementation)
            content = await self._fetch_article_content(content_id)

            # Format with YAML frontmatter
            return self._format_content_markdown(metadata, content)

        raise ValueError(f"Unsupported content type: {content_type}")

    async def _fetch_article_content(self, cvid: str) -> str:
        """Fetch article body content

        Args:
            cvid: Article CV号

        Returns:
            Raw article content string

        Note:
            This method currently returns a placeholder.
            Full implementation would use additional API calls
            or Playwright to fetch the article body.
        """
        # TODO: Implement full article content fetching
        # This would require additional API endpoints or Playwright
        logger.warning(f"Article content fetching not fully implemented for {cvid}")
        return ""

    def _format_content_markdown(
        self,
        metadata: Dict[str, Any],
        content: str,
        content_type: str = "article"
    ) -> str:
        """Format content with YAML frontmatter

        Args:
            metadata: Article metadata dictionary
            content: Article body content
            content_type: Content type string

        Returns:
            Markdown string with YAML frontmatter
        """
        import yaml
        from datetime import datetime

        frontmatter = {
            "title": metadata.get("title", ""),
            "author": metadata.get("author", ""),
            "source_url": metadata.get("url", ""),
            "platform": self.PLATFORM,
            "content_type": content_type,
            "fetched_at": datetime.now().isoformat(),
        }

        # Add additional metadata if available
        if "cvid" in metadata:
            frontmatter["cvid"] = metadata["cvid"]
        if "mid" in metadata:
            frontmatter["author_id"] = metadata["mid"]
        if "word_count" in metadata:
            frontmatter["word_count"] = metadata["word_count"]

        yaml_str = yaml.dump(frontmatter, allow_unicode=True, sort_keys=False)
        return f"---\n{yaml_str}---\n\n{content}"

    def get_output_filename(
        self,
        content_id: str,
        content_type: str,
        suffix: str
    ) -> str:
        """Generate output filename for content

        Args:
            content_id: BV号 or CV号
            content_type: "video" or "article"
            suffix: Filename suffix (e.g., "metadata.json", "content.md")

        Returns:
            Filename string in format "bilibili_{content_id}_{suffix}"

        Example:
            adapter.get_output_filename("BV1xx...", "video", "metadata.json")
            -> "bilibili_BV1xx..._metadata.json"
        """
        return f"{self.PLATFORM}_{content_id}_{suffix}"

    def _record_step_failed(self, step_name: str, error_message: str) -> None:
        """Record step failure to state manager

        Args:
            step_name: Name of the failed step
            error_message: Error message to record
        """
        if self._state_manager is not None:
            self._state_manager.mark_step_failed(step_name, error_message)
            logger.warning(f"Step '{step_name}' failed: {error_message}")

    async def download_video(
        self,
        bvid: str,
        output_dir: Path,
        quality: int = 32
    ) -> Dict[str, Any]:
        """Download Bilibili video using yt-dlp.

        Downloads only audio (sufficient for Whisper ASR) to save bandwidth.

        Args:
            bvid: Bilibili video ID (BV号)
            output_dir: Directory to save the video
            quality: Video quality (16=360P, 32=480P, 64=720P, 80=1080P)

        Returns:
            Dictionary with:
            - success: bool
            - filepath: str or None
            - message: str
        """
        result = {"success": False, "filepath": None, "message": ""}

        try:
            from ..config import get_settings

            url = f"https://www.bilibili.com/video/{bvid}"

            # Get cookie from settings or environment
            settings = get_settings()
            cookie = settings.bilibili_cookie.get_secret_value() or os.environ.get('BILIBILI_COOKIE', '')

            # Create cookie file if needed
            cookie_file = None
            if cookie:
                cookie_file = self._create_cookie_file(cookie)

            try:
                # Get video title for filename
                title = await self._get_video_title(url, cookie_file)
                safe_title = self._sanitize_filename(title, bvid)

                # Build yt-dlp command - download best audio only for ASR
                # This saves bandwidth since Whisper only needs audio
                output_template = str(output_dir / f"{safe_title}.%(ext)s")

                # Use best audio format for ASR
                cmd = [
                    "yt-dlp",
                    "-f", "bestaudio/best",
                    "-o", output_template,
                    "--no-download-archive",
                    "-q",
                    url
                ]

                if cookie_file:
                    cmd.extend(["--cookies", cookie_file])

                logger.info(f"Downloading video: {bvid}")
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE
                )
                stdout, stderr = await proc.communicate()

                # Find downloaded file
                downloaded_files = list(output_dir.glob(f"{safe_title}.*"))
                video_extensions = {'.mp4', '.mkv', '.webm', '.mp3', '.m4a', '.opus', '.wav'}
                downloaded_files = [f for f in downloaded_files if f.suffix.lower() in video_extensions]

                if downloaded_files:
                    video_file = downloaded_files[0]
                    result["success"] = True
                    result["filepath"] = str(video_file)
                    result["message"] = f"Downloaded: {video_file.name}"
                    logger.info(f"Video downloaded: {video_file}")
                else:
                    # Try to find any new media file in output_dir
                    all_media = list(output_dir.glob("*.*"))
                    all_media = [f for f in all_media if f.suffix.lower() in video_extensions]
                    if all_media:
                        # Get the most recently modified file
                        video_file = max(all_media, key=lambda f: f.stat().st_mtime)
                        result["success"] = True
                        result["filepath"] = str(video_file)
                        result["message"] = f"Downloaded: {video_file.name}"
                        logger.info(f"Video downloaded (fallback): {video_file}")
                    else:
                        result["message"] = "No downloaded file found"
                        logger.warning(f"No downloaded file found for {bvid}")

            finally:
                # Clean up cookie file
                if cookie_file and os.path.exists(cookie_file):
                    os.remove(cookie_file)

        except Exception as e:
            result["message"] = f"Download error: {str(e)}"
            logger.error(f"Video download failed for {bvid}: {e}")

        return result

    async def _get_video_title(self, url: str, cookie_file: Optional[str] = None) -> str:
        """Get video title using yt-dlp.

        Args:
            url: Bilibili video URL
            cookie_file: Optional path to cookie file

        Returns:
            Video title or fallback name
        """
        try:
            cmd = ["yt-dlp", "--get-title", url]
            if cookie_file:
                cmd.extend(["--cookies", cookie_file])

            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await proc.communicate()

            title = stdout.decode('utf-8').strip()
            if title and 'error' not in title.lower():
                return title
        except Exception as e:
            logger.debug(f"Failed to get video title: {e}")

        return "video"

    def _sanitize_filename(self, title: str, bvid: str) -> str:
        """Generate standardized filename for video download.

        Args:
            title: Video title (unused, kept for API compatibility)
            bvid: Video BV号

        Returns:
            Standardized filename in format "bilibili_{bvid}"
        """
        # Use standardized naming convention: platform_bvid
        return f"bilibili_{bvid}"

    def _create_cookie_file(self, cookie: str) -> str:
        """Create Netscape format cookie file for yt-dlp.

        Args:
            cookie: Cookie string (e.g., "SESSDATA=xxx; bili_jct=xxx")

        Returns:
            Path to temporary cookie file
        """
        # Parse cookie string
        cookies = {}
        for item in cookie.split(';'):
            item = item.strip()
            if '=' in item:
                key, value = item.split('=', 1)
                cookies[key.strip()] = value.strip()

        # Create Netscape format
        cookie_content = "# Netscape HTTP Cookie File\n"
        for name, value in cookies.items():
            cookie_content += f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n"

        # Write to temp file
        fd, path = tempfile.mkstemp(suffix='.txt')
        with os.fdopen(fd, 'w') as f:
            f.write(cookie_content)

        return path