openclaw/agents/haonan/workspace/memory/tasks/obsidian-kb-subtitle-plan.md

obsidian-kb 本地生成字幕功能 - 详细规划

创建时间: 2026-02-07 01:26 任务优先级: 🔵 P3（低，可选增强） 预计总工时: 4-6 小时

---

🎯 目标

为无字幕视频自动生成字幕，使 AI 分析功能不受字幕限制，覆盖更多视频内容。

---

📋 技术方案对比

方案对比表

方案	准确度	速度	成本	复杂度	推荐度
Whisper 本地	⭐⭐⭐⭐	⭐⭐⭐	免费	低	⭐⭐⭐⭐⭐
阿里云 ASR	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	按量付费	低	⭐⭐⭐⭐
腾讯云 ASR	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	按量付费	低	⭐⭐⭐⭐
百度云 ASR	⭐⭐⭐	⭐⭐⭐⭐	按量付费	低	⭐⭐⭐

推荐：Whisper 本地方案

理由：

• ✅ 完全免费，无 API 费用
• ✅ 数据本地处理，隐私安全
• ✅ 准确度足够（中英文识别效果良好）
• ✅ 可离线使用
• ✅ 社区活跃，持续更新

权衡：

• ⚠️ 需要下载模型文件（~150MB-3GB）
• ⚠️ CPU 速度较慢（10分钟视频约需 2-5 分钟）
• ⚠️ 需要安装 ffmpeg 依赖

优化方向：

• 使用 GPU 加速（如果有 CUDA）
• 选择合适的模型大小（base vs small vs medium）
• 添加进度反馈和取消功能

---

🏗️ 架构设计

核心模块

subtitle_generator.py  ← 新建
├── AudioDownloader      # 音频下载器
├── WhisperTranscriber    # Whisper 识别器
├── SubtitleFormatter    # SRT 格式化器
├── SubtitleCache        # 缓存管理器
└── SubtitleGenerator    # 统一接口

工作流程

用户请求视频处理
    ↓
检查是否有字幕
    ↓ [有字幕]
使用现有字幕
    ↓ [无字幕]
下载音频 (yt-dlp)
    ↓
语音识别 (Whisper)
    ↓
转换为 SRT 格式
    ↓
保存到缓存
    ↓
继续 AI 分析

---

📝 详细实施步骤

Step 1: 环境准备（30分钟）

1.1 安装依赖

# Python 包
pip install openai-whisper
pip install yt-dlp

# 系统依赖（检查）
ffmpeg -v

1.2 验证安装

# 测试 Whisper
whisper --help

# 测试 yt-dlp
yt-dlp --version

1.3 创建缓存目录

mkdir -p cache/audio
mkdir -p cache/subtitles

---

Step 2: 音频下载模块（1小时）

2.1 创建 subtitle_generator.py

class AudioDownloader:
    """使用 yt-dlp 下载视频音频"""

    def download_audio(self, url: str, output_path: str) -> str:
        """
        下载视频音频

        Args:
            url: 视频URL
            output_path: 输出音频路径

        Returns:
            str: 实际保存的音频文件路径
        """
        import yt_dlp

        ydl_opts = {
            'format': 'bestaudio/best',
            'outtmpl': output_path,
            'postprocessors': [{
                'key': 'FFmpegExtractAudio',
                'preferredcodec': 'mp3',
                'preferredquality': '192',
            }],
            'quiet': True,
            'no_warnings': True,
        }

        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.download([url])

        return output_path.replace('.%(ext)s', '.mp3')

2.2 测试下载功能

python3 -c "
from subtitle_generator import AudioDownloader
downloader = AudioDownloader()
downloader.download_audio('https://www.bilibili.com/video/BVxxxxx', 'test_audio.mp3')
"

---

Step 3: Whisper 识别模块（1.5小时）

3.1 实现 WhisperTranscriber

import whisper
import time

