feat: in-app Whisper (faster-whisper) — opt-in install + model management

Videos without embedded subtitles (e.g. B站 without AI字幕) need Whisper to
generate subtitles, but bundling it would bloat every install. Instead let
users install it on demand from Settings → 语音识别, and pick which model.

Backend uses faster-whisper (CTranslate2, no PyTorch, ~214MB installed, several
times faster than openai-whisper, cross-platform) — chosen over mlx-whisper,
which hard-depends on torch (~2-3GB).

- whisper_runtime.py (new): pip-install faster-whisper into a user-writable dir
  (<data>/whisper-runtime) using the bundled Python; add to sys.path; status +
  coarse progress; uninstall. Never writes into the signed .app bundle.
- whisper_model_manager.py: tiny→large-v3 from Systran/faster-whisper-*,
  background download via huggingface_hub, real status; cache under
  <data>/whisper-models.
- speech_recognizer.py: subtitle generation rewritten from the `whisper` CLI to
  faster-whisper's WhisperModel API → SRT; availability = runtime installed.
- speech_recognition.py API: /whisper/install, /whisper/uninstall,
  /whisper/runtime-status (+ existing /whisper-models*).
- SpeechRecognitionConfig.tsx: was a stub; now a full UI (install button +
  progress + log, model list with download/delete/status) wired to a new
  speechApi in services/api.ts.
