starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(03): create ASR processing plan

Browse Source
This commit is contained in:
starmen1265
2026-03-13 19:11:29 +08:00
parent eaa68acb9d
commit 2039a45bf0
2 changed files with 621 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.planning/ROADMAP.md
+9 -11
View File
@@ -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)*
.planning/phases/03-processing-skills/03-01-PLAN.md
+612
View File
@@ -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"
---
<objective>
创建 ASRProcessor 处理器，统一封装 B站官方字幕下载和 Whisper.cpp 语音识别，遵循 Phase 1/2 建立的管道模式。
Purpose: 为视频内容提供统一的文字提取入口，优先使用高质量官方字幕，无字幕时自动降级到本地语音识别。
Output: ASRProcessor 类，支持官方字幕下载、Whisper.cpp 调用、说话者分离、状态记录。
</objective>
<execution_context>
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
</execution_context>
<context>
@.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}`
</context>
<interfaces>
<!-- Key types and contracts the executor needs -->
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),
)
```
</interfaces>
<tasks>
<task type="auto" tdd="true">
<name>Task 1: Define ASR Schemas</name>
<files>kb_pipeline/schemas.py, tests/test_processors/test_asr.py</files>
<behavior>
- 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"
</behavior>
<action>
在 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 继承体系。
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py::test_asr_input_schema -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_asr_output_schema -x -v</automated>
</verify>
<done>
- ASRInput 类定义完整，继承 StepInput
- ASROutput 类定义完整，继承 StepOutput
- 两个测试通过
</done>
</task>
<task type="auto" tdd="true">
<name>Task 2: Create ASRProcessor with subtitle download</name>
<files>kb_pipeline/processors/__init__.py, kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py</files>
<behavior>
- Test 1: ASRProcessor 初始化成功
- Test 2: _try_official_subtitle 调用成功时返回 ASROutput (tool="subtitle")
- Test 3: _try_official_subtitle 无字幕时返回 None
- Test 4: process() 方法优先尝试官方字幕
</behavior>
<action>
创建 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）。
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py::test_asr_processor_init -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_official_subtitle_found -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_official_subtitle_not_found -x -v</automated>
</verify>
<done>
- ASRProcessor 类创建完成
- _try_official_subtitle 方法实现
- 三个测试通过
</done>
</task>
<task type="auto" tdd="true">
<name>Task 3: Implement Whisper.cpp integration</name>
<files>kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py</files>
<behavior>
- Test 1: _run_whisper 调用成功返回 ASROutput (tool="whisper")
- Test 2: 临时音频文件处理后自动清理
- Test 3: Whisper 路径不存在时抛出 RuntimeError
</behavior>
<action>
在 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 实现。
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py::test_whisper_basic -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_temp_file_cleanup -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_whisper_path_not_found -x -v</automated>
</verify>
<done>
- _run_whisper 方法实现完成
- _run_whisper_basic 实现完成
- 临时文件清理逻辑正确
- 三个测试通过
</done>
</task>
<task type="auto" tdd="true">
<name>Task 4: Implement speaker diarization</name>
<files>kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py</files>
<behavior>
- Test 1: HF_TOKEN 未配置时使用普通 Whisper
- Test 2: HF_TOKEN 配置时自动启用说话者分离
- Test 3: 说话者分离结果包含 speaker 字段
</behavior>
<action>
在 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
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py::test_no_hf_token_uses_basic_whisper -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_diarization_with_hf_token -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_diarization_result_has_speaker -x -v</automated>
</verify>
<done>
- _run_whisper_with_diarization 方法实现
- HF_TOKEN 自动检测逻辑正确
- 说话者分离结果格式正确
- 三个测试通过
</done>
</task>
<task type="auto" tdd="true">
<name>Task 5: Add state recording</name>
<files>kb_pipeline/processors/asr.py, tests/test_processors/test_asr.py</files>
<behavior>
- Test 1: ASR 成功时状态记录为 completed
- Test 2: ASR 失败时状态记录为 failed
- Test 3: output_summary 包含 tool 和 segments 数量
</behavior>
<action>
在 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 时不记录状态）
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py::test_state_recording_success -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_state_recording_failure -x -v</automated>
<automated>pytest tests/test_processors/test_asr.py::test_state_no_manager -x -v</automated>
</verify>
<done>
- 状态记录逻辑正确
- output_summary 格式正确
- state_manager 可选处理正确
- 三个测试通过
</done>
</task>
<task type="auto">
<name>Task 6: Create test infrastructure</name>
<files>tests/test_processors/__init__.py, tests/test_processors/conftest.py, tests/test_processors/test_asr.py</files>
<action>
创建测试基础设施：
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）。
</action>
<verify>
<automated>pytest tests/test_processors/test_asr.py --collect-only | grep "test_" | wc -l</automated>
</verify>
<done>
- tests/test_processors/ 目录创建完成
- conftest.py 包含所有 fixtures
- test_asr.py 包含所有测试用例骨架
- 测试数量 >= 13 个
</done>
</task>
</tasks>
<verification>
## 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
```
</verification>
<success_criteria>
1. ASRProcessor 类可被实例化和调用
2. B站视频可以自动下载官方字幕（优先策略）
3. 无官方字幕时自动调用 Whisper.cpp 进行语音识别
4. HF_TOKEN 配置时自动启用说话者分离
5. 输出包含 txt/srt/json 三种格式
6. ASR 完成后状态正确记录到 workflow_state.json
7. 临时音频文件处理后自动清理
8. 所有测试通过（>= 13 个测试）
</success_criteria>
<output>
After completion, create `.planning/phases/03-processing-skills/03-01-SUMMARY.md`
</output>