class WhisperTranscriber:
    """Whisper 语音识别器"""

    # 模型选择
    MODELS = {
        'tiny': 39M,      # 最快，准确度最低
        'base': 74M,      # 推荐，速度/准确度平衡
        'small': 244M,    # 准确度更高
        'medium': 769M,   # 接近专业级
        'large': 1550M,   # 最高准确度
    }

    def __init__(self, model_size: str = 'base'):
        """
        初始化识别器

        Args:
            model_size: 模型大小，默认 'base'
        """
        self.model_size = model_size
        self.model = None
        self.device = None

    def load_model(self):
        """加载 Whisper 模型"""
        if self.model is None:
            print(f"加载 Whisper 模型 ({self.model_size})...")
            start_time = time.time()

            # 自动检测 GPU
            import torch
            self.device = "cuda" if torch.cuda.is_available() else "cpu"

            self.model = whisper.load_model(
                self.model_size,
                device=self.device
            )

            load_time = time.time() - start_time
            print(f"模型加载完成 ({self.device}) - 耗时: {load_time:.1f}秒")

    def transcribe(
        self,
        audio_path: str,
        language: str = None,
        progress_callback = None
    ) -> dict:
        """
        语音识别

        Args:
            audio_path: 音频文件路径
            language: 语言代码（'zh'、'en'、None=自动检测）
            progress_callback: 进度回调函数

        Returns:
            dict: Whisper 识别结果
        """
        self.load_model()

        print(f"开始识别: {audio_path}")
        start_time = time.time()

        result = self.model.transcribe(
            audio_path,
            language=language,
            task='transcribe',
            verbose=False
        )

        duration = time.time() - start_time
        print(f"识别完成 - 耗时: {duration:.1f}秒")

        return result

    def get_segments(self, result: dict) -> list:
        """获取分段信息"""
        return result.get('segments', [])

3.2 测试识别功能

python3 -c "
from subtitle_generator import WhisperTranscriber
transcriber = WhisperTranscriber('base')
result = transcriber.transcribe('test_audio.mp3', language='zh')
print(f'识别文本: {result[\"text\"][:100]}...')
"

---

Step 4: SRT 格式化模块（30分钟）

4.1 实现 SubtitleFormatter

class SubtitleFormatter:
    """字幕格式化器（SRT 格式）"""

    @staticmethod
    def to_srt(segments: list) -> str:
        """
        将 Whisper segments 转换为 SRT 格式

        Args:
            segments: Whisper segments 列表

        Returns:
            str: SRT 格式文本
        """
        srt_lines = []

        for i, segment in enumerate(segments, 1):
            # SRT 格式：
            # 序号
            # 开始时间 --> 结束时间
            # 文本
            # 空行

            start_time = SubtitleFormatter._seconds_to_srt_time(segment['start'])
            end_time = SubtitleFormatter._seconds_to_srt_time(segment['end'])
            text = segment['text'].strip()

            srt_lines.append(str(i))
            srt_lines.append(f"{start_time} --> {end_time}")
            srt_lines.append(text)
            srt_lines.append("")

        return "\n".join(srt_lines)

    @staticmethod
    def _seconds_to_srt_time(seconds: float) -> str:
        """
        将秒数转换为 SRT 时间格式 (HH:MM:SS,mmm)

        Args:
            seconds: 秒数

        Returns:
            str: SRT 时间格式
        """
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        millis = int((seconds - int(seconds)) * 1000)

        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"

4.2 测试格式化

python3 -c "
from subtitle_generator import SubtitleFormatter
segments = [{'start': 0.0, 'end': 2.5, 'text': '测试字幕'}]
srt = SubtitleFormatter.to_srt(segments)
print(srt)
"

---

Step 5: 缓存管理模块（30分钟）

5.1 实现 SubtitleCache

import os
import hashlib
from pathlib import Path

class SubtitleCache:
    """字幕缓存管理器"""

    def __init__(self, cache_dir: str = 'cache/subtitles'):
        """
        初始化缓存

        Args:
            cache_dir: 缓存目录
        """
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def get_cache_key(self, url: str) -> str:
        """
        生成缓存键

        Args:
            url: 视频URL

        Returns:
            str: 缓存键（SHA256 哈希的前8位）
        """
        return hashlib.sha256(url.encode()).hexdigest()[:8]

    def get_cache_path(self, url: str) -> str:
        """
        获取缓存文件路径

        Args:
            url: 视频URL

        Returns:
            str: 缓存文件路径
        """
        cache_key = self.get_cache_key(url)
        return str(self.cache_dir / f"{cache_key}.srt")

    def get(self, url: str) -> str:
        """
        获取缓存字幕

        Args:
            url: 视频URL

        Returns:
            str | None: 缓存的字幕内容，如果不存在返回 None
        """
        cache_path = self.get_cache_path(url)

        if os.path.exists(cache_path):
            with open(cache_path, 'r', encoding='utf-8') as f:
                return f.read()

        return None

    def set(self, url: str, subtitle: str) -> str:
        """
        保存缓存字幕

        Args:
            url: 视频URL
            subtitle: 字幕内容

        Returns:
            str: 缓存文件路径
        """
        cache_path = self.get_cache_path(url)

        with open(cache_path, 'w', encoding='utf-8') as f:
            f.write(subtitle)

        return cache_path

    def exists(self, url: str) -> bool:
        """检查缓存是否存在"""
        return os.path.exists(self.get_cache_path(url))

