diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md
index b8ff5f9..5fbf4e4 100644
--- a/.planning/ROADMAP.md
+++ b/.planning/ROADMAP.md
@@ -2,7 +2,7 @@
## Overview
-将现有分散的技能模块重构为统一的管道处理系统。从基础设施（状态管理、Schema验证）开始，逐步集成平台技能（B站/知乎）、处理技能（ASR/OCR），最终形成完整的端到端工作流。每个阶段都交付可独立验证的能力，确保长时任务可恢复、可追溯。
+将现有分散的技能模块重构为统一的管道处理系统。从基础设施（状态管理、Schema验证）开始，逐步集成平台技能（B站/知乎）、处理技能（ASR），最终形成完整的端到端工作流。每个阶段都交付可独立验证的能力，确保长时任务可恢复、可追溯。
## Phases
@@ -14,7 +14,7 @@ Decimal phases appear between their surrounding integers in numeric order.
- [x] **Phase 1: Foundation** - 状态管理、Schema验证、错误处理、统一配置
- [x] **Phase 2: Platform Skills** - B站/知乎链接解析、元数据提取、内容下载
-- [ ] **Phase 3: Processing Skills** - ASR/OCR集成、多模态内容处理
+- [ ] **Phase 3: Processing Skills** - ASR 集成、视频内容文字提取
- [ ] **Phase 4: Analysis & Orchestration** - 深度分析集成、完整管道编排、端到端验证
## Phase Details
@@ -54,20 +54,18 @@ Plans:
- [x] 02-03-PLAN.md — Zhihu Adapter: 问答 + 专栏内容 (PLAT-03)
### Phase 3: Processing Skills
-**Goal**: 集成ASR和OCR处理能力，支持视频和图片的多模态内容提取
+**Goal**: 集成 ASR 处理能力，支持视频内容的文字提取（官方字幕优先，无字幕时使用 Whisper.cpp）
**Depends on**: Phase 2
**Requirements**: None (平台支持已在Phase 2覆盖)
**Success Criteria** (what must be TRUE):
1. B站视频自动下载字幕（优先官方字幕），无字幕时调用语音识别
2. Whisper识别完成后，转录结果保存为结构化JSON（含时间戳）
- 3. 视频帧自动提取并OCR识别，文字结果与时间戳关联
- 4. 处理结果自动关联到对应的内容目录
- 5. HF_TOKEN配置时自动启用带说话者分离的Whisper
-**Plans**: TBD
+ 3. 处理结果自动关联到对应的内容目录
+ 4. HF_TOKEN配置时自动启用带说话者分离的Whisper
+**Plans:** 1 plan
Plans:
-- [ ] 03-01: B站字幕下载 & Whisper/FunASR 集成
-- [ ] 03-02: OCR 处理集成
+- [ ] 03-01-PLAN.md — ASRProcessor: 官方字幕下载 + Whisper.cpp 集成 + 说话者分离
### Phase 4: Analysis & Orchestration
**Goal**: 集成深度分析能力，完成端到端工作流编排，实现从链接到分析报告的全自动处理
@@ -95,10 +93,10 @@ Phases execute in numeric order: 1 → 2 → 3 → 4
|-------|----------------|--------|-----------|
| 1. Foundation | 3/3 | Complete | 01-02, 01-01, 01-03 |
| 2. Platform Skills | 3/3 | Complete | 02-01, 02-02, 02-03 |
-| 3. Processing Skills | 0/2 | Not started | - |
+| 3. Processing Skills | 0/1 | Not started | - |
| 4. Analysis & Orchestration | 0/3 | Not started | - |
---
*Roadmap created: 2026-03-13*
-*Last updated: 2026-03-13 - Phase 2 complete (3/3 plans)*
+*Last updated: 2026-03-13 - Phase 3 planned (OCR excluded per user decision)*
*Granularity: coarse (4 phases)*
diff --git a/.planning/phases/03-processing-skills/03-01-PLAN.md b/.planning/phases/03-processing-skills/03-01-PLAN.md
new file mode 100644
index 0000000..dd4638c
--- /dev/null
+++ b/.planning/phases/03-processing-skills/03-01-PLAN.md
@@ -0,0 +1,612 @@
+---
+phase: 03-processing-skills
+plan: 01
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+ - kb_pipeline/processors/__init__.py
+ - kb_pipeline/processors/asr.py
+ - kb_pipeline/schemas.py
+ - tests/test_processors/__init__.py
+ - tests/test_processors/test_asr.py
+autonomous: true
+requirements: [] # 平台支持已在Phase 2覆盖，ASR是内部处理能力
+user_setup:
+ - service: whisper.cpp
+ why: "本地语音识别引擎"
+ env_vars: []
+ dashboard_config:
+ - task: "安装 whisper.cpp 和下载模型"
+ location: "git clone https://github.com/ggml-org/whisper.cpp.git ~/whisper.cpp && cd ~/whisper.cpp && make && ./models/download-ggml-model.sh base"
+ - service: pyannote.audio (可选)
+ why: "说话者分离功能"
+ env_vars:
+ - name: HF_TOKEN
+ source: "HuggingFace Settings -> Access Tokens"
+ dashboard_config:
+ - task: "接受 pyannote 模型用户协议"
+ location: "https://huggingface.co/pyannote/speaker-diarization-3.1"
+
+must_haves:
+ truths:
+ - "B站视频可以自动下载官方字幕"
+ - "无官方字幕时自动调用 Whisper.cpp 进行语音识别"
+ - "HF_TOKEN 配置时自动启用说话者分离"
+ - "输出包含 txt/srt/json 三种格式"
+ - "ASR 完成后状态记录到 workflow_state.json"
+ - "临时音频文件处理后自动清理"
+ artifacts:
+ - path: "kb_pipeline/processors/__init__.py"
+ provides: "处理器模块初始化"
+ exports: ["ASRProcessor"]
+ - path: "kb_pipeline/processors/asr.py"
+ provides: "ASR 处理器实现"
+ contains: "class ASRProcessor"
+ min_lines: 150
+ - path: "kb_pipeline/schemas.py"
+ provides: "ASR 输入输出 Schema"
+ contains: "class ASRInput"
+ contains: "class ASROutput"
+ key_links:
+ - from: "kb_pipeline/processors/asr.py"
+ to: "kb_pipeline/schemas.py"
+ via: "from ..schemas import ASRInput, ASROutput"
+ pattern: "ASRInput|ASROutput"
+ - from: "kb_pipeline/processors/asr.py"
+ to: "kb_pipeline/state.py"
+ via: "WorkflowStateManager"
+ pattern: "state_manager\\.mark_step"
+ - from: "kb_pipeline/processors/asr.py"
+ to: "kb-bilibili-content/bilibili_content.py"
+ via: "download_subtitle 逻辑复用"
+ pattern: "download_subtitle|has_subtitle"
+ - from: "kb_pipeline/processors/asr.py"
+ to: "kb-whisper/whisper.py"
+ via: "Whisper.cpp 调用逻辑复用"
+ pattern: "whisper-cli|convert_to_wav"
+---
+
+
+创建 ASRProcessor 处理器，统一封装 B站官方字幕下载和 Whisper.cpp 语音识别，遵循 Phase 1/2 建立的管道模式。
+
+Purpose: 为视频内容提供统一的文字提取入口，优先使用高质量官方字幕，无字幕时自动降级到本地语音识别。
+Output: ASRProcessor 类，支持官方字幕下载、Whisper.cpp 调用、说话者分离、状态记录。
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+@.planning/phases/03-processing-skills/03-CONTEXT.md
+@.planning/phases/03-processing-skills/03-RESEARCH.md
+
+## 现有代码模式
+
+### Phase 1 Schema 体系 (kb_pipeline/schemas.py)
+```python
+class StepInput(BaseModel):
+ content_id: str
+ platform: Literal["bilibili", "zhihu"]
+
+class StepOutput(BaseModel):
+ success: bool
+ error_message: Optional[str] = None
+ completed_at: datetime = Field(default_factory=datetime.now)
+
+class WorkflowState(BaseModel):
+ def mark_step_completed(self, step_name: str, output_summary: Optional[str] = None) -> None
+ def mark_step_failed(self, step_name: str, error_message: str) -> None
+```
+
+### Phase 2 平台适配器模式 (kb_pipeline/platforms/bilibili.py)
+```python
+class BilibiliAdapter(PlatformAdapter):
+ async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]
+ async def fetch_content(self, content_id: str, content_type: str) -> str
+```
+
+### 现有 ASR 工具 (kb-whisper/whisper.py)
+- `convert_to_wav()`: 音频转换逻辑
+- `run_whisper()`: Whisper.cpp 调用
+- `create_workflow_json()`: 标准输出格式
+
+### 现有字幕下载 (kb-bilibili-content/bilibili_content.py)
+- `download_subtitle()`: B站官方字幕下载和 SRT 转换
+- 返回格式：`{success, has_subtitle, filepath, message}`
+
+
+
+
+
+
+From kb_pipeline/schemas.py:
+```python
+class StepInput(BaseModel):
+ content_id: str = Field(..., min_length=1)
+ platform: Literal["bilibili", "zhihu"] = Field(...)
+
+class StepOutput(BaseModel):
+ success: bool = Field(...)
+ error_message: Optional[str] = Field(None)
+ completed_at: datetime = Field(default_factory=datetime.now)
+
+class WorkflowState(BaseModel):
+ def mark_step_completed(self, step_name: str, output_summary: Optional[str] = None) -> None
+ def mark_step_failed(self, step_name: str, error_message: str) -> None
+```
+
+From kb_pipeline/state.py:
+```python
+class WorkflowStateManager:
+ STATE_FILENAME = "workflow_state.json"
+ def load(self) -> Optional[WorkflowState]
+ def save(self, state: WorkflowState) -> None
+ def get_or_create(self, content_id: str, platform: str, source_url: str) -> WorkflowState
+```
+
+From kb_pipeline/config.py:
+```python
+class KBSettings(BaseSettings):
+ hf_token: SecretStr = Field(default=SecretStr(""))
+ # 其他配置...
+```
+
+From kb_pipeline/retry.py:
+```python
+retry_on_network_error = retry(
+ stop=stop_after_attempt(3),
+ wait=wait_exponential(multiplier=1, min=2, max=10),
+ retry=retry_if_exception_type(NetworkError),
+)
+```
+
+
+
+
+
+ Task 1: Define ASR Schemas
+ kb_pipeline/schemas.py, tests/test_processors/test_asr.py
+
+ - Test 1: ASRInput 继承 StepInput，包含 video_path 和 content_type 字段
+ - Test 2: ASROutput 继承 StepOutput，包含 tool/has_subtitle/output_files/segments/text/duration_seconds 字段
+ - Test 3: ASROutput 的 tool 字段只能是 "subtitle" | "whisper" | "whisper-diarization"
+
+
+在 kb_pipeline/schemas.py 中添加 ASR 专用 Schema：
+
+1. ASRInput 类（继承 StepInput）：
+ - video_path: str - 视频文件路径
+ - content_type: str = "video" - 内容类型
+
+2. ASROutput 类（继承 StepOutput）：
+ - tool: Literal["subtitle", "whisper", "whisper-diarization"] - 使用的工具
+ - has_subtitle: bool - 是否有官方字幕
+ - output_files: Dict[str, str] - 输出文件路径 {txt, srt, json}
+ - segments: List[Dict] - 带时间戳的分段列表
+ - text: str - 完整文本
+ - duration_seconds: float - 音频时长
+ - speakers: Optional[List[str]] - 说话者列表（仅说话者分离时）
+
+3. 添加必要的 import（List, Dict, Literal, Optional）
+
+确保 Schema 符合 Phase 1 的 StepInput/StepOutput 继承体系。
+
+
+ pytest tests/test_processors/test_asr.py::test_asr_input_schema -x -v
+ pytest tests/test_processors/test_asr.py::test_asr_output_schema -x -v
+
+
+ - ASRInput 类定义完整，继承 StepInput
+ - ASROutput 类定义完整，继承 StepOutput
+ - 两个测试通过
+
+
+
+
+ Task 2: Create ASRProcessor with subtitle download
+ kb_pipeline/processors/__init__.py, kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py
+
+ - Test 1: ASRProcessor 初始化成功
+ - Test 2: _try_official_subtitle 调用成功时返回 ASROutput (tool="subtitle")
+ - Test 3: _try_official_subtitle 无字幕时返回 None
+ - Test 4: process() 方法优先尝试官方字幕
+
+
+创建 kb_pipeline/processors/ 目录和 ASRProcessor 类：
+
+1. 创建 kb_pipeline/processors/__init__.py：
+ ```python
+ from .asr import ASRProcessor
+ __all__ = ["ASRProcessor"]
+ ```
+
+2. 创建 kb_pipeline/processors/asr.py：
+
+ ASRProcessor 类结构：
+ ```python
+ class ASRProcessor:
+ STEP_NAME = "asr"
+
+ def __init__(self, state_manager: Optional[WorkflowStateManager] = None):
+ self.settings = get_settings()
+ self.state_manager = state_manager
+
+ async def process(self, input_data: ASRInput) -> ASROutput:
+ """主入口：先尝试官方字幕，无则调用 Whisper"""
+ # 1. 如果是 B站视频，尝试官方字幕
+ # 2. 无官方字幕则调用 Whisper
+ # 3. 记录状态
+
+ async def _try_official_subtitle(self, input_data: ASRInput) -> Optional[ASROutput]:
+ """尝试下载 B站官方字幕"""
+ # 复用 kb-bilibili-content/bilibili_content.py 的 download_subtitle 逻辑
+ # 返回 None 表示无字幕（不是错误）
+ ```
+
+3. _try_official_subtitle 实现：
+ - 使用 aiohttp 调用 B站 API 获取字幕列表
+ - API: https://api.bilibili.com/x/player/pagelist?bvid={bvid}
+ - API: https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}
+ - 下载字幕 JSON 并转换为 SRT 格式
+ - 同时生成 txt 和 json 输出
+ - 参考 kb-bilibili-content/bilibili_content.py 的 download_subtitle 方法
+
+4. 关键点：
+ - 无字幕返回 None（不是抛异常），让流程继续到 Whisper
+ - 有字幕时返回 ASROutput，tool="subtitle"
+ - 使用 @retry_on_network_error 装饰器处理网络错误
+
+不要实现 Whisper 调用部分（Task 3）。
+
+
+ pytest tests/test_processors/test_asr.py::test_asr_processor_init -x -v
+ pytest tests/test_processors/test_asr.py::test_official_subtitle_found -x -v
+ pytest tests/test_processors/test_asr.py::test_official_subtitle_not_found -x -v
+
+
+ - ASRProcessor 类创建完成
+ - _try_official_subtitle 方法实现
+ - 三个测试通过
+
+
+
+
+ Task 3: Implement Whisper.cpp integration
+ kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py
+
+ - Test 1: _run_whisper 调用成功返回 ASROutput (tool="whisper")
+ - Test 2: 临时音频文件处理后自动清理
+ - Test 3: Whisper 路径不存在时抛出 RuntimeError
+
+
+在 ASRProcessor 中添加 Whisper.cpp 调用逻辑：
+
+1. _run_whisper 方法：
+ ```python
+ async def _run_whisper(self, input_data: ASRInput) -> ASROutput:
+ """执行 Whisper.cpp 语音识别"""
+ # 1. 检查 HF_TOKEN 决定是否启用说话者分离
+ # 2. 转换音频格式
+ # 3. 调用 whisper-cli
+ # 4. 清理临时文件
+ ```
+
+2. 复用 kb-whisper/whisper.py 的核心逻辑：
+ - get_whisper_cpp_path(): 获取 whisper.cpp 路径
+ - convert_to_wav(): 音频转换（ffmpeg 16kHz WAV）
+ - run_whisper(): 调用 whisper-cli 命令
+
+3. 命令构建：
+ ```python
+ cmd = [
+ os.path.join(whisper_path, 'whisper-cli'),
+ '-m', os.path.join(whisper_path, 'models', 'ggml-base.bin'),
+ '-f', audio_path,
+ '-of', output_file,
+ '-l', 'zh',
+ '--output-txt', '--output-srt', '--output-json',
+ ]
+ ```
+
+4. 临时文件清理：
+ - 处理完成后删除 *_audio.wav 文件
+ - 使用 try/except 静默处理删除失败
+
+5. HF_TOKEN 检查：
+ ```python
+ 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)
+ ```
+
+6. 错误处理：
+ - whisper.cpp 路径不存在 → RuntimeError
+ - 模型不存在 → FileNotFoundError
+ - whisper-cli 执行失败 → RuntimeError
+
+先实现 _run_whisper_basic，_run_whisper_with_diarization 在 Task 4 实现。
+
+
+ pytest tests/test_processors/test_asr.py::test_whisper_basic -x -v
+ pytest tests/test_processors/test_asr.py::test_temp_file_cleanup -x -v
+ pytest tests/test_processors/test_asr.py::test_whisper_path_not_found -x -v
+
+
+ - _run_whisper 方法实现完成
+ - _run_whisper_basic 实现完成
+ - 临时文件清理逻辑正确
+ - 三个测试通过
+
+
+
+
+ Task 4: Implement speaker diarization
+ kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py
+
+ - Test 1: HF_TOKEN 未配置时使用普通 Whisper
+ - Test 2: HF_TOKEN 配置时自动启用说话者分离
+ - Test 3: 说话者分离结果包含 speaker 字段
+
+
+在 ASRProcessor 中添加说话者分离支持：
+
+1. _run_whisper_with_diarization 方法：
+ ```python
+ async def _run_whisper_with_diarization(self, input_data: ASRInput) -> ASROutput:
+ """执行带说话者分离的 Whisper"""
+ # 1. 先运行普通 Whisper
+ # 2. 运行 pyannote.audio 说话者分离
+ # 3. 合并结果
+ ```
+
+2. 复用 kb-whisper-diarization/whisper_diarization.py 的逻辑：
+ - run_diarization(): pyannote.audio 调用
+ - assign_speaker(): 为段落分配说话者
+ - merge_results(): 合并 Whisper 和说话者分离结果
+
+3. pyannote.audio 调用：
+ ```python
+ from huggingface_hub import login
+ from pyannote.audio import Pipeline
+
+ login(token=hf_token)
+ pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1")
+ diarization = pipeline(audio_path)
+ ```
+
+4. 合并逻辑：
+ - 遍历 Whisper 的每个 segment
+ - 根据 start/end 时间匹配说话者
+ - 在 segment 中添加 speaker 字段
+
+5. 输出格式：
+ - tool: "whisper-diarization"
+ - speakers: ["SPEAKER_00", "SPEAKER_01", ...]
+ - 每个 segment 包含 speaker 字段
+
+6. 依赖处理：
+ - 使用 try/except 导入 pyannote.audio
+ - 导入失败时回退到普通 Whisper
+
+
+ pytest tests/test_processors/test_asr.py::test_no_hf_token_uses_basic_whisper -x -v
+ pytest tests/test_processors/test_asr.py::test_diarization_with_hf_token -x -v
+ pytest tests/test_processors/test_asr.py::test_diarization_result_has_speaker -x -v
+
+
+ - _run_whisper_with_diarization 方法实现
+ - HF_TOKEN 自动检测逻辑正确
+ - 说话者分离结果格式正确
+ - 三个测试通过
+
+
+
+
+ Task 5: Add state recording
+ kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py
+
+ - Test 1: ASR 成功时状态记录为 completed
+ - Test 2: ASR 失败时状态记录为 failed
+ - Test 3: output_summary 包含 tool 和 segments 数量
+
+
+在 ASRProcessor.process() 方法中添加状态记录：
+
+1. 成功时记录：
+ ```python
+ if self.state_manager:
+ state = self.state_manager.load()
+ if state:
+ state.mark_step_completed(
+ self.STEP_NAME,
+ output_summary=f"tool={result.tool}, segments={len(result.segments)}"
+ )
+ self.state_manager.save(state)
+ ```
+
+2. 失败时记录：
+ ```python
+ 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
+ ```
+
+3. 状态记录时机：
+ - 只在所有重试都失败后记录失败（不是每次重试）
+ - 成功后立即记录
+
+4. output_summary 格式：
+ - "tool=subtitle, segments=120"
+ - "tool=whisper, segments=85"
+ - "tool=whisper-diarization, segments=85, speakers=2"
+
+5. 确保 state_manager 是可选的（None 时不记录状态）
+
+
+ pytest tests/test_processors/test_asr.py::test_state_recording_success -x -v
+ pytest tests/test_processors/test_asr.py::test_state_recording_failure -x -v
+ pytest tests/test_processors/test_asr.py::test_state_no_manager -x -v
+
+
+ - 状态记录逻辑正确
+ - output_summary 格式正确
+ - state_manager 可选处理正确
+ - 三个测试通过
+
+
+
+
+ Task 6: Create test infrastructure
+ tests/test_processors/__init__.py, tests/test_processors/conftest.py, tests/test_processors/test_asr.py
+
+创建测试基础设施：
+
+1. 创建 tests/test_processors/__init__.py（空文件）
+
+2. 创建 tests/test_processors/conftest.py：
+ ```python
+ import pytest
+ from pathlib import Path
+ from unittest.mock import AsyncMock, MagicMock, patch
+
+ @pytest.fixture
+ def mock_whisper_path(tmp_path):
+ """Mock whisper.cpp 路径"""
+ whisper_dir = tmp_path / "whisper.cpp"
+ whisper_dir.mkdir()
+ (whisper_dir / "whisper-cli").write_text("")
+ (whisper_dir / "models").mkdir()
+ (whisper_dir / "models" / "ggml-base.bin").write_text("")
+ return str(whisper_dir)
+
+ @pytest.fixture
+ def mock_bilibili_api():
+ """Mock B站 API 响应"""
+ with patch("aiohttp.ClientSession.get") as mock_get:
+ yield mock_get
+
+ @pytest.fixture
+ def sample_video(tmp_path):
+ """创建测试视频文件"""
+ video_path = tmp_path / "test_video.mp4"
+ video_path.write_text("fake video content")
+ return str(video_path)
+
+ @pytest.fixture
+ def sample_subtitle_response():
+ """B站字幕 API 响应示例"""
+ return {
+ "data": {
+ "subtitle": {
+ "list": [{
+ "lan": "ai-zh",
+ "lan_doc": "中文（自动生成）",
+ "subtitle_url": "//example.com/subtitle.json"
+ }]
+ }
+ }
+ }
+ ```
+
+3. 创建 tests/test_processors/test_asr.py 骨架：
+ - 包含所有测试用例的占位符
+ - 每个测试使用 pytest.raises 或 assert 验证
+
+4. 测试文件结构：
+ ```python
+ # Schema tests
+ def test_asr_input_schema(): ...
+ def test_asr_output_schema(): ...
+
+ # Processor tests
+ def test_asr_processor_init(): ...
+ def test_official_subtitle_found(): ...
+ def test_official_subtitle_not_found(): ...
+ def test_whisper_basic(): ...
+ def test_temp_file_cleanup(): ...
+ def test_whisper_path_not_found(): ...
+ def test_no_hf_token_uses_basic_whisper(): ...
+ def test_diarization_with_hf_token(): ...
+ def test_diarization_result_has_speaker(): ...
+ def test_state_recording_success(): ...
+ def test_state_recording_failure(): ...
+ def test_state_no_manager(): ...
+ ```
+
+使用 unittest.mock 来模拟外部依赖（aiohttp、subprocess、whisper.cpp）。
+
+
+ pytest tests/test_processors/test_asr.py --collect-only | grep "test_" | wc -l
+
+
+ - tests/test_processors/ 目录创建完成
+ - conftest.py 包含所有 fixtures
+ - test_asr.py 包含所有测试用例骨架
+ - 测试数量 >= 13 个
+
+
+
+
+
+
+## Phase Verification
+
+完成所有任务后，验证以下条件：
+
+1. **Schema 验证**
+ ```bash
+ pytest tests/test_processors/test_asr.py::test_asr_input_schema tests/test_processors/test_asr.py::test_asr_output_schema -v
+ ```
+
+2. **官方字幕下载**
+ ```bash
+ pytest tests/test_processors/test_asr.py::test_official_subtitle_found tests/test_processors/test_asr.py::test_official_subtitle_not_found -v
+ ```
+
+3. **Whisper 集成**
+ ```bash
+ pytest tests/test_processors/test_asr.py::test_whisper_basic tests/test_processors/test_asr.py::test_temp_file_cleanup -v
+ ```
+
+4. **说话者分离**
+ ```bash
+ pytest tests/test_processors/test_asr.py::test_diarization_with_hf_token tests/test_processors/test_asr.py::test_diarization_result_has_speaker -v
+ ```
+
+5. **状态记录**
+ ```bash
+ pytest tests/test_processors/test_asr.py::test_state_recording_success tests/test_processors/test_asr.py::test_state_recording_failure -v
+ ```
+
+6. **完整测试套件**
+ ```bash
+ pytest tests/test_processors/test_asr.py -v
+ ```
+
+
+
+1. ASRProcessor 类可被实例化和调用
+2. B站视频可以自动下载官方字幕（优先策略）
+3. 无官方字幕时自动调用 Whisper.cpp 进行语音识别
+4. HF_TOKEN 配置时自动启用说话者分离
+5. 输出包含 txt/srt/json 三种格式
+6. ASR 完成后状态正确记录到 workflow_state.json
+7. 临时音频文件处理后自动清理
+8. 所有测试通过（>= 13 个测试）
+
+
+