From c1a835428223f1d19a5cd0b9d8ba49ab9e175b6c Mon Sep 17 00:00:00 2001 From: cat-shark Date: Thu, 13 Aug 2026 22:09:55 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E4=B8=BA=E5=85=A8=E9=83=A8=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E8=A1=A5=E5=85=85=E4=B8=AD=E6=96=87=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E5=B9=B6=E5=8A=A0=E5=85=A5=20AGENTS=20=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E8=A7=84=E8=8C=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 7 +++++++ pyproject.toml | 6 ++++++ tests/test_whisper_node.py | 21 +++++++++++++++++++++ wov_node_whisper/__init__.py | 5 ++++- wov_node_whisper/__main__.py | 17 +++++++++++++++++ 5 files changed, 55 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 239f306..099f227 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,3 +18,10 @@ uv add faster-whisper ``` Linux 上同样在 `wov-node-whisper` 目录执行 `uv add faster-whisper`；节点管理器会自动使用该仓库 `.venv/bin/python` 启动节点。 + +## 代码注释规范 + +- 本仓库所有源码（Python、TOML 等支持注释的文件）必须配有详细中文注释，说明模块职责、模型加载参数与 SRT 生成逻辑，确保后续维护人员可以快速理解代码工作原理。 +- 新增或修改代码时，必须同步补充或更新对应注释；不得删除已有注释。 +- 测试代码同样必须配有中文注释，说明每条测试验证的行为。 +- JSON 数据文件（`node.manifest.json`）不支持注释，字段语义以 `wov-sdk` 的 `NodeManifest` 模型注释和本文档输入/输出说明为准。 diff --git a/pyproject.toml b/pyproject.toml index ccab6d8..9224cb8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,8 +1,10 @@ +# WOV faster-whisper 节点配置：使用 uv 管理环境与依赖。 [project] name = "wov-node-whisper" version = "0.1.0" description = "WOV faster-whisper ASR node" requires-python = ">=3.11" +# 显式加入 NVIDIA 动态库包，保证 GPU 场景下 cublas/cudnn 可被加载。 dependencies = [ "faster-whisper>=1.2.1", "nvidia-cublas-cu12>=12.9.2.10", @@ -10,16 +12,20 @@ dependencies = [ "wov-sdk", ] +# 本地路径依赖 wov-sdk。 [tool.uv.sources] wov-sdk = { path = "../wov-sdk" } +# 开发依赖：pytest 与覆盖率工具。 [dependency-groups] dev = ["pytest", "pytest-cov"] +# pytest 配置：强制 100% 行覆盖率。 [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] addopts = "--cov=wov_node_whisper --cov-report=term-missing --cov-fail-under=100" +# 仅打包节点包本身。 [tool.setuptools] packages = ["wov_node_whisper"] diff --git a/tests/test_whisper_node.py b/tests/test_whisper_node.py index 34f4b9f..b88aad8 100644 --- a/tests/test_whisper_node.py +++ b/tests/test_whisper_node.py @@ -1,3 +1,9 @@ +"""faster-whisper 转写节点测试。 + +通过注入假 faster_whisper 模块覆盖时间戳格式化、参数传递、SRT 生成、 +异常处理和入口点启动等真实代码路径。 +""" + import runpy import sys import types @@ -8,6 +14,8 @@ from wov_sdk.models import InvokeRequest class FakeSegment: + """模拟 faster-whisper 的分段对象，只提供转写测试需要的字段。""" + def __init__(self, start, end, text): self.start = start self.end = end @@ -15,14 +23,18 @@ class FakeSegment: class FakeWhisperModel: + """记录构造参数并返回固定分段的假 WhisperModel。""" + instances: list[tuple[tuple, dict]] = [] def __init__(self, *args, **kwargs): + # 记录每次构造参数，测试据此断言 device/compute_type 传递。 FakeWhisperModel.instances.append((args, kwargs)) self.args = args self.kwargs = kwargs def transcribe(self, path, **kwargs): + # 返回固定两个分段：一个普通时长，一个跨小时验证时间戳格式。 self.transcribe_args = (path, kwargs) return ( [ @@ -34,11 +46,13 @@ class FakeWhisperModel: def _install_fake_whisper(monkeypatch, model_class=FakeWhisperModel) -> None: + """把假 faster_whisper 模块注入 sys.modules，替代真实依赖。""" fake_module = types.SimpleNamespace(WhisperModel=model_class) monkeypatch.setitem(sys.modules, "faster_whisper", fake_module) def _request(tmp_path, **overrides) -> InvokeRequest: + """构造默认音频输入与日语参数的调用请求。""" payload = { "run_id": "run_1", "node_instance_id": "ni_1", @@ -51,12 +65,14 @@ def _request(tmp_path, **overrides) -> InvokeRequest: def test_format_timestamp() -> None: + """验证秒数到 SRT 时间戳的格式化结果。""" assert format_timestamp(0) == "00:00:00,000" assert format_timestamp(3600.5) == "01:00:00,500" assert format_timestamp(61.25) == "00:01:01,250" def test_success(tmp_path, monkeypatch) -> None: + """验证成功转写会生成 SRT 并默认使用 auto 设备/计算类型。""" FakeWhisperModel.instances.clear() _install_fake_whisper(monkeypatch) (tmp_path / "audio.wav").write_bytes(b"fake") @@ -71,6 +87,7 @@ def test_success(tmp_path, monkeypatch) -> None: def test_compute_type_override(tmp_path, monkeypatch) -> None: + """验证请求参数可以覆盖默认计算类型。""" FakeWhisperModel.instances.clear() _install_fake_whisper(monkeypatch) (tmp_path / "audio.wav").write_bytes(b"fake") @@ -81,6 +98,7 @@ def test_compute_type_override(tmp_path, monkeypatch) -> None: def test_model_raises(tmp_path, monkeypatch) -> None: + """验证模型加载失败时返回 failed 与错误信息。""" class BrokenModel: def __init__(self, *args, **kwargs): raise RuntimeError("model load failed") @@ -93,17 +111,20 @@ def test_model_raises(tmp_path, monkeypatch) -> None: def test_missing_input(tmp_path) -> None: + """验证缺少 audio_uri 时返回失败。""" response = invoke(_request(tmp_path, inputs={})) assert response.status == "failed" def test_missing_file(tmp_path) -> None: + """验证音频文件不存在时返回失败。""" response = invoke(_request(tmp_path)) assert response.status == "failed" assert "audio file not found" in response.error def test_entrypoint(monkeypatch) -> None: + """验证 python -m wov_node_whisper 会加载 faster-whisper manifest。""" module_path = Path(__file__).resolve().parent.parent / "wov_node_whisper" / "__main__.py" captured = {} diff --git a/wov_node_whisper/__init__.py b/wov_node_whisper/__init__.py index 42e65fd..fce5e75 100644 --- a/wov_node_whisper/__init__.py +++ b/wov_node_whisper/__init__.py @@ -1 +1,4 @@ -"""WOV faster-whisper ASR node.""" +"""WOV faster-whisper 语音转写节点。 + +把 FFmpeg 节点产出的标准化音频转写为 SRT 字幕，供后续 LLM 翻译节点使用。 +""" diff --git a/wov_node_whisper/__main__.py b/wov_node_whisper/__main__.py index 29f140f..bfcf67b 100644 --- a/wov_node_whisper/__main__.py +++ b/wov_node_whisper/__main__.py @@ -1,3 +1,9 @@ +"""faster-whisper ASR 节点入口。 + +使用 faster-whisper 加载 Whisper 模型，将音频转写为带时间轴的 SRT 文件。 +模型、设备与计算类型均可通过参数或环境变量配置。 +""" + from __future__ import annotations import json @@ -9,6 +15,8 @@ from wov_sdk.server import run_node def format_timestamp(seconds: float) -> str: + """把秒数格式化为 SRT 时间戳，例如 01:00:00,500。""" + # 先换算成毫秒再逐级拆分为时/分/秒/毫秒，避免浮点误差。 total_ms = int(seconds * 1000) hours, remainder = divmod(total_ms, 3600000) minutes, remainder = divmod(remainder, 60000) @@ -17,28 +25,34 @@ def format_timestamp(seconds: float) -> str: def invoke(request: InvokeRequest) -> InvokeResponse: + """转写音频并生成 SRT 字幕，产物为 transcript.srt。""" audio_uri = request.inputs.get("audio_uri") if not audio_uri: return InvokeResponse(status="failed", error="audio_uri is required") + # 文件不存在时提前失败，避免进入耗时的模型加载流程。 audio_path = Path(audio_uri) if not audio_path.is_file(): return InvokeResponse(status="failed", error="audio file not found") try: + # 延迟导入 faster-whisper，保证健康检查等轻量路径不依赖重型依赖。 from faster_whisper import WhisperModel + # 参数优先于环境变量；模型路径缺省使用 faster-whisper 的 large-v3。 model_path = str( request.params.get("model_path") or os.getenv("WHISPER_MODEL_PATH", "large-v3") ) device = str(request.params.get("device") or os.getenv("WHISPER_DEVICE", "auto")) + # auto 让 faster-whisper 根据硬件自动选择 float16/int8 等计算类型。 compute_type = str(request.params.get("compute_type") or "auto") model = WhisperModel( model_path, device=device, compute_type=compute_type, ) + # language 默认日语，vad_filter 过滤静音段以提升转写质量。 segments, _info = model.transcribe( str(audio_path), language=str(request.params.get("language", "ja")), @@ -49,6 +63,7 @@ def invoke(request: InvokeRequest) -> InvokeResponse: output_dir = Path(request.output_dir) output_dir.mkdir(parents=True, exist_ok=True) output_path = output_dir / "transcript.srt" + # 按 SRT 标准输出：序号、时间轴、文本和空行交替。 lines: list[str] = [] for index, segment in enumerate(segments, start=1): lines.extend( @@ -62,10 +77,12 @@ def invoke(request: InvokeRequest) -> InvokeResponse: output_path.write_text("\n".join(lines), encoding="utf-8") return InvokeResponse(status="completed", outputs={"srt_uri": str(output_path)}) except Exception as exc: # noqa: BLE001 + # 模型加载或转写异常统一转换为 failed 响应，不让节点进程退出。 return InvokeResponse(status="failed", error=str(exc)) def main() -> None: + """加载节点清单并以本模块的 invoke 处理器启动服务。""" manifest_path = Path(__file__).resolve().parent.parent / "node.manifest.json" with open(manifest_path, "r", encoding="utf-8") as f: manifest = NodeManifest.from_dict(json.load(f))