feat(download): add video download capability for videos without official subtitles

- Add download_video() to BilibiliAdapter using yt-dlp
- Add bilibili_cookie config to Settings
- Update DownloadProcessor to call video download for bilibili videos
- Add idempotency check to ASRProcessor for existing subtitle files
- Fix indentation error in asr.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

kb_pipeline/config.py

@@ -50,6 +50,9 @@ class KBSettings(BaseSettings):
     zhihu_cookie: SecretStr = Field(
         default=SecretStr(""), description="知乎 Cookie"
     )
+    bilibili_cookie: SecretStr = Field(
+        default=SecretStr(""), description="B站 Cookie (for video download)"
+    )
 
 
 @lru_cache

kb_pipeline/platforms/bilibili.py

@@ -6,6 +6,7 @@ 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)
@@ -13,9 +14,12 @@ Key features:
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import os
 import re
 import sys
+import tempfile
 from pathlib import Path
 from typing import Dict, Any, Optional
 
@@ -321,3 +325,188 @@ class BilibiliAdapter(PlatformAdapter):
         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:
+        """Sanitize title for use as filename.
+
+        Args:
+            title: Video title
+            bvid: Video BV号 for fallback
+
+        Returns:
+            Safe filename string
+        """
+        # Keep alphanumeric, spaces, and common Chinese-safe chars
+        safe_title = "".join(
+            c for c in title
+            if c.isalnum() or c in (' ', '-', '_', '丨', '｜')
+        ).strip()[:50]
+
+        # Use bvid if title is empty
+        if not safe_title:
+            safe_title = f"video_{bvid}"
+
+        return safe_title
+
+    def _create_cookie_file(self, cookie: str) -> str:
+        """Create Netscape format cookie file for yt-dlp.
+
+        Args:
+            cookie: Cookie string (e.g., "SESSDATA=xxx; bili_jct=xxx")
+
+        Returns:
+            Path to temporary cookie file
+        """
+        # Parse cookie string
+        cookies = {}
+        for item in cookie.split(';'):
+            item = item.strip()
+            if '=' in item:
+                key, value = item.split('=', 1)
+                cookies[key.strip()] = value.strip()
+
+        # Create Netscape format
+        cookie_content = "# Netscape HTTP Cookie File\n"
+        for name, value in cookies.items():
+            cookie_content += f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n"
+
+        # Write to temp file
+        fd, path = tempfile.mkstemp(suffix='.txt')
+        with os.fdopen(fd, 'w') as f:
+            f.write(cookie_content)
+
+        return path

kb_pipeline/platforms/zhihu.py

@@ -10,6 +10,7 @@ Provides content extraction for Zhihu platform:
 from typing import Dict, Any, Optional
 from datetime import datetime
 import re
+import asyncio
 
 from kb_pipeline.platforms.base import PlatformAdapter
 from kb_pipeline.errors import AuthError
@@ -146,12 +147,13 @@ class ZhihuAdapter(PlatformAdapter):
 
         return False
 
-    async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]:
+    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.)
@@ -169,15 +171,19 @@ class ZhihuAdapter(PlatformAdapter):
 
         # Build URL based on content type
         if content_type == "answer":
-            # For answers, we need the question ID too, but extractor handles this
+            # For answers, try to use source_url first, then fallback to constructed URL
-            url = f"https://www.zhihu.com/question/0/answer/{content_id}"
+            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
+        # Extract content in a separate thread to avoid blocking asyncio
-        result = self._extract_with_playwright(url)
+        result = await asyncio.to_thread(self._extract_with_playwright, url)
 
         # Check for login page
         if self._detect_login_page(result):
@@ -198,12 +204,13 @@ class ZhihuAdapter(PlatformAdapter):
 
         return result
 
-    async def fetch_content(self, content_id: str, content_type: str) -> str:
+    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
@@ -212,7 +219,7 @@ class ZhihuAdapter(PlatformAdapter):
             AuthError: If cookie is missing or expired
         """
         # Get metadata (which also validates auth)
-        metadata = await self.fetch_metadata(content_id, content_type)
+        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", "")

kb_pipeline/processors/asr.py

@@ -115,6 +115,7 @@ class ASRProcessor:
         """执行 ASR 处理
 
         策略：
+        0. 先检查是否已有字幕文件（幂等性）
         1. 先尝试官方字幕（仅 B站）
         2. 无官方字幕则调用 Whisper
         3. 记录状态
@@ -131,13 +132,19 @@ class ASRProcessor:
         result = None
 
         try:
-            # Step 1: 尝试官方字幕（仅 B站）
+            # Step 0: 检查是否已有字幕文件（幂等性）
-            if input_data.platform == "bilibili":
+            existing_result = self._check_existing_subtitle_files(input_data)
-                result = await self._try_official_subtitle(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
+                # Step 2: 无官方字幕则调用 Whisper
-            if result is None:
+                if result is None:
-                result = await self._run_whisper(input_data)
+                    result = await self._run_whisper(input_data)
 
             # Step 3: 记录成功状态
             if self.state_manager:

kb_pipeline/schemas.py

@@ -291,9 +291,11 @@ class DownloadInput(StepInput):
     Attributes:
         content_dir: 保存下载内容的目录路径
         content_type: 内容类型 (video, article, answer, zhuanlan)
+        source_url: 原始 URL（可选，用于知乎回答等需要完整 URL 的情况）
     """
     content_dir: str = Field(..., min_length=1, description="Directory to save downloaded content")
     content_type: str = Field(..., min_length=1, description="Content type (video, article, answer, zhuanlan)")
+    source_url: Optional[str] = Field(None, description="Original source URL")
 
 
 class DownloadOutput(StepOutput):