starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
This repository has been archived on 2026-08-07. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
8d67f3c95331f4bf2ffbdd9b8823568b10fdc8a9
kb-skills/.planning/phases/03-processing-skills/03-01-PLAN.md
T

20 KiB
Raw Blame History

phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, user_setup, must_haves
phase plan type wave depends_on files_modified autonomous requirements user_setup must_haves
03-processing-skills 01 execute 1
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
true
service why env_vars dashboard_config
whisper.cpp 本地语音识别引擎
task location
安装 whisper.cpp 和下载模型 git clone https://github.com/ggml-org/whisper.cpp.git ~/whisper.cpp && cd ~/whisper.cpp && make && ./models/download-ggml-model.sh base
service why env_vars dashboard_config
pyannote.audio (可选) 说话者分离功能
name source
HF_TOKEN HuggingFace Settings -> Access Tokens
task location
接受 pyannote 模型用户协议 https://huggingface.co/pyannote/speaker-diarization-3.1
truths artifacts key_links
B站视频可以自动下载官方字幕
无官方字幕时自动调用 Whisper.cpp 进行语音识别
HF_TOKEN 配置时自动启用说话者分离
输出包含 txt/srt/json 三种格式
ASR 完成后状态记录到 workflow_state.json
临时音频文件处理后自动清理
path provides exports
kb_pipeline/processors/__init__.py 处理器模块初始化
ASRProcessor
path provides contains min_lines
kb_pipeline/processors/asr.py ASR 处理器实现 class ASRProcessor 150
path provides contains contains
kb_pipeline/schemas.py ASR 输入输出 Schema class ASRInput class ASROutput
from to via pattern
kb_pipeline/processors/asr.py kb_pipeline/schemas.py from ..schemas import ASRInput, ASROutput ASRInput|ASROutput
from to via pattern
kb_pipeline/processors/asr.py kb_pipeline/state.py WorkflowStateManager state_manager.mark_step
from to via pattern
kb_pipeline/processors/asr.py kb-bilibili-content/bilibili_content.py download_subtitle 逻辑复用 download_subtitle|has_subtitle
from to via pattern
kb_pipeline/processors/asr.py kb-whisper/whisper.py Whisper.cpp 调用逻辑复用 whisper-cli|convert_to_wav
创建 ASRProcessor 处理器，统一封装 B站官方字幕下载和 Whisper.cpp 语音识别，遵循 Phase 1/2 建立的管道模式。

Purpose: 为视频内容提供统一的文字提取入口，优先使用高质量官方字幕，无字幕时自动降级到本地语音识别。 Output: ASRProcessor 类，支持官方字幕下载、Whisper.cpp 调用、说话者分离、状态记录。

<execution_context> @/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md @/home/starmen1265/.claude/get-shit-done/templates/summary.md </execution_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)

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)

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:

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:

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:

class KBSettings(BaseSettings):
    hf_token: SecretStr = Field(default=SecretStr(""))
    # 其他配置...

From kb_pipeline/retry.py:

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：

    from .asr import ASRProcessor
__all__ = ["ASRProcessor"]

  2. 创建 kb_pipeline/processors/asr.py：

    ASRProcessor 类结构：

    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 实现：

  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 方法：

    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. 命令构建：

    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 检查：

    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 方法：

    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 调用：

    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. 成功时记录：

    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. 失败时记录：

    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：

    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. 测试文件结构：

    # 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 验证

    pytest tests/test_processors/test_asr.py::test_asr_input_schema tests/test_processors/test_asr.py::test_asr_output_schema -v

  2. 官方字幕下载

    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 集成

    pytest tests/test_processors/test_asr.py::test_whisper_basic tests/test_processors/test_asr.py::test_temp_file_cleanup -v

  4. 说话者分离

    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. 状态记录

    pytest tests/test_processors/test_asr.py::test_state_recording_success tests/test_processors/test_asr.py::test_state_recording_failure -v

  6. 完整测试套件

    pytest tests/test_processors/test_asr.py -v

<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>
After completion, create `.planning/phases/03-processing-skills/03-01-SUMMARY.md`
Reference in New Issue View Git Blame Copy Permalink