- build_macos_arm.sh: allowlist faster_whisper/ctranslate2/huggingface_hub in
  the dependency guard (they're installed at runtime, imported lazily).

Verified end-to-end: install runtime → download tiny model → transcribe a real
video into a valid SRT, all through the API/runtime.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
周小舟
2026-05-31 01:06:26 +08:00
parent 46163d94e7
commit 6e59f40f01
7 changed files with 686 additions and 423 deletions

View File

@@ -40,6 +40,36 @@ def get_speech_recognizer() -> SpeechRecognizer:
_speech_recognizer = SpeechRecognizer()
return _speech_recognizer
# ===== Whisper 运行时(按需安装)=====
@router.get("/whisper/runtime-status")
async def whisper_runtime_status():
"""Whisper 运行时安装状态(前端轮询)。"""
from backend.services import whisper_runtime
return whisper_runtime.get_status()
@router.post("/whisper/install")
async def whisper_install():
"""开始在后台安装 Whisper 运行时mlx-whisper"""
from backend.services import whisper_runtime
if sys_is_not_darwin():
raise HTTPException(status_code=400, detail="mlx-whisper 仅支持 Apple Silicon (macOS)")
return whisper_runtime.start_install()
@router.post("/whisper/uninstall")
async def whisper_uninstall():
"""卸载 Whisper 运行时(不影响已下载的模型缓存可单独删除)。"""
from backend.services import whisper_runtime
return whisper_runtime.uninstall()
def sys_is_not_darwin() -> bool:
import sys
return sys.platform != "darwin"
class SpeechConfigRequest(BaseModel):
"""语音识别配置请求"""
method: str

View File

@@ -1,34 +1,32 @@
"""
Whisper模型管理服务
负责模型的下载、状态检查、存储管理等功能
Whisper 模型管理服务mlx-whisper
负责 mlx-community Whisper 模型的下载、状态检查、删除。模型从 HuggingFace 拉取,
统一缓存到 `<data_dir>/whisper-models`(由 whisper_runtime 设置 HF_HOME
依赖huggingface_hub来自运行时安装目录所有相关 import 都延迟到函数内。
"""
import os
import json
import logging
import subprocess
import asyncio
from typing import Dict, List, Optional, Tuple
import threading
from typing import Dict, List, Optional
from pathlib import Path
from dataclasses import dataclass
from enum import Enum
import requests
from concurrent.futures import ThreadPoolExecutor
from . import whisper_runtime
logger = logging.getLogger(__name__)
class ModelStatus(str, Enum):
"""模型状态枚举"""
AVAILABLE = "available" # 可用
AVAILABLE = "available" # 运行时就绪、可下载
DOWNLOADING = "downloading" # 下载中
DOWNLOADED = "downloaded" # 已下载
ERROR = "error" # 错误
NOT_FOUND = "not_found" # 未找到
DOWNLOADED = "downloaded" # 已下载
ERROR = "error" # 错误(通常是运行时未安装)
NOT_FOUND = "not_found"
@dataclass
class ModelInfo:
"""模型信息"""
name: str
size: str
size_bytes: int
@@ -36,303 +34,174 @@ class ModelInfo:
accuracy: str
speed: str
status: ModelStatus
repo_id: str = ""
download_progress: Optional[int] = None
local_path: Optional[str] = None
error_message: Optional[str] = None
# 模型名 -> HuggingFace 仓库 + 展示信息faster-whisper / CTranslate2 模型)
_MODELS = {
"tiny": {
"repo_id": "Systran/faster-whisper-tiny",
"size": "~75 MB", "size_bytes": 75 * 1024 * 1024,
"description": "最快,准确度较低,适合快速预览", "accuracy": "较低", "speed": "最快",
},
"base": {
"repo_id": "Systran/faster-whisper-base",
"size": "~145 MB", "size_bytes": 145 * 1024 * 1024,
"description": "平衡之选,推荐日常使用", "accuracy": "中等", "speed": "",
},
"small": {
"repo_id": "Systran/faster-whisper-small",
"size": "~488 MB", "size_bytes": 488 * 1024 * 1024,
"description": "较好准确度,适合重要内容", "accuracy": "较好", "speed": "中等",
},
"medium": {
"repo_id": "Systran/faster-whisper-medium",
"size": "~1.5 GB", "size_bytes": 1500 * 1024 * 1024,
"description": "高准确度,适合专业用途", "accuracy": "", "speed": "较慢",
},
"large-v3": {
"repo_id": "Systran/faster-whisper-large-v3",
"size": "~3 GB", "size_bytes": 3000 * 1024 * 1024,
"description": "最高准确度", "accuracy": "最高", "speed": "最慢",
},
}
def repo_id_for(model_name: str) -> Optional[str]:
cfg = _MODELS.get(model_name)
return cfg["repo_id"] if cfg else None
class WhisperModelManager:
"""Whisper模型管理器"""
def __init__(self, models_dir: Optional[Path] = None):
self.models_dir = models_dir or self._get_default_models_dir()
self.models_dir.mkdir(parents=True, exist_ok=True)
self.download_tasks: Dict[str, asyncio.Task] = {}
# 模型信息配置
self.model_configs = {
"tiny": {
"size": "39 MB",
"size_bytes": 39 * 1024 * 1024,
"description": "最快速度,适合实时处理",
"accuracy": "较低",
"speed": "最快"
},
"base": {
"size": "74 MB",
"size_bytes": 74 * 1024 * 1024,
"description": "平衡选择,推荐日常使用",
"accuracy": "中等",
"speed": ""
},
"small": {
"size": "244 MB",
"size_bytes": 244 * 1024 * 1024,
"description": "较好准确度,适合重要内容",
"accuracy": "较好",
"speed": "中等"
},
"medium": {
"size": "769 MB",
"size_bytes": 769 * 1024 * 1024,
"description": "高准确度,适合专业用途",
"accuracy": "",
"speed": "较慢"
},
"large": {
"size": "1550 MB",
"size_bytes": 1550 * 1024 * 1024,
"description": "最高准确度,适合重要项目",
"accuracy": "最高",
"speed": "最慢"
}
}
def _get_default_models_dir(self) -> Path:
"""获取默认模型目录"""
from backend.core.desktop_config import get_desktop_data_dir
return get_desktop_data_dir() / "whisper_models"
def get_all_models_info(self) -> List[ModelInfo]:
"""获取所有模型信息"""
models = []
for model_name, config in self.model_configs.items():
status = self._check_model_status(model_name)
local_path = self._get_model_path(model_name) if status == ModelStatus.DOWNLOADED else None
models.append(ModelInfo(
name=model_name,
size=config["size"],
size_bytes=config["size_bytes"],
description=config["description"],
accuracy=config["accuracy"],
speed=config["speed"],
status=status,
local_path=local_path
))
return models
def get_model_info(self, model_name: str) -> Optional[ModelInfo]:
"""获取指定模型信息"""
if model_name not in self.model_configs:
return None
config = self.model_configs[model_name]
def __init__(self):
self.model_configs = _MODELS
# model_name -> {"status","progress","error"}
self._download_state: Dict[str, Dict] = {}
self._lock = threading.Lock()
# ---- 路径 / 状态 ----
def _model_cache_dir(self, model_name: str) -> Path:
repo = self.model_configs[model_name]["repo_id"]
# HF 缓存目录命名models--<org>--<name>
return whisper_runtime.get_models_dir() / "hub" / ("models--" + repo.replace("/", "--"))
def _is_downloaded(self, model_name: str) -> bool:
d = self._model_cache_dir(model_name)
snaps = d / "snapshots"
return snaps.exists() and any(snaps.iterdir())
def _check_model_status(self, model_name: str) -> ModelStatus:
with self._lock:
st = self._download_state.get(model_name)
if st and st.get("status") == "downloading":
return ModelStatus.DOWNLOADING
if st and st.get("status") == "error":
return ModelStatus.ERROR
if self._is_downloaded(model_name):
return ModelStatus.DOWNLOADED
if not whisper_runtime.is_installed():
return ModelStatus.ERROR # 运行时没装,模型也用不了
return ModelStatus.AVAILABLE
def _info(self, model_name: str) -> ModelInfo:
cfg = self.model_configs[model_name]
status = self._check_model_status(model_name)
local_path = self._get_model_path(model_name) if status == ModelStatus.DOWNLOADED else None
with self._lock:
st = self._download_state.get(model_name, {})
return ModelInfo(
name=model_name,
size=config["size"],
size_bytes=config["size_bytes"],
description=config["description"],
accuracy=config["accuracy"],
speed=config["speed"],
status=status,
local_path=local_path
size=cfg["size"], size_bytes=cfg["size_bytes"],
description=cfg["description"], accuracy=cfg["accuracy"], speed=cfg["speed"],
status=status, repo_id=cfg["repo_id"],
download_progress=st.get("progress"),
local_path=str(self._model_cache_dir(model_name)) if status == ModelStatus.DOWNLOADED else None,
error_message=st.get("error"),
)
def _check_model_status(self, model_name: str) -> ModelStatus:
"""检查模型状态"""
try:
# 检查是否正在下载
if model_name in self.download_tasks and not self.download_tasks[model_name].done():
return ModelStatus.DOWNLOADING
# 检查本地文件是否存在
model_path = self._get_model_path(model_name)
if model_path.exists():
return ModelStatus.DOWNLOADED
# 检查whisper是否可用
if not self._check_whisper_available():
return ModelStatus.ERROR
return ModelStatus.AVAILABLE
except Exception as e:
logger.error(f"检查模型状态失败: {e}")
return ModelStatus.ERROR
def _get_model_path(self, model_name: str) -> Path:
"""获取模型文件路径"""
# Whisper模型通常存储在 ~/.cache/whisper/ 目录
home_dir = Path.home()
whisper_cache = home_dir / ".cache" / "whisper"
# 检查常见的模型文件扩展名
possible_files = [
f"{model_name}.pt",
f"{model_name}.bin",
f"{model_name}.model"
]
for filename in possible_files:
model_file = whisper_cache / filename
if model_file.exists():
return model_file
# 如果没找到,返回默认路径
return whisper_cache / f"{model_name}.pt"
def _check_whisper_available(self) -> bool:
"""检查Whisper是否可用"""
try:
result = subprocess.run(['whisper', '--help'],
capture_output=True, text=True, timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def get_all_models_info(self) -> List[ModelInfo]:
return [self._info(name) for name in self.model_configs]
def get_model_info(self, model_name: str) -> Optional[ModelInfo]:
if model_name not in self.model_configs:
return None
return self._info(model_name)
# ---- 下载(后台线程,非阻塞)----
async def download_model(self, model_name: str) -> bool:
"""下载模型"""
if model_name not in self.model_configs:
raise ValueError(f"不支持的模型: {model_name}")
if model_name in self.download_tasks and not self.download_tasks[model_name].done():
raise ValueError(f"模型 {model_name} 正在下载中")
if self._check_model_status(model_name) == ModelStatus.DOWNLOADED:
logger.info(f"模型 {model_name} 已存在")
if not whisper_runtime.is_installed():
raise RuntimeError("请先安装 Whisper 运行时")
if self._is_downloaded(model_name):
return True
try:
# 创建下载任务
task = asyncio.create_task(self._download_model_async(model_name))
self.download_tasks[model_name] = task
result = await task
return result
except Exception as e:
logger.error(f"下载模型 {model_name} 失败: {e}")
return False
finally:
# 清理任务
if model_name in self.download_tasks:
del self.download_tasks[model_name]
async def _download_model_async(self, model_name: str) -> bool:
"""异步下载模型"""
try:
logger.info(f"开始下载模型: {model_name}")
# 使用whisper命令下载模型
cmd = ['whisper', '--model', model_name, '--help']
# 在后台线程中执行下载
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
result = await loop.run_in_executor(
executor,
self._run_whisper_download,
model_name
)
if result:
logger.info(f"模型 {model_name} 下载完成")
with self._lock:
st = self._download_state.get(model_name)
if st and st.get("status") == "downloading":
return True
else:
logger.error(f"模型 {model_name} 下载失败")
return False
except Exception as e:
logger.error(f"下载模型 {model_name} 时发生错误: {e}")
return False
def _run_whisper_download(self, model_name: str) -> bool:
"""运行whisper下载命令"""
self._download_state[model_name] = {"status": "downloading", "progress": 0, "error": None}
threading.Thread(
target=self._download_blocking, args=(model_name,),
name=f"whisper-dl-{model_name}", daemon=True,
).start()
return True
def _download_blocking(self, model_name: str) -> None:
repo_id = self.model_configs[model_name]["repo_id"]
try:
# 使用whisper的模型下载功能
# 这里我们通过运行一个简单的命令来触发模型下载
cmd = ['python', '-c', f'import whisper; whisper.load_model("{model_name}")']
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=3600 # 1小时超时
whisper_runtime.ensure_on_path()
from huggingface_hub import snapshot_download
logger.info(f"开始下载 Whisper 模型 {model_name} ({repo_id})")
snapshot_download(
repo_id=repo_id,
cache_dir=str(whisper_runtime.get_models_dir() / "hub"),
)
return result.returncode == 0
except subprocess.TimeoutExpired:
logger.error(f"模型 {model_name} 下载超时")
return False
except Exception as e:
logger.error(f"运行whisper下载命令失败: {e}")
return False
with self._lock:
self._download_state[model_name] = {"status": "downloaded", "progress": 100, "error": None}
logger.info(f"Whisper 模型 {model_name} 下载完成")
except Exception as e: # noqa: BLE001
logger.error(f"下载 Whisper 模型 {model_name} 失败: {e}", exc_info=True)
with self._lock:
self._download_state[model_name] = {"status": "error", "progress": 0, "error": str(e)}
def get_download_progress(self, model_name: str) -> Optional[int]:
"""获取下载进度"""
if model_name not in self.download_tasks:
return None
task = self.download_tasks[model_name]
if task.done():
return 100
# 这里可以实现更精确的进度跟踪
# 目前返回一个估算值
return 50 # 占位符
with self._lock:
st = self._download_state.get(model_name)
if not st:
return 100 if self._is_downloaded(model_name) else None
return st.get("progress")
def cancel_download(self, model_name: str) -> bool:
"""取消下载"""
if model_name not in self.download_tasks:
return False
task = self.download_tasks[model_name]
if not task.done():
task.cancel()
del self.download_tasks[model_name]
logger.info(f"已取消模型 {model_name} 的下载")
return True
return False
def delete_model(self, model_name: str) -> bool:
"""删除模型"""
try:
model_path = self._get_model_path(model_name)
if not model_path.exists():
logger.warning(f"模型文件不存在: {model_path}")
# snapshot_download 不易中断;这里只清状态,已下载分片保留可续传
with self._lock:
if model_name in self._download_state and self._download_state[model_name].get("status") == "downloading":
self._download_state[model_name] = {"status": "available", "progress": 0, "error": None}
return True
# 删除模型文件
model_path.unlink()
logger.info(f"模型 {model_name} 已删除")
return True
except Exception as e:
logger.error(f"删除模型 {model_name} 失败: {e}")
return False
def delete_model(self, model_name: str) -> bool:
if model_name not in self.model_configs:
return False
def set_models_directory(self, directory: str) -> bool:
"""设置模型存储目录"""
try:
new_dir = Path(directory)
new_dir.mkdir(parents=True, exist_ok=True)
# 更新模型目录
self.models_dir = new_dir
# 这里可以添加将现有模型移动到新目录的逻辑
logger.info(f"模型目录已设置为: {new_dir}")
import shutil
d = self._model_cache_dir(model_name)
if d.exists():
shutil.rmtree(d, ignore_errors=True)
with self._lock:
self._download_state.pop(model_name, None)
logger.info(f"Whisper 模型 {model_name} 已删除")
return True
except Exception as e:
logger.error(f"设置模型目录失败: {e}")
except Exception as e: # noqa: BLE001
logger.error(f"删除 Whisper 模型 {model_name} 失败: {e}")
return False
# 全局模型管理器实例
_model_manager: Optional[WhisperModelManager] = None
def get_model_manager() -> WhisperModelManager:
"""获取模型管理器实例"""
global _model_manager
if _model_manager is None:
_model_manager = WhisperModelManager()

View File

@@ -0,0 +1,188 @@
"""
Whisper 运行时管理(桌面模式,按需安装)
桌面安装包默认不带 Whisper运行时 + 模型有体积,没必要让所有用户都背)。用户在设置页
里可以自己决定是否安装、以及下载哪个模型。后端用 faster-whisperCTranslate2不依赖
PyTorch运行时 ~200-400MB比官方 whisper 快数倍,跨平台)。
设计要点:
- 安装到「用户可写目录」`<data_dir>/whisper-runtime`,而不是 .app 包内
/Applications 通常只读,且写入会破坏代码签名)。
- 用「当前正在跑后端的便携 Python」(sys.executable) 的 pip 安装,保证解释器一致。
- 模型缓存放 `<data_dir>/whisper-models`(通过 HF_HOME 收口)。
- 所有对 mlx_whisper / huggingface_hub 的 import 都延迟到函数内部,避免构建期
依赖扫描把它们当成缺失依赖而让打包失败。
"""
import os
import sys
import shutil
import logging
import threading
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional
logger = logging.getLogger(__name__)
# 要安装的运行时包faster-whisper 带上 ctranslate2、onnxruntime、av、huggingface_hub 等,
# 不含 PyTorch
WHISPER_PACKAGES = ["faster-whisper"]
# 运行时核心模块(用于探测是否已装)
WHISPER_IMPORT_NAME = "faster_whisper"
def _data_dir() -> Path:
try:
from backend.core.desktop_config import get_desktop_data_dir
return Path(get_desktop_data_dir())
except Exception:
return Path(os.getenv("AUTOCLIP_DATA_DIR", str(Path.home() / "Library/Application Support/AutoClip")))
def get_install_dir() -> Path:
d = _data_dir() / "whisper-runtime"
d.mkdir(parents=True, exist_ok=True)
return d
def get_models_dir() -> Path:
d = _data_dir() / "whisper-models"
d.mkdir(parents=True, exist_ok=True)
return d
def ensure_on_path() -> None:
"""把运行时目录加入 sys.path并把模型缓存目录收口到 HF_HOME。"""
install_dir = str(get_install_dir())
if install_dir not in sys.path:
sys.path.insert(0, install_dir)
# 模型统一缓存到数据目录,便于管理/卸载
os.environ.setdefault("HF_HOME", str(get_models_dir()))
# mlx-whisper 解码音频要用 ffmpeg把内置 ffmpeg 所在目录并入 PATH
ffmpeg_path = os.getenv("AUTOCLIP_FFMPEG_PATH")
if ffmpeg_path:
ffmpeg_dir = str(Path(ffmpeg_path).parent)
if ffmpeg_dir not in os.environ.get("PATH", "").split(os.pathsep):
os.environ["PATH"] = ffmpeg_dir + os.pathsep + os.environ.get("PATH", "")
def is_installed() -> bool:
"""运行时是否已就绪mlx_whisper 可被导入)。"""
ensure_on_path()
try:
import importlib.util
return importlib.util.find_spec(WHISPER_IMPORT_NAME) is not None
except Exception:
return False
# ---- 安装状态(供前端轮询)----
_state_lock = threading.Lock()
_state: Dict[str, Any] = {
"status": "unknown", # not_installed | installing | installed | error
"progress": 0, # 粗粒度百分比
"message": "",
"log_tail": "",
}
def _set_state(**kw) -> None:
with _state_lock:
_state.update(kw)
def get_status() -> Dict[str, Any]:
with _state_lock:
st = dict(_state)
# 没在安装时,用实际探测结果覆盖
if st["status"] not in ("installing",):
st["status"] = "installed" if is_installed() else "not_installed"
if st["status"] == "installed":
st["progress"] = 100
st["platform_supported"] = True # faster-whisper 跨平台
st["packages"] = WHISPER_PACKAGES
return st
def _do_install(index_url: Optional[str]) -> None:
install_dir = get_install_dir()
cmd = [
sys.executable, "-m", "pip", "install",
"--upgrade",
"--target", str(install_dir),
*WHISPER_PACKAGES,
]
if index_url:
cmd += ["--index-url", index_url]
logger.info(f"开始安装 Whisper 运行时: {' '.join(cmd)}")
_set_state(status="installing", progress=5, message="正在准备安装…", log_tail="")
try:
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, bufsize=1,
)
lines: list[str] = []
for line in iter(proc.stdout.readline, ""):
line = line.rstrip()
if not line:
continue
lines.append(line)
lines[:] = lines[-40:]
# 粗粒度进度:根据 pip 的阶段词推进,纯属观感
low = line.lower()
if low.startswith("collecting") or "downloading" in low:
_bump_progress(min_v=10, max_v=70, message=line)
elif "installing collected packages" in low or "building" in low:
_bump_progress(min_v=70, max_v=95, message="正在安装依赖…")
_set_state(log_tail="\n".join(lines[-12:]))
proc.wait()
if proc.returncode == 0 and is_installed():
_set_state(status="installed", progress=100, message="安装完成")
logger.info("Whisper 运行时安装完成")
else:
_set_state(status="error", message=f"安装失败pip 退出码 {proc.returncode}")
logger.error(f"Whisper 运行时安装失败pip 退出码 {proc.returncode}")
except Exception as e: # noqa: BLE001
logger.error(f"安装 Whisper 运行时异常: {e}", exc_info=True)
_set_state(status="error", message=f"安装异常: {e}")
def _bump_progress(min_v: int, max_v: int, message: str) -> None:
with _state_lock:
cur = _state.get("progress", 0)
_state["progress"] = max(min_v, min(max_v, cur + 2))
_state["message"] = message
def start_install(index_url: Optional[str] = None) -> Dict[str, Any]:
with _state_lock:
if _state["status"] == "installing":
return {"started": False, "message": "正在安装中"}
if is_installed():
_set_state(status="installed", progress=100, message="已安装")
return {"started": False, "message": "已安装"}
# 默认走环境变量里的 pip 源(构建脚本/桌面默认清华),否则 PyPI
idx = index_url or os.getenv("PIP_INDEX_URL")
threading.Thread(target=_do_install, args=(idx,), name="whisper-install", daemon=True).start()
return {"started": True, "message": "已开始安装"}
def uninstall() -> Dict[str, Any]:
with _state_lock:
if _state["status"] == "installing":
return {"success": False, "message": "正在安装中,无法卸载"}
install_dir = get_install_dir()
try:
shutil.rmtree(install_dir, ignore_errors=True)
# 从 sys.modules 里剔除,避免本进程仍能 import
for mod in [m for m in list(sys.modules) if m.startswith("faster_whisper") or m.startswith("ctranslate2")]:
sys.modules.pop(mod, None)
p = str(install_dir)
if p in sys.path:
sys.path.remove(p)
_set_state(status="not_installed", progress=0, message="已卸载")
return {"success": True, "message": "已卸载 Whisper 运行时"}
except Exception as e: # noqa: BLE001
logger.error(f"卸载 Whisper 运行时失败: {e}")
return {"success": False, "message": str(e)}

View File

@@ -151,12 +151,11 @@ class SpeechRecognizer:
return methods
def _check_whisper_availability(self) -> bool:
"""检查本地Whisper是否可用"""
"""检查本地 Whisper(mlx) 运行时是否已安装。"""
try:
result = subprocess.run(['whisper', '--help'],
capture_output=True, text=True, timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
from backend.services import whisper_runtime
return whisper_runtime.is_installed()
except Exception:
logger.warning("本地Whisper未安装或不可用")
return False
@@ -330,136 +329,78 @@ class SpeechRecognizer:
return False
return False
def _generate_subtitle_whisper_local(self, video_path: Path, output_path: Path,
@staticmethod
def _format_srt_timestamp(seconds: float) -> str:
if seconds is None or seconds < 0:
seconds = 0.0
ms = int(round(seconds * 1000.0))
h, ms = divmod(ms, 3600000)
m, ms = divmod(ms, 60000)
s, ms = divmod(ms, 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
@classmethod
def _segments_to_srt(cls, segments: List[Dict[str, Any]]) -> str:
lines = []
for i, seg in enumerate(segments, start=1):
text = (seg.get("text") or "").strip()
if not text:
continue
start = cls._format_srt_timestamp(seg.get("start", 0.0))
end = cls._format_srt_timestamp(seg.get("end", 0.0))
lines.append(f"{i}\n{start} --> {end}\n{text}\n")
return "\n".join(lines) + "\n"
def _generate_subtitle_whisper_local(self, video_path: Path, output_path: Path,
config: SpeechRecognitionConfig) -> Path:
"""使用本地Whisper生成字幕"""
if not self.available_methods[SpeechRecognitionMethod.WHISPER_LOCAL]:
"""使用本地 faster-whisper 生成字幕(桌面按需安装的运行时)。"""
from backend.services import whisper_runtime
if not whisper_runtime.is_installed():
raise SpeechRecognitionError(
"本地Whisper不可用请安装whisper: pip install openai-whisper\n"
"同时确保已安装ffmpeg:\n"
" macOS: brew install ffmpeg\n"
" Ubuntu: sudo apt install ffmpeg\n"
" Windows: 下载ffmpeg并添加到PATH"
"本地 Whisper 运行时未安装。请到「设置 → 语音识别」里点击安装 Whisper"
"并下载一个模型后再试。"
)
if not video_path.exists():
raise SpeechRecognitionError(f"视频文件不存在: {video_path}")
if video_path.stat().st_size == 0:
raise SpeechRecognitionError(f"视频文件为空: {video_path}")
if output_path.exists():
logger.info(f"字幕文件已存在跳过Whisper处理: {output_path}")
return output_path
try:
logger.info(f"开始使用本地Whisper生成字幕: {video_path}")
# 检查视频文件是否存在
if not video_path.exists():
raise SpeechRecognitionError(f"视频文件不存在: {video_path}")
# 检查视频文件大小
file_size = video_path.stat().st_size
if file_size == 0:
raise SpeechRecognitionError(f"视频文件为空: {video_path}")
# 检查输出文件是否已存在,避免重复处理
if output_path.exists():
logger.info(f"字幕文件已存在跳过Whisper处理: {output_path}")
return output_path
# 构建whisper命令
cmd = [
'whisper',
str(video_path),
'--output_dir', str(output_path.parent),
'--output_format', config.output_format,
'--model', config.model
]
# 添加语言参数
if config.language != LanguageCode.AUTO:
cmd.extend(['--language', config.language])
# 添加超时处理
logger.info(f"执行Whisper命令: {' '.join(cmd)}")
# 使用进程锁文件防止重复处理
lock_file = output_path.parent / f".{video_path.stem}.whisper.lock"
try:
# 创建锁文件
with open(lock_file, 'w') as f:
f.write(str(os.getpid()))
# 根据超时配置决定是否设置超时
if config.timeout > 0:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=config.timeout,
cwd=str(video_path.parent) # 设置工作目录
)
else:
# 无超时限制
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=str(video_path.parent) # 设置工作目录
)
finally:
# 清理锁文件
if lock_file.exists():
lock_file.unlink()
if result.returncode == 0:
# 检查输出文件是否存在
if output_path.exists():
logger.info(f"本地Whisper字幕生成成功: {output_path}")
return output_path
else:
# 尝试查找其他可能的输出文件
possible_outputs = list(output_path.parent.glob(f"{video_path.stem}*.{config.output_format}"))
if possible_outputs:
actual_output = possible_outputs[0]
logger.info(f"找到Whisper输出文件: {actual_output}")
return actual_output
else:
raise SpeechRecognitionError(f"Whisper执行成功但未找到输出文件: {output_path}")
else:
error_msg = f"本地Whisper执行失败 (返回码: {result.returncode}):\n"
if result.stderr:
error_msg += f"错误信息: {result.stderr}\n"
if result.stdout:
error_msg += f"输出信息: {result.stdout}"
# 提供具体的错误解决建议
if "command not found" in result.stderr:
error_msg += "\n\n解决方案: 请安装whisper: pip install openai-whisper"
elif "ffmpeg" in result.stderr.lower():
error_msg += "\n\n解决方案: 请安装ffmpeg:\n macOS: brew install ffmpeg\n Ubuntu: sudo apt install ffmpeg"
elif "timeout" in result.stderr.lower():
error_msg += f"\n\n解决方案: 视频处理超时,请尝试使用更小的模型 (--model tiny) 或增加超时时间"
logger.error(error_msg)
raise SpeechRecognitionError(error_msg)
except subprocess.TimeoutExpired:
error_msg = f"本地Whisper执行超时{config.timeout}秒)\n"
error_msg += "解决方案:\n"
error_msg += "1. 使用更小的模型: --model tiny\n"
error_msg += "2. 增加超时时间\n"
error_msg += "3. 检查视频文件是否损坏"
logger.error(error_msg)
raise SpeechRecognitionError(error_msg)
except FileNotFoundError:
error_msg = "找不到whisper命令\n"
error_msg += "解决方案:\n"
error_msg += "1. 安装whisper: pip install openai-whisper\n"
error_msg += "2. 确保whisper在PATH中: which whisper\n"
error_msg += "3. 重新安装: pip uninstall openai-whisper && pip install openai-whisper"
logger.error(error_msg)
raise SpeechRecognitionError(error_msg)
except Exception as e:
error_msg = f"本地Whisper生成字幕时发生错误: {e}\n"
error_msg += "请检查:\n"
error_msg += "1. 视频文件格式是否支持\n"
error_msg += "2. 系统是否有足够的内存\n"
error_msg += "3. 是否有足够的磁盘空间"
logger.error(error_msg)
raise SpeechRecognitionError(error_msg)
whisper_runtime.ensure_on_path() # 让 faster_whisper 可导入
from faster_whisper import WhisperModel # 延迟导入:运行时安装目录里的包
language = None if config.language == LanguageCode.AUTO else str(config.language).split("-")[0]
models_dir = str(whisper_runtime.get_models_dir() / "hub")
logger.info(f"使用 faster-whisper 生成字幕: model={config.model} lang={language or 'auto'}")
# device=autoMac 上走 CPUCTranslate2int8 量化兼顾速度与体积
model = WhisperModel(
config.model, device="auto", compute_type="int8", download_root=models_dir,
)
seg_iter, _info = model.transcribe(str(video_path), language=language, vad_filter=True)
segments = [{"start": s.start, "end": s.end, "text": s.text} for s in seg_iter]
if not segments:
raise SpeechRecognitionError("Whisper 未识别出任何语音内容")
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(self._segments_to_srt(segments), encoding="utf-8")
logger.info(f"本地 faster-whisper 字幕生成成功: {output_path}")
return output_path
except SpeechRecognitionError:
raise
except ModuleNotFoundError as e:
raise SpeechRecognitionError(
f"Whisper 运行时缺少依赖({e})。请到「设置 → 语音识别」重新安装 Whisper。"
)
except Exception as e: # noqa: BLE001
logger.error(f"本地 faster-whisper 生成字幕失败: {e}", exc_info=True)
raise SpeechRecognitionError(f"本地 Whisper 生成字幕失败: {e}")
def _generate_subtitle_openai_api(self, video_path: Path, output_path: Path,
config: SpeechRecognitionConfig) -> Path:

View File

@@ -1,19 +1,217 @@
import React from 'react'
import { Alert } from 'antd'
import React, { useCallback, useEffect, useRef, useState } from 'react'
import {
Alert, Button, Card, Progress, Tag, Space, Typography, message, List, Popconfirm, Spin, Tooltip,
} from 'antd'
import {
DownloadOutlined, DeleteOutlined, CheckCircleFilled, ReloadOutlined, ThunderboltOutlined,
} from '@ant-design/icons'
import { speechApi, WhisperRuntimeStatus, WhisperModel } from '../services/api'
const { Text, Paragraph } = Typography
interface SpeechRecognitionConfigProps {
config?: Record<string, unknown>
onConfigChange?: (config: Record<string, unknown>) => void
}
const accuracyColor: Record<string, string> = {
: 'green', : 'green', : 'blue', : 'gold', : 'default',
}
const SpeechRecognitionConfig: React.FC<SpeechRecognitionConfigProps> = () => {
const [runtime, setRuntime] = useState<WhisperRuntimeStatus | null>(null)
const [models, setModels] = useState<WhisperModel[]>([])
const [loading, setLoading] = useState(true)
const timer = useRef<number | null>(null)
const refresh = useCallback(async () => {
try {
const [rt, ms] = await Promise.all([speechApi.getRuntimeStatus(), speechApi.getModels()])
setRuntime(rt)
setModels(Array.isArray(ms) ? ms : [])
} catch (e) {
// 后端可能尚未就绪,静默重试
} finally {
setLoading(false)
}
}, [])
// 安装中或有模型下载中时,加快轮询
const needsFastPoll = (rt: WhisperRuntimeStatus | null, ms: WhisperModel[]) =>
rt?.status === 'installing' || ms.some((m) => m.status === 'downloading')
useEffect(() => {
refresh()
return () => { if (timer.current) window.clearInterval(timer.current) }
}, [refresh])
useEffect(() => {
if (timer.current) window.clearInterval(timer.current)
const interval = needsFastPoll(runtime, models) ? 2000 : 15000
timer.current = window.setInterval(refresh, interval)
return () => { if (timer.current) window.clearInterval(timer.current) }
}, [runtime, models, refresh])
const handleInstall = async () => {
try {
const r = await speechApi.installRuntime()
message.info(r.message || '已开始安装')
setRuntime((p) => (p ? { ...p, status: 'installing', progress: 5 } : p))
refresh()
} catch (e: any) {
message.error(e?.response?.data?.detail || '安装失败')
}
}
const handleUninstall = async () => {
try {
const r = await speechApi.uninstallRuntime()
message.success(r.message || '已卸载')
refresh()
} catch (e: any) {
message.error('卸载失败')
}
}
const handleDownload = async (model: string) => {
try {
await speechApi.downloadModel(model)
message.info(`开始下载模型 ${model}`)
setModels((prev) => prev.map((m) => (m.name === model ? { ...m, status: 'downloading' } : m)))
refresh()
} catch (e: any) {
message.error(e?.response?.data?.detail || '下载失败')
}
}
const handleDelete = async (model: string) => {
try {
await speechApi.deleteModel(model)
message.success(`已删除模型 ${model}`)
refresh()
} catch (e) {
message.error('删除失败')
}
}
if (loading) return <Spin />
const installed = runtime?.status === 'installed'
const installing = runtime?.status === 'installing'
const supported = runtime?.platform_supported !== false
return (
<Alert
type="info"
showIcon
message="语音识别配置暂不可用"
description="当前版本会在未上传字幕时尝试使用后端默认 ASR 配置。"
/>
<Space direction="vertical" size="large" style={{ width: '100%' }}>
<Alert
type="info"
showIcon
message="什么时候需要 Whisper"
description="当导入的视频自带字幕(例如 B站 的 AI 字幕)时,会直接使用现成字幕,无需 Whisper。只有当视频没有字幕时才需要本地 Whisper 来自动转写生成字幕。Whisper 为按需安装,装不装、装哪个模型都由你决定。"
/>
{!supported && (
<Alert type="warning" showIcon message="当前平台不支持"
description="mlx-whisper 仅支持 Apple Silicon (M 系列) Mac。" />
)}
{/* 运行时 */}
<Card size="small" title={<Space><ThunderboltOutlined />Whisper </Space>}>
{installed && (
<Space direction="vertical" style={{ width: '100%' }}>
<Space>
<CheckCircleFilled style={{ color: '#52c41a' }} />
<Text strong></Text>
<Text type="secondary">{(runtime?.packages || []).join(', ')}</Text>
</Space>
<Popconfirm title="卸载 Whisper 运行时?已下载的模型不会被删除。" onConfirm={handleUninstall} okText="卸载" cancelText="取消">
<Button danger size="small" icon={<DeleteOutlined />}></Button>
</Popconfirm>
</Space>
)}
{installing && (
<Space direction="vertical" style={{ width: '100%' }}>
<Text> {runtime?.message}</Text>
<Progress percent={runtime?.progress ?? 5} status="active" />
{runtime?.log_tail && (
<pre style={{ maxHeight: 120, overflow: 'auto', background: '#1a1a1a', color: '#bbb', padding: 8, fontSize: 11, borderRadius: 4, margin: 0 }}>
{runtime.log_tail}
</pre>
)}
</Space>
)}
{runtime?.status === 'not_installed' && (
<Space direction="vertical" style={{ width: '100%' }}>
<Paragraph type="secondary" style={{ marginBottom: 8 }}>
faster-whisper 200400MB PyTorch使
</Paragraph>
<Button type="primary" icon={<DownloadOutlined />} onClick={handleInstall} disabled={!supported}>
Whisper
</Button>
</Space>
)}
{runtime?.status === 'error' && (
<Space direction="vertical" style={{ width: '100%' }}>
<Alert type="error" showIcon message="安装出错" description={runtime?.message} />
<Button icon={<ReloadOutlined />} onClick={handleInstall} disabled={!supported}></Button>
</Space>
)}
</Card>
{/* 模型 */}
<Card size="small" title="Whisper 模型">
{!installed && (
<Text type="secondary"> Whisper </Text>
)}
{installed && (
<List
dataSource={models}
renderItem={(m) => {
const downloaded = m.status === 'downloaded'
const downloading = m.status === 'downloading'
return (
<List.Item
actions={[
downloaded ? (
<Popconfirm title={`删除模型 ${m.name}`} onConfirm={() => handleDelete(m.name)} okText="删除" cancelText="取消">
<Button size="small" danger icon={<DeleteOutlined />}></Button>
</Popconfirm>
) : downloading ? (
<Button size="small" loading disabled></Button>
) : (
<Button size="small" type="primary" icon={<DownloadOutlined />} onClick={() => handleDownload(m.name)}>
</Button>
),
]}
>
<List.Item.Meta
title={
<Space>
<Text strong>{m.name}</Text>
<Text type="secondary">{m.size}</Text>
{downloaded && <Tag color="green"></Tag>}
<Tag color={accuracyColor[m.accuracy] || 'default'}> {m.accuracy}</Tag>
<Tooltip title="速度"><Tag>{m.speed}</Tag></Tooltip>
</Space>
}
description={
<Space direction="vertical" style={{ width: '100%' }}>
<Text type="secondary">{m.description}</Text>
{downloading && <Progress percent={m.downloadProgress ?? undefined} status="active" />}
{m.status === 'error' && m.errorMessage && <Text type="danger">{m.errorMessage}</Text>}
</Space>
}
/>
</List.Item>
)
}}
/>
)}
</Card>
</Space>
)
}

View File

@@ -591,4 +591,36 @@ export const systemApi = {
}
}
export interface WhisperRuntimeStatus {
status: 'unknown' | 'not_installed' | 'installing' | 'installed' | 'error'
progress: number
message: string
log_tail?: string
platform_supported: boolean
packages: string[]
}
export interface WhisperModel {
name: string
size: string
sizeBytes: number
description: string
accuracy: string
speed: string
status: 'available' | 'downloading' | 'downloaded' | 'error' | 'not_found'
downloadProgress?: number | null
localPath?: string | null
errorMessage?: string | null
}
// 语音识别 / Whisper 运行时与模型管理
export const speechApi = {
getRuntimeStatus: (): Promise<WhisperRuntimeStatus> => api.get('/whisper/runtime-status'),
installRuntime: (): Promise<{ started: boolean; message: string }> => api.post('/whisper/install'),
uninstallRuntime: (): Promise<{ success: boolean; message: string }> => api.post('/whisper/uninstall'),
getModels: (): Promise<WhisperModel[]> => api.get('/whisper-models'),
downloadModel: (model: string): Promise<unknown> => api.post('/whisper-models/download', { model }),
deleteModel: (model: string): Promise<unknown> => api.delete(`/whisper-models/${model}`),
}
export default api

View File

@@ -130,6 +130,10 @@ backend_dir = sys.argv[1]
sys.path.insert(0, os.path.dirname(backend_dir)) # parent → resolves `backend`
sys.path.insert(0, backend_dir) # backend → resolves `core`, `app`, ...
stdlib = set(sys.stdlib_module_names)
# Modules that are installed AT RUNTIME by the user (Whisper feature), not
# bundled. They are imported lazily inside functions and must NOT fail the
# build. Keep this list tight.
runtime_optional = {"faster_whisper", "ctranslate2", "huggingface_hub"}
mods = set()
for root, _, files in os.walk(backend_dir):
if '__pycache__' in root:
@@ -150,6 +154,7 @@ for root, _, files in os.walk(backend_dir):
missing = sorted(
m for m in mods
if m and not m.startswith('_') and m not in stdlib
and m not in runtime_optional
and importlib.util.find_spec(m) is None
)
if missing: