From 29dd236bb9374dc341131e5074f21f795bc7d414 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 09:04:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(kb-workflow):=20=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E4=B8=BA=20Whisper=20=E8=AF=AD=E9=9F=B3=E8=AF=86=E5=88=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将步骤5从 OCR 改为 Whisper 语音识别 - 更新 kb-whisper 支持自动音频转换 - 默认使用中文语言 (zh) 和 base 模型 - 更新输出格式为 {platform}_{id}.txt/.srt/.json - 添加 FunASR 作为备用方案 --- skills/kb-funasr/.env.example | 2 + skills/kb-funasr/README.md | 212 +++++++++++++++++++ skills/kb-funasr/SKILL.md | 113 ++++++++++ skills/kb-funasr/funasr_tool.py | 313 ++++++++++++++++++++++++++++ skills/kb-whisper/whisper.py | 331 +++++++++++++++++------------- skills/kb-workflow/SKILL.md | 23 +-- skills/kb-workflow/kb_workflow.py | 5 - 7 files changed, 844 insertions(+), 155 deletions(-) create mode 100644 skills/kb-funasr/.env.example create mode 100644 skills/kb-funasr/README.md create mode 100644 skills/kb-funasr/SKILL.md create mode 100755 skills/kb-funasr/funasr_tool.py diff --git a/skills/kb-funasr/.env.example b/skills/kb-funasr/.env.example new file mode 100644 index 0000000..1c21fe6 --- /dev/null +++ b/skills/kb-funasr/.env.example @@ -0,0 +1,2 @@ +# FunASR 模型缓存目录 +FUNASR_CACHE_DIR=~/.cache/modelscope/hub diff --git a/skills/kb-funasr/README.md b/skills/kb-funasr/README.md new file mode 100644 index 0000000..baef2d8 --- /dev/null +++ b/skills/kb-funasr/README.md @@ -0,0 +1,212 @@ +# KB FunASR - 本地语音识别 + +基于阿里 FunASR 的本地语音识别工具，支持将音视频文件转换为文本和 JSON 格式。 + +## 功能特性 + +- 🎯 **本地执行** - 完全离线运行，保护隐私 +- 🚀 **多模型支持** - Paraformer、SenseVoice 等模型可选 +- 📝 **多格式输出** - JSON、TXT 两种格式 +- ⚡ **异步执行** - 支持后台运行，不阻塞主代理 +- 🎬 **广泛兼容** - 支持常见音视频格式（MP3、MP4、WAV 等） +- 🇨🇳 **中文优化** - Paraformer 模型针对中文语音优化 + +## 安装前提 + +### 1. 安装依赖 + +```bash +pip install funasr modelscope +``` + +### 2. 配置环境变量（可选） + +在 `~/.clawdbot/clawdbot.json` 中添加： + +```json5 +{ + skills: { + entries: { + "kb-funasr": { + env: { + FUNASR_CACHE_DIR: "~/.cache/modelscope/hub" + } + } + } + } +} +``` + +或通过命令行配置： + +```bash +clawdbot skills config kb-funasr FUNASR_CACHE_DIR "~/.cache/modelscope/hub" +``` + +## 使用方法 + +### 基本用法 + +```bash +# 使用默认 Paraformer 中文模型处理 +python funasr.py --input video.mp4 + +# 后台异步执行 +python funasr.py --input video.mp4 --background +``` + +### 指定模型 + +```bash +# Paraformer 中文模型（推荐） +python funasr.py --input video.mp4 --model paraformer-zh + +# Paraformer 多语言模型 +python funasr.py --input video.mp4 --model paraformer + +# SenseVoice 多语言模型 +python funasr.py --input video.mp4 --model sensevoice +``` + +### 指定输出目录 + +```bash +python funasr.py --input video.mp4 --output ./transcripts +``` + +## 输出文件 + +对于输入 `video.mp4`，将生成以下文件： + +``` +output/ +├── video.json # 详细结果（含时间戳） +└── video.txt # 纯文本 +``` + +### JSON 格式 + +```json +{ + "success": true, + "input_file": "video.mp4", + "model": "paraformer-zh", + "output_files": { + "json": "video.json", + "txt": "video.txt" + }, + "duration_seconds": 120.5, + "segments": [ + { + "id": 0, + "start": 0.0, + "end": 3.2, + "text": "这是第一段识别结果" + }, + { + "id": 1, + "start": 3.2, + "end": 6.8, + "text": "这是第二段识别结果" + } + ] +} +``` + +## 模型对比 + +| 模型 | 速度 | 精度 | 适用场景 | +|------|------|------|----------| +| paraformer-zh | ⚡⚡⚡ | ⭐⭐⭐⭐ | 中文语音（推荐） | +| paraformer | ⚡⚡ | ⭐⭐⭐⭐ | 多语言识别 | +| sensevoice | ⚡ | ⭐⭐⭐⭐⭐ | 高精度多语言 | + +## 完整示例 + +### 示例 1: 转录播客 + +```bash +python funasr.py --input podcast.mp3 --model paraformer-zh --background +``` + +### 示例 2: 生成视频识别结果 + +```bash +python funasr.py --input tutorial.mp4 --model paraformer +``` + +### 示例 3: 批量处理（配合脚本） + +```bash +#!/bin/bash +for video in videos/*.mp4; do + python funasr.py --input "$video" --model paraformer-zh +done +``` + +## 在 Workflow 中使用 + +在知识库处理流程中，如果视频没有识别结果： + +```bash +# 步骤4: 下载视频后 +# 步骤5: 使用 FunASR 生成识别结果 +python funasr.py --input ${video_path} --model paraformer-zh --background + +# 等待 FunASR 完成，然后继续后续步骤 +``` + +## 技术细节 + +- **引擎**: 阿里 FunASR +- **模型框架**: ModelScope +- **音频采样率**: 自动转换为 16kHz +- **语言**: 根据模型支持不同语言 + +## 故障排查 + +### 问题: 模型下载失败 + +```bash +# 首次运行会自动下载，确保网络稳定 +python funasr.py --input test.mp4 + +# 手动检查缓存目录 +ls ~/.cache/modelscope/hub/ +``` + +### 问题: 音频格式不支持 + +```bash +# 转换音频格式 +ffmpeg -i input.avi -acodec libmp3lame output.mp3 +python funasr.py --input output.mp3 +``` + +### 问题: 依赖安装失败 + +```bash +# 使用清华镜像源 +pip install funasr modelscope -i https://pypi.tuna.tsinghua.edu.cn/simple +``` + +## 性能参考 + +在 CPU 环境下（Intel i7）： + +- **paraformer-zh 模型**: ~2x 实时速度（1分钟音频 ~30秒处理） +- **paraformer 模型**: ~1.5x 实时速度（1分钟音频 ~40秒处理） +- **sensevoice 模型**: ~1x 实时速度（1分钟音频 ~1分钟处理） + +## 与 Whisper 对比 + +| 特性 | FunASR | Whisper | +|------|--------|---------| +| 中文识别 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | +| 速度 | ⚡⚡⚡ | ⚡⚡ | +| 模型大小 | 较小 | 较大 | +| 多语言支持 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | + +## 许可证 + +本技能使用 MIT 许可证，FunASR 使用 Apache 2.0 许可证。 diff --git a/skills/kb-funasr/SKILL.md b/skills/kb-funasr/SKILL.md new file mode 100644 index 0000000..886012c --- /dev/null +++ b/skills/kb-funasr/SKILL.md @@ -0,0 +1,113 @@ +--- +name: kb-funasr +description: 本地语音识别，支持音视频转文字（基于阿里 FunASR） +--- + +# KB FunASR Skill + +**kb-funasr** - 本地语音识别，使用阿里 FunASR 将音视频文件转换为文本和 JSON 格式输出。 + +## 什么时候使用 + +当用户需要： +- 将视频/音频文件转换为文字 +- 获取带时间戳的详细识别结果 +- 需要中文优化的语音识别（Paraformer 模型） + +## 环境变量 + +在使用之前，必须检查环境变量是否已经设置。 + +| 变量名 | 用途 | 示例 | +|--------|------|------| +| `FUNASR_CACHE_DIR` | FunASR 模型缓存目录 | `~/.cache/modelscope/hub` | + +## 使用方法 + +```bash +# 基本用法（使用 Paraformer 模型） +python funasr.py --input video.mp4 + +# 指定模型 +python funasr.py --input video.mp4 --model paraformer-zh + +# 指定输出目录 +python funasr.py --input video.mp4 --output ./output + +# 异步执行（推荐） +python funasr.py --input video.mp4 --background +``` + +## 命令参数 + +| 参数 | 说明 | 默认值 | +|------|------|--------| +| `--input` | 输入音视频文件路径 | 必需 | +| `--model` | FunASR 模型类型 | `paraformer-zh` | +| `--output` | 输出目录 | 文件所在目录 | +| `--background` | 后台异步执行 | False | + +## 模型选项 + +- `paraformer-zh` - Paraformer 中文模型（轻量，中文优化，推荐） +- `paraformer` - Paraformer 多语言模型 +- `sensevoice` - SenseVoice 多语言模型 + +## 输出文件 + +对于输入 `video.mp4`，将生成： + +- `video.json` - 详细结果（包含时间戳、分段信息） +- `video.txt` - 纯文本内容 + +## 输出格式示例 + +### JSON 格式 + +```json +{ + "success": true, + "input_file": "video.mp4", + "model": "paraformer-zh", + "output_files": { + "json": "video.json", + "txt": "video.txt" + }, + "duration_seconds": 120.5, + "segments": [ + { + "id": 0, + "start": 0.0, + "end": 3.2, + "text": "这是第一段识别结果" + } + ] +} +``` + +## Workflow 中的使用 + +```bash +# 步骤4之后：如果已有音频但无字幕，可以使用 FunASR +python funasr.py --input ${video_path} --model paraformer-zh --background +``` + +## 安装 + +```bash +# 1. 安装依赖 +pip install funasr modelscope + +# 2. 首次运行时会自动下载模型到缓存目录 +python funasr.py --input test.mp4 + +# 3. 复制技能到 skills 目录 +cp -r kb-funasr/ ~/skills/ +``` + +## 注意事项 + +- **异步执行**: 使用 `--background` 参数时，命令会在后台运行，不阻塞主代理 +- **模型选择**: paraformer-zh 模型适合中文语音识别，速度较快，精度较高 +- **文件格式**: 支持 MP3、MP4、WAV、FLAC、OGG、M4A 等常见格式 +- **首次运行**: 首次运行时会自动下载模型文件，需要稳定的网络连接 diff --git a/skills/kb-funasr/funasr_tool.py b/skills/kb-funasr/funasr_tool.py new file mode 100755 index 0000000..b082155 --- /dev/null +++ b/skills/kb-funasr/funasr_tool.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 +""" +KB FunASR - 本地语音识别工具 +基于阿里 FunASR 实现音视频转文字 +""" + +import argparse +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path +from datetime import datetime +from typing import Dict, List, Optional + + +class FunASRTranscriber: + """FunASR 语音识别器""" + + # 模型映射 + MODEL_MAP = { + "paraformer-zh": "damo/speech_paraformer-large-vad_punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch", + "paraformer": "damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch", + "sensevoice": "FunASR/SenseVoiceSmall" + } + + def __init__(self, model: str = "paraformer-zh", cache_dir: Optional[str] = None): + """初始化 FunASR 识别器 + + Args: + model: 模型名称 + cache_dir: 模型缓存目录 + """ + self.model_name = model + self.model_id = self.MODEL_MAP.get(model, self.MODEL_MAP["paraformer-zh"]) + self.cache_dir = cache_dir or os.path.expanduser("~/.cache/modelscope/hub") + self._model = None + + def _load_model(self): + """延迟加载模型""" + if self._model is None: + try: + from funasr import AutoModel + print(f"正在加载模型: {self.model_name} ({self.model_id})") + print(f"缓存目录: {self.cache_dir}") + self._model = AutoModel( + model=self.model_id, + disable_pbar=False, + disable_log=False, + ) + print("模型加载完成") + except ImportError: + print("错误: 未安装 funasr 库") + print("请运行: pip install funasr modelscope") + sys.exit(1) + + def _convert_audio(self, input_path: str) -> str: + """转换音频格式为 16kHz WAV + + Args: + input_path: 输入文件路径 + + Returns: + 转换后的文件路径（如果不需要转换则返回原路径） + """ + # 检查是否已经是 WAV 格式 + if input_path.lower().endswith('.wav'): + # 检查采样率 + try: + result = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', + 'stream=sample_rate', '-of', 'default=noprint_wrappers=1:nokey=1', + input_path], + capture_output=True, text=True + ) + if result.returncode == 0: + sample_rate = int(result.stdout.strip()) + if sample_rate == 16000: + return input_path + except Exception: + pass + + # 需要转换 + output_path = input_path.rsplit('.', 1)[0] + '_converted.wav' + print(f"正在转换音频格式: {input_path} -> {output_path}") + + try: + subprocess.run([ + 'ffmpeg', '-y', '-i', input_path, + '-acodec', 'pcm_s16le', + '-ar', '16000', + '-ac', '1', + output_path + ], check=True, capture_output=True) + print("音频转换完成") + return output_path + except subprocess.CalledProcessError as e: + print(f"音频转换失败: {e}") + # 如果转换失败，尝试使用原文件 + print("尝试使用原文件继续处理") + return input_path + except FileNotFoundError: + print("警告: 未找到 ffmpeg，跳过音频转换") + return input_path + + def transcribe(self, input_path: str) -> Dict: + """转录音频文件 + + Args: + input_path: 输入文件路径 + + Returns: + 识别结果字典 + """ + input_path = os.path.abspath(input_path) + + if not os.path.exists(input_path): + return { + "success": False, + "error": f"文件不存在: {input_path}" + } + + print(f"\n开始处理: {input_path}") + print(f"使用模型: {self.model_name}") + + # 转换音频格式 + audio_path = self._convert_audio(input_path) + + # 加载模型 + self._load_model() + + # 执行识别 + print("正在进行语音识别...") + try: + result = self._model.generate( + input=audio_path, + batch_size_s=300, + ) + except Exception as e: + return { + "success": False, + "error": f"识别失败: {str(e)}" + } + + # 解析结果 + if not result or not isinstance(result, list): + return { + "success": False, + "error": "识别结果格式错误" + } + + # FunASR 返回格式处理 + segments = [] + full_text = "" + + for seg in result: + if isinstance(seg, dict): + text = seg.get("text", "").strip() + if "timestamp" in seg and seg["timestamp"]: + # 包含时间戳 + timestamp = seg["timestamp"] + if len(timestamp) >= 2: + start_time = timestamp[0][0] if timestamp[0] else 0 + end_time = timestamp[-1][1] if timestamp[-1] else start_time + else: + start_time = 0 + end_time = 0 + else: + start_time = 0 + end_time = 0 + + if text: + segments.append({ + "id": len(segments), + "start": round(start_time, 2), + "end": round(end_time, 2), + "text": text + }) + full_text += text + "\n" + + # 清理临时文件 + if audio_path != input_path and "_converted.wav" in audio_path: + try: + os.remove(audio_path) + except Exception: + pass + + # 计算时长 + duration = segments[-1]["end"] if segments else 0 + + return { + "success": True, + "input_file": os.path.basename(input_path), + "model": self.model_name, + "output_files": {}, + "duration_seconds": round(duration, 2), + "segments": segments, + "text": full_text.strip() + } + + def save_outputs(self, result: Dict, output_dir: str): + """保存输出文件 + + Args: + result: 识别结果 + output_dir: 输出目录 + """ + if not result.get("success"): + return + + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + basename = Path(result["input_file"]).stem + + # 保存 JSON + json_path = output_dir / f"{basename}.json" + with open(json_path, "w", encoding="utf-8") as f: + json.dump(result, f, ensure_ascii=False, indent=2) + print(f"已保存: {json_path}") + result["output_files"]["json"] = str(json_path) + + # 保存 TXT + txt_path = output_dir / f"{basename}.txt" + with open(txt_path, "w", encoding="utf-8") as f: + f.write(result.get("text", "")) + print(f"已保存: {txt_path}") + result["output_files"]["txt"] = str(txt_path) + + +def run_as_background(input_path: str, model: str, output_dir: str): + """后台运行识别 + + Args: + input_path: 输入文件路径 + model: 模型名称 + output_dir: 输出目录 + """ + # 创建日志文件 + log_file = os.path.join(output_dir, f"funasr_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") + + # 重新执行命令（后台模式） + cmd = [ + sys.executable, __file__, + "--input", input_path, + "--model", model, + "--output", output_dir + ] + + with open(log_file, "w") as f: + process = subprocess.Popen( + cmd, + stdout=f, + stderr=subprocess.STDOUT, + start_new_session=True + ) + + print(f"\n✓ 后台任务已启动 (PID: {process.pid})") + print(f"日志文件: {log_file}") + print(f"输出目录: {output_dir}") + print("\n使用以下命令查看进度:") + print(f" tail -f {log_file}") + + +def main(): + parser = argparse.ArgumentParser( + description="KB FunASR - 本地语音识别工具", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + python funasr.py --input video.mp4 + python funasr.py --input podcast.mp3 --model paraformer-zh + python funasr.py --input video.mp4 --output ./transcripts --background + """ + ) + + parser.add_argument("--input", "-i", required=True, help="输入音视频文件路径") + parser.add_argument("--model", "-m", default="paraformer-zh", + choices=["paraformer-zh", "paraformer", "sensevoice"], + help="模型名称 (默认: paraformer-zh)") + parser.add_argument("--output", "-o", default=None, help="输出目录 (默认: 文件所在目录)") + parser.add_argument("--background", "-b", action="store_true", help="后台异步执行") + + args = parser.parse_args() + + # 确定输出目录 + if args.output: + output_dir = args.output + else: + output_dir = os.path.dirname(args.input) or "." + + # 后台模式 + if args.background: + run_as_background(args.input, args.model, output_dir) + return + + # 前台模式 + transcriber = FunASRTranscriber(model=args.model) + result = transcriber.transcribe(args.input) + + if result.get("success"): + transcriber.save_outputs(result, output_dir) + print(f"\n✓ 处理完成!") + print(f" 时长: {result['duration_seconds']} 秒") + print(f" 分段数: {len(result['segments'])}") + else: + print(f"\n✗ 处理失败: {result.get('error')}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/skills/kb-whisper/whisper.py b/skills/kb-whisper/whisper.py index 18707af..4d95f3a 100755 --- a/skills/kb-whisper/whisper.py +++ b/skills/kb-whisper/whisper.py @@ -2,6 +2,11 @@ """ KB Whisper - 本地语音识别工具 基于 Whisper.cpp 的音视频转文字 + +功能： +- 自动将视频/音频转换为 16kHz WAV 格式 +- 支持指定语言（默认中文） +- 输出标准 JSON 格式供 KB Workflow 使用 """ import argparse @@ -9,6 +14,7 @@ import json import os import subprocess import sys +import tempfile from pathlib import Path import re @@ -22,7 +28,7 @@ def parse_arguments(): 示例: python whisper.py --input video.mp4 python whisper.py --input video.mp4 --model base - python whisper.py --input audio.mp3 --output ./output --background + python whisper.py --input video.mp4 --output ./output --background """ ) parser.add_argument( @@ -32,9 +38,9 @@ def parse_arguments(): ) parser.add_argument( '--model', - default='tiny', + default='base', choices=['tiny', 'base', 'small', 'medium', 'large'], - help='Whisper 模型类型 (默认: tiny)' + help='Whisper 模型类型 (默认: base)' ) parser.add_argument( '--output', @@ -47,7 +53,8 @@ def parse_arguments(): ) parser.add_argument( '--language', - help='指定语言 (默认: 自动检测)，例如: en, zh, ja' + default='zh', + help='指定语言 (默认: zh)，例如: en, zh, ja' ) return parser.parse_args() @@ -92,119 +99,184 @@ def check_model(whisper_path, model): return model_path -def run_whisper_cpp(whisper_path, input_file, model, output_dir, language=None): - """运行 whisper.cpp 命令行工具""" +def convert_to_wav(input_path, output_dir): + """将音视频转换为 16kHz WAV 格式 + + Whisper 需要 WAV 格式的输入文件。 + """ + input_path = os.path.abspath(input_path) + + # 如果已经是 WAV 且采样率正确，直接返回 + if input_path.lower().endswith('.wav'): + try: + result = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', + 'stream=sample_rate', '-of', 'default=noprint_wrappers=1:nokey=1', + input_path], + capture_output=True, text=True + ) + if result.returncode == 0 and int(result.stdout.strip()) == 16000: + print(f"输入文件已是 16kHz WAV 格式，直接使用") + return input_path + except Exception: + pass + + # 转换 + output_path = os.path.join(output_dir, f"{Path(input_path).stem}_audio.wav") + + print(f"转换音频格式: {input_path} -> {output_path}") + + try: + subprocess.run([ + 'ffmpeg', '-y', '-i', input_path, + '-acodec', 'pcm_s16le', + '-ar', '16000', + '-ac', '1', + '-loglevel', 'error', + output_path + ], check=True, capture_output=True) + print("音频转换完成") + return output_path + except subprocess.CalledProcessError as e: + print(f"音频转换失败: {e}") + raise RuntimeError(f"无法转换音频格式: {e}") + except FileNotFoundError: + raise RuntimeError("未找到 ffmpeg，请安装: sudo apt-get install ffmpeg") + + +def run_whisper(whisper_path, audio_path, model, output_dir, language='zh'): + """运行 Whisper 语音识别 + + Args: + whisper_path: whisper.cpp 目录 + audio_path: WAV 音频文件路径 + model: 模型名称 + output_dir: 输出目录 + language: 语言代码 + """ + # 输出文件名（使用原始输入文件名，不是 audio.wav） + # 从 output_dir 和 audio_path 推断原始文件名 + original_stem = Path(audio_path).stem.replace('_audio', '') + output_basename = original_stem # 构建命令 cmd = [ - os.path.join(whisper_path, 'main'), + os.path.join(whisper_path, 'whisper-cli'), '-m', os.path.join(whisper_path, 'models', f'ggml-{model}.bin'), - '-f', input_file, - '-of', os.path.join(output_dir, Path(input_file).stem), - '-otxt', # 输出 txt - '-osrt', # 输出 srt - '-oj', # 输出 json + '-f', audio_path, + '-of', os.path.join(output_dir, output_basename), + '-l', language, + '--output-txt', + '--output-srt', + '--output-json', ] - # 添加语言参数（如果指定） - if language: - cmd.extend(['-l', language]) - print(f"执行命令: {' '.join(cmd)}") - # 运行命令 + # 运行 result = subprocess.run( cmd, capture_output=True, text=True, - cwd=whisper_path + cwd=whisper_path, + timeout=3600 # 1小时超时 ) if result.returncode != 0: - raise RuntimeError( - f"Whisper.cpp 执行失败:\n" - f"标准输出: {result.stdout}\n" - f"错误输出: {result.stderr}" - ) + # 检查是否有警告但仍成功 + if "error" in result.stderr.lower() and not os.path.exists( + os.path.join(output_dir, f"{output_basename}.txt") + ): + raise RuntimeError( + f"Whisper 执行失败:\n" + f"标准输出: {result.stdout}\n" + f"错误输出: {result.stderr}" + ) return result -def parse_whisper_json(json_file): - """解析 whisper.cpp 输出的 JSON 文件""" - try: - with open(json_file, 'r', encoding='utf-8') as f: - data = json.load(f) - return data - except json.JSONDecodeError: - # whisper.cpp 的 JSON 格式可能不同，尝试解析标准格式 - return None +def parse_srt_for_duration(srt_file): + """从 SRT 文件解析时长""" + if not os.path.exists(srt_file): + return 0 + + with open(srt_file, 'r', encoding='utf-8') as f: + content = f.read() + + # 找到最后一个时间戳 + pattern = r'(\d{2}:\d{2}:\d{2}),\d{3} --> \d{2}:\d{2}:\d{2},\d{3}' + matches = re.findall(pattern, content) + + if not matches: + return 0 + + # 解析最后一个时间戳 + last_time = matches[-1] + h, m, s = last_time.split(':') + return int(h) * 3600 + int(m) * 60 + int(s) -def create_standard_json(input_file, output_dir, srt_file, txt_file, model): - """创建标准格式的 JSON 输出""" - basename = Path(input_file).stem +def create_workflow_json(input_file, output_dir, model, language, duration): + """创建标准格式的 JSON 输出，供 KB Workflow 使用""" + input_path = Path(input_file) + basename = input_path.stem - # 解析 SRT 文件获取时间戳 + txt_file = os.path.join(output_dir, f"{basename}.txt") + srt_file = os.path.join(output_dir, f"{basename}.srt") + json_file = os.path.join(output_dir, f"{basename}.json") + + # 读取文本内容 + text_content = "" + if os.path.exists(txt_file): + with open(txt_file, 'r', encoding='utf-8') as f: + text_content = f.read().strip() + + # 解析 SRT 获取分段 segments = [] if os.path.exists(srt_file): with open(srt_file, 'r', encoding='utf-8') as f: content = f.read() - # 解析 SRT 格式 + # SRT 格式: 序号 -> 时间 -> 文本 pattern = r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3}) --> (\d{2}:\d{2}:\d{2},\d{3})\n(.+?)(?=\n\n|\Z)' matches = re.findall(pattern, content, re.DOTALL) - for idx, (num, start, end, text) in enumerate(matches): - # 转换时间戳 - def parse_time(t): - h, m, s_ms = t.split(':') - s, ms = s_ms.split(',') - return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000 + def parse_time(t): + h, m, s_ms = t.split(':') + s, ms = s_ms.split(',') + return float(h) * 3600 + float(m) * 60 + float(s) + float(ms) / 1000 + for idx, (num, start, end, text) in enumerate(matches): segments.append({ "id": idx, - "start": parse_time(start), - "end": parse_time(end), + "start": round(parse_time(start), 2), + "end": round(parse_time(end), 2), "text": text.strip() }) - # 读取纯文本 - txt_content = "" - if os.path.exists(txt_file): - with open(txt_file, 'r', encoding='utf-8') as f: - txt_content = f.read().strip() - - # 获取音频时长（使用 ffprobe） - duration = 0 - try: - result = subprocess.run( - ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', - '-of', 'default=noprint_wrappers=1:nokey=1', input_file], - capture_output=True, - text=True - ) - if result.returncode == 0: - duration = float(result.stdout.strip()) - except FileNotFoundError: - pass - - # 构建标准 JSON - json_data = { + # 构建 JSON + workflow_json = { "success": True, - "input_file": os.path.basename(input_file), - "input_path": input_file, + "tool": "whisper", "model": model, + "language": language, + "input_file": str(input_path), "output_files": { - "json": f"{basename}.json", + "txt": f"{basename}.txt", "srt": f"{basename}.srt", - "txt": f"{basename}.txt" + "json": f"{basename}.json" }, "duration_seconds": duration, - "segments": segments + "segments": segments, + "text": text_content } - return json_data + # 写入 JSON + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(workflow_json, f, ensure_ascii=False, indent=2) + + return workflow_json def main(): @@ -228,24 +300,22 @@ def main(): # 获取 whisper.cpp 路径 try: whisper_path = get_whisper_cpp_path() - print(f"Whisper.cpp 路径: {whisper_path}") + print(f"✓ Whisper.cpp 路径: {whisper_path}") except FileNotFoundError as e: - print(f"错误: {e}", file=sys.stderr) + print(f"✗ 错误: {e}", file=sys.stderr) sys.exit(1) # 检查模型 try: check_model(whisper_path, args.model) - print(f"使用模型: {args.model}") + print(f"✓ 使用模型: {args.model}") except FileNotFoundError as e: - print(f"错误: {e}", file=sys.stderr) + print(f"✗ 错误: {e}", file=sys.stderr) sys.exit(1) # 后台执行 if args.background: print("启动后台处理...") - - # 使用 nohup 在后台运行 log_file = os.path.join(output_dir, f"{Path(input_file).stem}_whisper.log") with open(log_file, 'w') as log: @@ -253,91 +323,76 @@ def main(): [sys.executable, __file__, '--input', input_file, '--model', args.model, - '--output', output_dir] + - (['--language', args.language] if args.language else []), + '--output', output_dir, + '--language', args.language], stdout=log, stderr=log, start_new_session=True ) - print(f"后台任务已启动，日志文件: {log_file}") - print(f"输出文件将生成到: {output_dir}") + print(f"✓ 后台任务已启动") + print(f" 日志: {log_file}") + print(f" 输出: {output_dir}") - # 返回任务信息 result = { "success": True, "background": True, "status": "running", "log_file": log_file, - "output_dir": output_dir, - "message": "后台处理中，请稍候检查输出文件" + "output_dir": output_dir } - print(json.dumps(result, ensure_ascii=False, indent=2)) return # 前台执行 - print(f"处理文件: {input_file}") - print(f"输出目录: {output_dir}") + print(f"\n📁 处理文件: {input_file}") + print(f"📁 输出目录: {output_dir}") try: - # 运行 whisper.cpp - result = run_whisper_cpp( - whisper_path, - input_file, - args.model, - output_dir, - args.language - ) + # 1. 转换为 WAV + print(f"\n🔄 步骤1: 转换音频格式...") + audio_path = convert_to_wav(input_file, output_dir) - # 输出文件路径 + # 2. 运行 Whisper + print(f"\n🎙️ 步骤2: 语音识别 (语言: {args.language})...") + run_whisper(whisper_path, audio_path, args.model, output_dir, args.language) + + # 3. 清理临时 WAV 文件 + if audio_path != input_file and audio_path.endswith('_audio.wav'): + try: + os.remove(audio_path) + print("✓ 清理临时文件") + except Exception: + pass + + # 4. 获取时长并创建 JSON + print(f"\n📝 步骤3: 生成输出文件...") basename = Path(input_file).stem - json_file = os.path.join(output_dir, f"{basename}.json") srt_file = os.path.join(output_dir, f"{basename}.srt") - txt_file = os.path.join(output_dir, f"{basename}.txt") + duration = parse_srt_for_duration(srt_file) - # 检查输出文件 - output_files = {} - if os.path.exists(srt_file): - output_files['srt'] = srt_file - if os.path.exists(txt_file): - output_files['txt'] = txt_file - - # 创建标准 JSON 格式 - standard_json = create_standard_json( - input_file, - output_dir, - srt_file, - txt_file, - args.model + workflow_json = create_workflow_json( + input_file, output_dir, args.model, args.language, duration ) - # 写入标准 JSON - with open(json_file, 'w', encoding='utf-8') as f: - json.dump(standard_json, f, ensure_ascii=False, indent=2) - - output_files['json'] = json_file - # 输出结果 - result = { - "success": True, - "input_file": input_file, - "model": args.model, - "output_files": { - k: os.path.basename(v) for k, v in output_files.items() - }, - "output_dir": output_dir, - "total_segments": len(standard_json.get('segments', [])), - "duration": standard_json.get('duration_seconds', 0) - } + print(f"\n✅ 处理完成!") + print(f" 时长: {duration} 秒") + print(f" 分段数: {len(workflow_json['segments'])}") + print(f"\n📄 输出文件:") + for fmt in ['txt', 'srt', 'json']: + fpath = os.path.join(output_dir, f"{basename}.{fmt}") + if os.path.exists(fpath): + size = os.path.getsize(fpath) + print(f" • {fmt.upper()}: {fpath} ({size:,} bytes)") - print("\n✓ 处理完成!") - print(f"\n输出文件:") - for fmt, fname in output_files.items(): - print(f" - {fname.upper()}: {fname}") - - print(f"\n完整结果:") - print(json.dumps(result, ensure_ascii=False, indent=2)) + print(f"\n📋 Workflow JSON:") + print(json.dumps({ + "success": workflow_json["success"], + "tool": "whisper", + "text_length": len(workflow_json["text"]), + "segments_count": len(workflow_json["segments"]) + }, ensure_ascii=False, indent=2)) except Exception as e: print(f"\n✗ 错误: {e}", file=sys.stderr) diff --git a/skills/kb-workflow/SKILL.md b/skills/kb-workflow/SKILL.md index a62dfc8..9d74533 100644 --- a/skills/kb-workflow/SKILL.md +++ b/skills/kb-workflow/SKILL.md @@ -108,28 +108,28 @@ python3 /path/to/bilibili_content.py --id "[ID]" --type subtitle --output [subti 3. 解析输出 JSON，提取 video_filename 和 subtitle_filename（如有） -### 步骤 5: OCR 处理(禁止跳过) +### 步骤 5: 语音识别转文字(禁止跳过) -> ⚠️ **重要**：此步骤为耗时步骤，视频按帧截取和 OCR 识别需要较长时间。 -> - **严禁跳过**：即使等待时间较长，也必须等待 OCR 完成 -> - **定时汇报**：每2-3分钟向用户汇报进度（处理帧数/预计剩余时间） +> ⚠️ **重要**：此步骤为耗时步骤，使用 Whisper 进行语音识别。 +> - **严禁跳过**：即使等待时间较长，也必须等待识别完成 +> - **定时汇报**：每2-3分钟向用户汇报进度 > - **超时时间**：30分钟 > - **耐心等待**：不要中途启动其他步骤，必须串行执行 -如果步骤4没有获取到字幕，需要从视频截帧 OCR： +如果步骤4没有获取到字幕，需要使用 Whisper 进行语音识别： 1. 获取目录路径： ```bash python3 kb_workflow.py --content ``` -2. 启动子代理调用 kb-ocr-processor（传入完整路径）： +2. 启动子代理调用 kb-whisper（传入完整路径）： ```bash # 必须传入 --output 完整路径 -python3 /path/to/ocr_processor.py --video [video_path] --output [content_dir]/ocr.json --merge +python3 /path/to/whisper.py --input [video_path] --model base --output [content_dir] ``` -3. 等待 OCR 完成（定时汇报进度） +3. 等待 Whisper 完成（定时汇报进度） 4. 解析输出 JSON，生成 content 文件供步骤6使用 @@ -173,7 +173,7 @@ python3 kb_workflow.py --archive-file "[分析报告文件路径]" | 3 | kb-bilibili-metadata | `--id "BVxxx" --type video` | `{p}_{id}.meta.json` | | 3 | kb-zhihu-metadata | `--id "xxx" --type question` | `{p}_{id}.meta.json` | | 4 | kb-bilibili-content | `--id "BVxxx" --type video` | `{p}_{id}.{ext}` | -| 5 | kb-ocr-processor | `--video xxx.flv --output ocr.json` | `ocr.json`, `ocr.txt` | +| 5 | kb-whisper | `--input xxx.mp4 --model base` | `{p}_{id}.json`, `{p}_{id}.txt` | | 6 | kb-deep-analyzer | `--metadata x.meta.json --content x.json --output x.md` | `{p}_{id}.md` | ## 目录结构 @@ -189,9 +189,8 @@ python3 kb_workflow.py --archive-file "[分析报告文件路径]" ├── content/ │ ├── downloads/ │ │ └── {platform}_{id}.mp4 -│ ├── subtitles/ -│ │ └── {platform}_{id}.srt -│ └── {platform}_{id}.txt +│ ├── {platform}_{id}.txt # Whisper 识别的文本 +│ └── {platform}_{id}.json # Whisper 完整结果 └── analyzed/ └── {platform}_{id}.md ``` diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index 72a34dd..693ed4e 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -67,7 +67,6 @@ def ensure_directories(base_path: Path = None) -> Dict[str, Path]: "metadata": base_path / "metadata", "content": base_path / "content", "content_downloads": base_path / "content" / "downloads", - "content_subtitles": base_path / "content" / "subtitles", "analyzed": base_path / "analyzed", "archive": base_path / "archive", "archive_subscriptions": base_path / "archive" / "subscriptions", @@ -250,7 +249,6 @@ def main(): parser.add_argument('--metadata', action='store_true', help='获取元数据目录') parser.add_argument('--content', action='store_true', help='获取内容目录') parser.add_argument('--content_downloads', action='store_true', help='获取主内容下载目录') - parser.add_argument('--content_subtitles', action='store_true', help='获取字幕目录') parser.add_argument('--analyzed', action='store_true', help='获取深度分析目录') parser.add_argument('--archive_dir', action='store_true', help='获取归档目录') parser.add_argument('--archive_subscriptions', action='store_true', help='获取订阅信息目录') @@ -280,9 +278,6 @@ def main(): elif args.content_downloads: print(json.dumps({"path": str(dirs["content_downloads"])}, ensure_ascii=False, indent=2)) - elif args.content_subtitles: - print(json.dumps({"path": str(dirs["content_subtitles"])}, ensure_ascii=False, indent=2)) - elif args.analyzed: print(json.dumps({"path": str(dirs["analyzed"])}, ensure_ascii=False, indent=2))