.planning/debug/bilibili-video-subtitle-empty.md

@@ -0,0 +1,116 @@
+---
+status: resolved
+trigger: "kb-workflow 处理 Bilibili 视频 BV16cNMzpExh 时，字幕内容为空，导致生成的笔记没有真实内容"
+created: 2026-03-15T10:30:00Z
+updated: 2026-03-16T00:30:00Z
+---
+
+## Current Focus
+
+hypothesis: 已实现视频下载功能，DownloadProcessor 现在会为 bilibili video 类型下载视频文件
+test: 验证导入成功，等待用户端到端测试
+expecting: video_path 非空，ASR Whisper 能正常执行语音识别
+next_action: 用户验证端到端流程
+
+## Symptoms
+
+expected: 视频字幕应该被提取；如果没有官方字幕，应该通过 Whisper 语音分析生成字幕
+actual: 字幕内容为空，生成的笔记显示"【内容】（完整字幕）部分为空白"
+errors: 无显式错误，kb-workflow 命令执行成功，但内容为空
+reproduction: 对 Bilibili 视频 BV16cNMzpExh 运行 kb-workflow process
+started: asyncio 修复后首次测试发现
+additional_context: 用户期望即使没有官方字幕，系统也应该执行 Whisper 语音分析生成字幕
+
+## Eliminated
+
+<!-- 暂无 -->
+
+## Evidence
+
+- timestamp: 2026-03-15T10:30:00Z
+  checked: kb_pipeline/processors/download.py 第 86-99 行
+  found: DownloadProcessor 对于 video 类型，content_file 被设置为空字符串（第 93 行：`content_file = ""`）
+  implication: ASR 步骤无法获得视频文件路径
+
+- timestamp: 2026-03-15T10:31:00Z
+  checked: kb_pipeline/pipeline.py 第 186-192 行
+  found: ASRInput 从 download 步骤的 `content_file` 获取 video_path
+  implication: 如果 download 步骤没有提供有效的 content_file，ASR 将收到空路径
+
+- timestamp: 2026-03-15T10:32:00Z
+  checked: kb_pipeline/processors/asr.py 第 163-290 行
+  found: _try_official_subtitle 方法只依赖 bvid 调用 B站 API，不依赖 video_path；但 Whisper 降级需要 video_path
+  implication: 官方字幕可能正常获取（如果有），但 Whisper 降级会失败
+
+- timestamp: 2026-03-15T10:40:00Z
+  checked: kb_pipeline/processors/asr.py 第 175-177 行和 Python Path("") 行为
+  found: `_try_official_subtitle` 第 175 行直接使用 `Path(input_data.video_path)`；当 video_path 为空时，Path("") 的 parent 是 "."（当前目录），stem 是空字符串
+  implication: 字幕文件会写入错误的目录（当前工作目录而不是内容目录）
+
+- timestamp: 2026-03-15T10:42:00Z
+  checked: kb_pipeline/pipeline.py 第 184-195 行和 ASRInput 构建逻辑
+  found: ASRInput 使用 `download_result.get("content_file", "")` 作为 video_path，并传入 `content_dir`；但 ASRProcessor 的 _try_official_subtitle 只用 video_path 而忽略 content_dir
+  implication: content_dir 已经传入 ASRInput 但没有被正确使用
+
+- timestamp: 2026-03-15T12:00:00Z
+  checked: pipeline.py 第 155-157 行的跳过逻辑
+  found: 当 ASR 步骤被跳过（已完成）时，results 字典中不包含 "asr" 键，导致 analyze 步骤找不到字幕文件
+  implication: analyze 步骤的 content_file 为空，无法分析内容
+
+- timestamp: 2026-03-15T12:10:00Z
+  checked: 已实现修复 - 在跳过步骤时添加用户确认和文件扫描逻辑
+  found: _populate_skipped_step_results 方法能扫描 content_dir 找到现有的字幕文件（{platform}_{content_id}_subtitle.json）
+  implication: 跳过 ASR 步骤时，results["asr"]["output_files"]["json"] 会被正确填充
+
+- timestamp: 2026-03-15T13:00:00Z
+  checked: kb_pipeline/processors/download.py 第 93-113 行的内容获取逻辑
+  found: content_file 只在 content_type in ["article", "answer", "zhuanlan"] 时被设置，video 类型被忽略
+  implication: **这是根本原因** - video 类型没有下载视频文件，导致 ASR Whisper 无法执行
+
+- timestamp: 2026-03-15T13:05:00Z
+  checked: kb-bilibili-content/bilibili_content.py 的 download_video 方法
+  found: 已有完整的 yt-dlp 视频下载实现，支持多种画质选择
+  implication: 可以复用此逻辑，但只需要下载音频（Whisper 只需要音频）
+
+## Resolution
+
+root_cause: |
+  **核心问题：DownloadProcessor 对于 video 类型只获取元数据，不下载视频文件**
+
+  在 kb_pipeline/processors/download.py 第 93-113 行：
+  - 只有 content_type in ["article", "answer", "zhuanlan"] 时才调用 adapter.fetch_content()
+  - 对于 video 类型，content_file 保持为空字符串
+
+  结果：
+  - pipeline.py 第 206 行：`video_path = download_result.get("content_file", "")` 返回空
+  - ASRProcessor 的 Whisper 需要 video_path 执行语音识别
+  - 没有 video_path，Whisper 无法运行，只能依赖官方字幕
+  - 如果视频没有官方字幕，最终笔记内容为空
+
+fix: |
+  在 DownloadProcessor.process() 中为 video 类型添加视频下载功能：
+
+  1. 在 BilibiliAdapter 中添加 download_video() 方法，使用 yt-dlp 下载音频
+  2. 在 DownloadProcessor.process() 中，对于 bilibili video 类型调用 adapter.download_video()
+  3. 将下载的文件路径设置到 content_file，供 ASR Whisper 使用
+
+  修改的文件：
+  - kb_pipeline/config.py: 添加 bilibili_cookie 配置项
+  - kb_pipeline/platforms/bilibili.py: 添加 download_video() 方法
+  - kb_pipeline/processors/download.py: 调用视频下载逻辑
+  - kb_pipeline/processors/asr.py: 修复缩进错误
+verification: |
+  ✅ 已通过端到端测试验证
+  - 视频下载成功 (m4a 格式)
+  - Whisper 语音识别成功生成字幕
+  - 分析步骤正确读取字幕内容
+  - 生成的分析报告包含真实视频内容（具体数据如 CPI、PPI、M1 等）
+files_changed:
+  - kb_pipeline/config.py
+  - kb_pipeline/platforms/bilibili.py
+  - kb_pipeline/processors/download.py
+  - kb_pipeline/processors/asr.py
+commits:
+  - eae0817: fix(asr): use content_dir for subtitle output
+  - 729f465: fix(pipeline): populate results for skipped steps
+  - 60b0eae: feat(download): add video download capability