---

Step 6: 统一接口模块（30分钟）

6.1 实现 SubtitleGenerator

class SubtitleGenerator:
    """字幕生成器统一接口"""

    def __init__(self, model_size: str = 'base', cache_dir: str = 'cache/subtitles'):
        """
        初始化生成器

        Args:
            model_size: Whisper 模型大小
            cache_dir: 缓存目录
        """
        self.audio_downloader = AudioDownloader()
        self.transcriber = WhisperTranscriber(model_size)
        self.formatter = SubtitleFormatter()
        self.cache = SubtitleCache(cache_dir)

    def generate(
        self,
        url: str,
        content_id: str,
        audio_cache_dir: str = 'cache/audio',
        language: str = None,
        use_cache: bool = True
    ) -> str:
        """
        生成字幕

        Args:
            url: 视频URL
            content_id: 内容ID
            audio_cache_dir: 音频缓存目录
            language: 语言代码（None=自动检测）
            use_cache: 是否使用缓存

        Returns:
            str: SRT 格式字幕
        """
        # 检查缓存
        if use_cache and self.cache.exists(url):
            print(f"使用缓存字幕: {url}")
            return self.cache.get(url)

        print(f"开始生成字幕: {url}")

        # 下载音频
        import os
        os.makedirs(audio_cache_dir, exist_ok=True)
        audio_path = os.path.join(audio_cache_dir, f"{content_id}.mp3")

        print(f"下载音频...")
        audio_path = self.audio_downloader.download_audio(url, audio_path)

        # 语音识别
        result = self.transcriber.transcribe(audio_path, language=language)
        segments = self.transcriber.get_segments(result)

        # 转换为 SRT
        srt_text = self.formatter.to_srt(segments)

        # 保存缓存
        cache_path = self.cache.set(url, srt_text)
        print(f"字幕已缓存: {cache_path}")

        # 清理音频文件（可选）
        # os.remove(audio_path)

        return srt_text

---

Step 7: 集成到现有流程（1小时）

7.1 修改 pipeline.py

在 ProcessingPipeline 中集成字幕生成：

from subtitle_generator import SubtitleGenerator

class ProcessingPipeline:

    def __init__(self, config: Config = None):
        self.config = config or get_config()
        # ... 现有初始化

        # 添加字幕生成器
        self.subtitle_generator = SubtitleGenerator(
            model_size=self.config.get('subtitle_model', 'base'),
            cache_dir=self.config.get('subtitle_cache_dir', 'cache/subtitles')
        )

    def _ensure_subtitle(self, content: Content) -> str:
        """
        确保内容有字幕，如果没有则生成

        Args:
            content: 内容对象

        Returns:
            str: 字幕文本
        """
        # 1. 检查是否已有字幕
        if content.transcript and len(content.transcript) > 100:
            print("使用现有字幕")
            return content.transcript

        # 2. 检查是否启用自动生成
        if not self.config.get('auto_generate_subtitles', False):
            print("自动生成字幕未启用，跳过")
            return ""

        # 3. 生成字幕
        try:
            print(f"开始自动生成字幕...")
            transcript = self.subtitle_generator.generate(
                url=content.url,
                content_id=content.id,
                language=self.config.get('subtitle_language', None),
                use_cache=True
            )

            # 更新内容
            content.transcript = transcript
            self.storage.save_content(content)

            return transcript

        except Exception as e:
            print(f"字幕生成失败: {e}")
            import traceback
            traceback.print_exc()
            return ""

7.2 更新配置文件

在 config.yaml 中添加字幕配置：

# 字幕生成配置
subtitle:
  enabled: false              # 是否启用自动生成
  model: base                # Whisper 模型大小: tiny/base/small/medium/large
  language: null             # 语言: zh/en/null(自动检测)
  cache_dir: cache/subtitles  # 缓存目录
  audio_cache_dir: cache/audio  # 音频缓存目录

---

Step 8: 测试与优化（1小时）

8.1 创建测试脚本

# test_subtitle_generator.py

def test_full_workflow():
    """测试完整工作流程"""

    generator = SubtitleGenerator(model_size='base')

    # 测试 URL（10分钟左右的视频）
    test_url = "https://www.bilibili.com/video/BV1xx411c7mD"

    print("=" * 50)
    print("测试完整字幕生成流程")
    print("=" * 50)

    srt_text = generator.generate(
        url=test_url,
        content_id='test001',
        language='zh',
        use_cache=True
    )

    print(f"\n生成字幕预览（前500字符）：")
    print(srt_text[:500])

    print(f"\n总字符数: {len(srt_text)}")

    # 验证缓存
    print("\n验证缓存...")
    cached = generator.generate(
        url=test_url,
        content_id='test002',
        use_cache=True
    )
    assert cached == srt_text, "缓存验证失败"
    print("✅ 缓存正常")

    print("\n✅ 测试通过")

if __name__ == '__main__':
    test_full_workflow()

8.2 性能测试

模型	10分钟视频时长	准确度	推荐场景
tiny	~30秒	⭐⭐	快速测试、批量处理
base	~2分钟	⭐⭐⭐⭐	日常使用（推荐）
small	~5分钟	⭐⭐⭐⭐⭐	高质量需求
medium	~10分钟	⭐⭐⭐⭐⭐	专业用途
large	~15分钟+	⭐⭐⭐⭐⭐	最高准确度

---

📦 文件清单

新增文件

skills/obsidian-kb/
├── subtitle_generator.py       # 核心模块
├── test_subtitle_generator.py  # 测试脚本
├── cache/
│   ├── audio/                 # 音频缓存
│   └── subtitles/             # 字幕缓存

修改文件

skills/obsidian-kb/
├── src/pipeline.py           # 集成字幕生成
├── config.yaml.example        # 添加字幕配置
├── SKILL.md                 # 更新文档
├── requirements.txt          # 添加依赖

依赖更新

requirements.txt:

openai-whisper>=20230314
yt-dlp>=2023.1.6
torch>=2.0.0  # Whisper 依赖

---

✅ 验收标准

功能验收

• 无字幕视频能成功生成字幕
• 字幕准确度可接受（测试 5 个视频）
• 处理时间合理（10分钟视频 < 5 分钟，使用 base 模型）
• 支持中文字幕
• 支持英文识别
• 缓存机制正常工作（避免重复生成）
• 集成到 pipeline，无字幕视频自动生成

代码验收

• 代码结构清晰，模块化设计
• 有完整的错误处理
• 有日志输出和进度反馈
• 单元测试通过
• 集成测试通过

文档验收

• SKILL.md 更新，说明字幕生成功能
• 配置文档说明（config.yaml.example）
• 有使用示例和注意事项

---

⚠️ 注意事项

1. 存储空间

• Whisper 模型文件：~150MB (tiny) - 3GB (large)
• 音频缓存：每分钟视频约 1-2MB
• 字幕缓存：每分钟视频约 1-3KB

建议：

• 定期清理 cache/audio 目录
• 字幕缓存可以长期保留
• 使用 base 模型作为默认

2. CPU/GPU

• 无 GPU: 使用 base 模型，10分钟视频约 2-5 分钟
• 有 GPU: 可使用 medium/large 模型，速度提升 5-10 倍

3. 隐私安全

• Whisper 本地处理，不上传音频
• 缓存文件在本地，不共享

4. 错误处理

• 下载失败 → 记录错误，继续处理
• 识别失败 → 记录错误，继续处理
• 格式化失败 → 使用原始文本

---

🎯 预期效果

用户体验

• 无字幕视频也能生成笔记
• 等待时间可接受（显示进度）
• 缓存加速重复处理

覆盖率提升

• 当前：只能处理有字幕视频（~60%）
• 目标：所有视频（~95%，部分音频质量问题除外）

成本

• 软件成本：0 元（开源免费）
• 硬件成本：0 元（复用现有机器）
• 网络流量：下载音频（每次约 5-20MB）

---

📅 时间线

阶段	任务	预计时间
Phase 1	环境准备 + 音频下载	1.5h
Phase 2	Whisper 识别 + SRT 格式化	2h
Phase 3	缓存 + 集成 + 测试	1.5h
Phase 4	文档 + 优化	1h
总计		4-6h

---

🚀 后续优化方向

1. GPU 加速 - 如果有 CUDA，可大幅提升速度
2. 多语言支持 - 支持更多语言识别
3. 批量处理 - 一次生成多个视频的字幕
4. 字幕编辑 - 提供 Web 界面编辑生成的字幕
5. 云端集成 - 可选切换到云 API（阿里云/腾讯云）

---

规划完成！是否开始实施？