.planning/debug/zhihu-article-images-missing.md

@@ -0,0 +1,68 @@
+---
+status: awaiting_human_verify
+trigger: "知乎文章处理时图片链接丢失"
+created: 2026-03-15T00:00:00Z
+updated: 2026-03-15T00:03:00Z
+---
+
+## Current Focus
+
+hypothesis: ROOT CAUSE CONFIRMED - 知乎懒加载图片的 src 属性是 SVG 占位符，真实 URL 在 data-actualsrc 属性中
+test: 已验证 markdownify 只读取 src 属性，得到无效的 SVG 占位符
+expecting: 修改 markdown_converter.py 处理知乎懒加载图片
+next_action: 修复 markdown_converter.py，提取真实图片 URL
+
+## Resolution
+
+root_cause: 知乎使用图片懒加载机制，`<img>` 标签的 `src` 属性包含 SVG 占位符（`data:image/svg+xml;...`），真实图片 URL 存储在 `data-actualsrc` 属性中。`markdownify` 库默认只读取 `src` 属性，导致转换后的 Markdown 包含无效的 SVG 占位符而不是真实图片链接。
+fix: 修改 `markdown_converter.py`，使用 BeautifulSoup 在转换前预处理 HTML，将 `data-actualsrc` 或 `data-original` 属性的值替换到 `src` 属性中
+verification: 单元测试全部通过（25/25），包括新的懒加载图片处理测试
+files_changed: [kb-zhihu-metadata/markdown_converter.py, tests/test_zhihu_metadata/test_markdown_converter.py]
+
+## Symptoms
+
+expected: 当分析知乎文章时，原文部分的图片链接应该保留在 markdown 中，用户可以在 markdown 中看到图片链接
+actual: 文字内容正确，图片无法看到，最后归档的分析内容中完全没有图片或者图片链接
+errors: 无显式错误，命令执行成功但图片丢失
+reproduction: 运行一键 CLI 命令处理知乎链接：https://www.zhihu.com/question/1936375725931361485/answer/2003232153480226496
+started: 第一次出现，可能是功能从未正确实现
+
+## Eliminated
+
+<!-- APPEND only -->
+
+## Evidence
+
+- timestamp: 2026-03-15T00:00:00Z
+  checked: markdown_converter.py 的图片转换
+  found: markdownify 正确转换 `<img>` 标签为 Markdown 格式 `![](url)`
+  implication: Markdown 转换器工作正常，问题不在转换层
+
+- timestamp: 2026-03-15T00:00:00Z
+  checked: zhihu_extractor.py 的 _extract_answer_content 方法
+  found: 使用 `.RichText` 选择器获取 HTML，然后调用 markdown_converter.convert()
+  implication: 提取逻辑看起来正确，需要验证选择器是否正确获取到包含图片的 HTML
+
+- timestamp: 2026-03-15T00:00:00Z
+  checked: zhihu_article_crawler.py（MediaCrawler 版本）
+  found: 第139行使用 `content_text` 字段，这是纯文本，不包含 HTML
+  implication: 这是另一个路径（MediaCrawler），但 kb_pipeline 使用的是 ZhihuPlaywrightExtractor
+
+- timestamp: 2026-03-15T00:01:00Z
+  checked: 知乎懒加载图片机制
+  found: 知乎图片的 src 属性是 SVG 占位符（data:image/svg+xml;...），真实 URL 在 data-actualsrc 或 data-original 属性中
+  implication: markdownify 只读取 src 属性，得到无效占位符而非真实图片链接
+
+- timestamp: 2026-03-15T00:01:00Z
+  checked: 模拟知乎懒加载 HTML 转换测试
+  found: 确认转换结果包含 `data:image/svg+xml` 占位符而不是真实图片 URL
+  implication: ROOT CAUSE CONFIRMED - 需要预处理 HTML 提取真实图片 URL
+
+## Resolution (问题2: 原文丢失)
+
+root_cause: 知乎回答 (answer) 和专栏 (zhuanlan) 的原文没有被追加到分析报告中。代码只在 `content_type == "article"` 时设置和追加原文，但知乎回答的类型是 `"answer"`
+fix:
+  - `deep_analyzer.py:184` - 扩展条件为 `content_type in ("article", "answer", "zhuanlan")`
+  - `analyze.py:244` - 扩展条件为 `content_type in ("article", "answer", "zhuanlan")`
+verification: 待用户测试
+files_changed: [kb-deep-analyzer/deep_analyzer.py, kb_pipeline/processors/analyze.py]

bilibili_BV16cNMzpExh/bilibili_BV16cNMzpExh_metadata.json

@@ -0,0 +1,24 @@
+{
+  "metadata": {
+    "title": "2-3月宏观经济运行状态复盘与推演 + 2026年政府工作报告解读 | 【阿周的Eureka】",
+    "author": "阿周的Eureka",
+    "bvid": "BV16cNMzpExh",
+    "aid": 116192639519967,
+    "mid": 46405906,
+    "pubtime": "2026-03-08 16:44:51",
+    "duration": "07:16",
+    "url": "https://www.bilibili.com/video/BV16cNMzpExh",
+    "tags": ""
+  },
+  "stats": {
+    "view": 78484,
+    "danmaku": 112,
+    "reply": 97,
+    "favorite": 4352,
+    "coin": 1762,
+    "share": 309,
+    "like": 5248
+  },
+  "subtitle": "",
+  "description": "各位玩家好，本期视频是对2-3月份的宏观经济运行状态，结合前几天发布的政府工作报告做的数据复盘与推演\n希望能够用一个更直观的方式和各位一起了解每个月宏观经济的运行情况\n这个【复盘经济系统运行状态】的系列，之后会作为每月的固定更新"
+}

bilibili_BV16cNMzpExh/workflow_state.json

@@ -0,0 +1,34 @@
+{
+  "schema_version": "1.0",
+  "content_id": "BV16cNMzpExh",
+  "platform": "bilibili",
+  "source_url": "https://www.bilibili.com/video/BV16cNMzpExh",
+  "created_at": "2026-03-15T23:52:15.244309",
+  "updated_at": "2026-03-15T23:52:18.638677",
+  "steps": {
+    "parse": {
+      "step_name": "parse",
+      "status": "completed",
+      "started_at": "2026-03-15T23:52:15.244320",
+      "completed_at": "2026-03-15T23:52:15.244530",
+      "error_message": null,
+      "output_summary": null
+    },
+    "download": {
+      "step_name": "download",
+      "status": "completed",
+      "started_at": "2026-03-15T23:52:15.244627",
+      "completed_at": "2026-03-15T23:52:18.525523",
+      "error_message": null,
+      "output_summary": null
+    },
+    "asr": {
+      "step_name": "asr",
+      "status": "failed",
+      "started_at": "2026-03-15T23:52:18.525793",
+      "completed_at": "2026-03-15T23:52:18.638670",
+      "error_message": "video_path 为空，无法执行 Whisper 语音识别。请确保视频文件已下载（对于没有官方字幕的视频，需要先下载视频文件）。",
+      "output_summary": null
+    }
+  }
+}

bilibili_BV16cNMzpExh/workflow_state.json.bak

@@ -0,0 +1,34 @@
+{
+  "schema_version": "1.0",
+  "content_id": "BV16cNMzpExh",
+  "platform": "bilibili",
+  "source_url": "https://www.bilibili.com/video/BV16cNMzpExh",
+  "created_at": "2026-03-15T23:52:15.244309",
+  "updated_at": "2026-03-15T23:52:18.525804",
+  "steps": {
+    "parse": {
+      "step_name": "parse",
+      "status": "completed",
+      "started_at": "2026-03-15T23:52:15.244320",
+      "completed_at": "2026-03-15T23:52:15.244530",
+      "error_message": null,
+      "output_summary": null
+    },
+    "download": {
+      "step_name": "download",
+      "status": "completed",
+      "started_at": "2026-03-15T23:52:15.244627",
+      "completed_at": "2026-03-15T23:52:18.525523",
+      "error_message": null,
+      "output_summary": null
+    },
+    "asr": {
+      "step_name": "asr",
+      "status": "running",
+      "started_at": "2026-03-15T23:52:18.525793",
+      "completed_at": null,
+      "error_message": null,
+      "output_summary": null
+    }
+  }
+}

kb-deep-analyzer/deep_analyzer.py

@@ -180,8 +180,8 @@ class DeepAnalyzer:
         # 解析结果
         result = self._parse_analysis(full_analysis, text, duration, content_type)
         
-        # 文章类型：追加原文
-        if content_type == "article":
+        # 文章/回答类型：追加原文
+        if content_type in ("article", "answer", "zhuanlan"):
             result.original_text = text
         
         return result

kb_deep_analyzer

@@ -0,0 +1 @@
+kb-deep-analyzer

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/pipeline.py

@@ -153,7 +153,25 @@ class Pipeline:
 
             # Check if step should be skipped
             if self.config.skip_completed and state.is_step_completed(step_name):
-                logger.info(f"Skipping completed step: {step_name}")
+                # Ask user whether to re-run the completed step
+                should_rerun = self._ask_rerun_step(step_name, content_dir)
+
+                if should_rerun:
+                    logger.info(f"Re-running completed step: {step_name}")
+                    # Clear the step's state record so it can be re-executed
+                    if step_name in state.steps:
+                        del state.steps[step_name]
+                        state_manager.save(state)
+                else:
+                    logger.info(f"Skipping completed step: {step_name}")
+                    # Scan existing files and populate results for skipped steps
+                    self._populate_skipped_step_results(step_name, results, content_dir, parsed)
+                    continue
+
+            # Skip ASR for text-based content types
+            text_content_types = {"article", "answer", "zhuanlan", "question"}
+            if step_name == "asr" and parsed.content_type in text_content_types:
+                logger.info(f"Skipping ASR for text content type: {parsed.content_type}")
                 continue
 
             processor = self.processors[step_name]
@@ -180,6 +198,7 @@ class Pipeline:
                         platform=parsed.platform,
                         content_dir=str(content_dir),
                         content_type=parsed.content_type,
+                        source_url=url,
                     )
                 elif step_name == "asr":
                     # ASR needs video path from download step
@@ -266,3 +285,102 @@ class Pipeline:
             results=results,
             content_dir=content_dir,
         )
+
+    def _ask_rerun_step(self, step_name: str, content_dir: Path) -> bool:
+        """Ask user whether to re-run a completed step.
+
+        Args:
+            step_name: Name of the step to potentially re-run
+            content_dir: Content directory path
+
+        Returns:
+            True if user wants to re-run, False to skip
+        """
+        try:
+            from rich.console import Console
+            from rich.prompt import Confirm
+
+            console = Console()
+            return Confirm.ask(
+                f"Step '{step_name}' is already completed. Re-run it?",
+                default=False
+            )
+        except ImportError:
+            # Fallback to basic input if rich is not available
+            response = input(f"Step '{step_name}' is already completed. Re-run it? [y/N] ")
+            return response.lower() in ('y', 'yes')
+
+    def _populate_skipped_step_results(
+        self,
+        step_name: str,
+        results: Dict,
+        content_dir: Path,
+        parsed: ParsedLink
+    ) -> None:
+        """Populate results for skipped steps by scanning existing files.
+
+        This ensures downstream steps can find the outputs they need,
+        even when a step is skipped.
+
+        Args:
+            step_name: Name of the skipped step
+            results: Results dictionary to populate
+            content_dir: Content directory path
+            parsed: Parsed link with platform and content_id
+        """
+        if step_name == "asr":
+            # Scan for subtitle files generated by ASR
+            # Try both new naming convention (with _subtitle suffix) and old convention (without suffix)
+            base_name_new = f"{parsed.platform}_{parsed.content_id}_subtitle"
+            base_name_old = f"{parsed.platform}_{parsed.content_id}"
+
+            json_path = content_dir / f"{base_name_new}.json"
+            if not json_path.exists():
+                json_path = content_dir / f"{base_name_old}.json"
+
+            srt_path = content_dir / f"{base_name_new}.srt"
+            if not srt_path.exists():
+                srt_path = content_dir / f"{base_name_old}.srt"
+
+            output_files = {}
+            if json_path.exists():
+                output_files["json"] = str(json_path)
+            if srt_path.exists():
+                output_files["srt"] = str(srt_path)
+
+            if output_files:
+                results["asr"] = {
+                    "success": True,
+                    "error_message": None,
+                    "output_files": output_files,
+                }
+                logger.info(f"Found existing ASR outputs: {output_files}")
+            else:
+                logger.warning(f"No existing ASR outputs found in {content_dir}")
+                results["asr"] = {
+                    "success": False,
+                    "error_message": "No existing subtitle files found",
+                    "output_files": {},
+                }
+
+        elif step_name == "download":
+            # Scan for metadata file
+            metadata_path = content_dir / "metadata.json"
+            if metadata_path.exists():
+                results["download"] = {
+                    "success": True,
+                    "error_message": None,
+                    "metadata_file": str(metadata_path),
+                }
+                logger.info(f"Found existing download metadata: {metadata_path}")
+
+        elif step_name == "analyze":
+            # Scan for analysis output
+            analysis_path = content_dir / "analysis.md"
+            if analysis_path.exists():
+                results["analyze"] = {
+                    "success": True,
+                    "error_message": None,
+                    "analysis_file": str(analysis_path),
+                }
+                logger.info(f"Found existing analysis: {analysis_path}")

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
-            url = f"https://www.zhihu.com/question/0/answer/{content_id}"
+            # 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
-        result = self._extract_with_playwright(url)
+        # 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):
@@ -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/analyze.py

@@ -81,6 +81,15 @@ class AnalyzeProcessor(BaseProcessor):
             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)
@@ -240,8 +249,8 @@ class AnalyzeProcessor(BaseProcessor):
         # 获取完整分析内容
         full_analysis = result.full_analysis
 
-        # 文章类型：追加原文
-        if content_type == "article" and result.original_text:
+        # 文章/回答类型：追加原文
+        if content_type in ("article", "answer", "zhuanlan") and result.original_text:
             full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text
 
         # 写入文件

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站）
-            if input_data.platform == "bilibili":
-                result = await self._try_official_subtitle(input_data)
+            # 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 2: 无官方字幕则调用 Whisper
+                if result is None:
+                    result = await self._run_whisper(input_data)
 
             # Step 3: 记录成功状态
             if self.state_manager:
@@ -160,6 +167,92 @@ class ASRProcessor:
                     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站官方字幕
 
@@ -426,10 +519,19 @@ class ASRProcessor:
                 # 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': item.get('offsets', [0, 0])[0] / 1000.0,
-                            'end': item.get('offsets', [0, 0])[1] / 1000.0,
+                            'start': start,
+                            'end': end,
                             'text': item.get('text', '')
                         })
                 text = ' '.join(s['text'] for s in segments)

kb_pipeline/processors/download.py

@@ -62,11 +62,18 @@ class DownloadProcessor(BaseProcessor):
             return self._record_failure(str(e))
 
         try:
-            # Fetch metadata
-            metadata = await adapter.fetch_metadata(
-                input_data.content_id,
-                input_data.content_type
-            )
+            # 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(
@@ -85,10 +92,17 @@ class DownloadProcessor(BaseProcessor):
             # Fetch content (for text-based content)
             content_file = ""
             if input_data.content_type in ["article", "answer", "zhuanlan"]:
-                content = await adapter.fetch_content(
-                    input_data.content_id,
-                    input_data.content_type
-                )
+                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,
@@ -98,6 +112,18 @@ class DownloadProcessor(BaseProcessor):
                 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,

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

kb_zhihu_metadata

@@ -0,0 +1 @@
+kb-zhihu-metadata