cat-shark
2c8bbb6469
无 VAD 的 decode_full 会在无语音段"编"出字幕（`こんにちは`/`おはようございます`/ `東京都交通局8800形電車`）。实测这类幻觉与"呻吟间隙里的真实短台词"在 `no_speech_prob`、`avg_logprob`、silero VAD 与音频能量四个维度上都不可分， 只能用 faster-whisper 自带的 `hallucination_silence_threshold`（HST）抑制： 怀疑幻觉时跳过超过阈值的静音部分（需 `word_timestamps=True`）。 - `nodes/whisper.py` 透传 `word_timestamps` / `hallucination_silence_threshold` / `no_speech_threshold` / `log_prob_threshold` / `compression_ratio_threshold`， 未配置的键不传，保持 faster-whisper 默认值与改造前行为。 - `learn-translate` 默认 HST=2.0 + word_timestamps：同一片头 120 秒无对话段由 15 条 字幕降到 4 条且无套话幻觉残留，呻吟段基本保留；代价转写约慢 1.8×。 - 实测数据与取舍记录见 docs/workflows.md#decode_full-与幻觉呻吟清洗。
618 lines
23 KiB
Python
618 lines
23 KiB
Python
"""nodes/whisper.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`nodes/whisper.py`（转写：模型解析 + 分块 + 时间轴合并 + 幻觉清洗
|
||
入口），可独立调用。模型推理属允许替身的 I/O 边界：单元用例注入结构真实的
|
||
假模型/假 ffmpeg；集成用例使用真实 faster-whisper 模型与真实语音。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import wave
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from nodes.ffmpeg import _ffmpeg_bin
|
||
from nodes.whisper import (
|
||
_append_srt_lines,
|
||
_is_windows,
|
||
_load_cuda_libraries,
|
||
_local_model_candidates,
|
||
_wav_duration_seconds,
|
||
format_timestamp,
|
||
invoke,
|
||
resolve_model_path,
|
||
)
|
||
from wov_sdk.models import InvokeRequest
|
||
|
||
# 模块专用真实素材：60 秒真实语音（16kHz 单声道 WAV）。
|
||
DATA_DIR = Path(__file__).resolve().parent / "data"
|
||
SPEECH_WAV = DATA_DIR / "speech_60s.wav"
|
||
|
||
|
||
class FakeSegment:
|
||
"""结构真实的 whisper 分段替身（start/end/text 与真实段一致）。"""
|
||
|
||
def __init__(self, start: float, end: float, text: str) -> None:
|
||
self.start = start
|
||
self.end = end
|
||
self.text = text
|
||
|
||
|
||
class FakeInfo:
|
||
"""结构真实的转写信息替身（含 language 字段）。"""
|
||
|
||
def __init__(self, language: str = "ja") -> None:
|
||
self.language = language
|
||
|
||
|
||
class FakeModel:
|
||
"""结构真实的假模型：按预置分段返回，并记录每次调用的参数。"""
|
||
|
||
def __init__(self, segments: list[FakeSegment]) -> None:
|
||
self._segments = segments
|
||
self.calls: list[dict] = []
|
||
|
||
def transcribe(self, audio, **kwargs):
|
||
self.calls.append({"audio": str(audio), **kwargs})
|
||
return iter(self._segments), FakeInfo()
|
||
|
||
|
||
def _request(tmp_path: Path, audio: Path | None, **params) -> InvokeRequest:
|
||
"""构造真实请求；audio 为 None 时表示不传 audio_uri。"""
|
||
inputs = {} if audio is None else {"audio_uri": str(audio)}
|
||
return InvokeRequest(
|
||
run_id="run-test",
|
||
node_instance_id="whisper-1",
|
||
params=params,
|
||
inputs=inputs,
|
||
output_dir=str(tmp_path / "out"),
|
||
)
|
||
|
||
|
||
def _inject_model(monkeypatch, model: FakeModel) -> None:
|
||
"""把假模型注入到 faster_whisper.WhisperModel（I/O 边界替身）。
|
||
|
||
节点在 invoke 内部 `from faster_whisper import WhisperModel` 延迟导入，
|
||
因此必须 patch 库模块上的名字，才能让真实调用路径拿到假模型。
|
||
"""
|
||
import faster_whisper
|
||
|
||
monkeypatch.setattr(faster_whisper, "WhisperModel", lambda *a, **k: model)
|
||
# 假模型不需要真实权重，屏蔽 CUDA 库预加载以避免无 GPU 环境的副作用。
|
||
monkeypatch.setattr("nodes.whisper._load_cuda_libraries", lambda: None)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 模型路径解析（本地优先）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_resolve_model_path_explicit_param_wins(tmp_path: Path) -> None:
|
||
"""请求参数 model_path 优先级最高。"""
|
||
# 数据：显式路径（含分隔符，按原样返回）。
|
||
explicit = str(tmp_path / "custom-model")
|
||
|
||
# 测试过程与验证结果
|
||
assert resolve_model_path({"model_path": explicit}, env={}) == explicit
|
||
|
||
|
||
def test_resolve_model_path_bare_name_resolves_locally(tmp_path: Path) -> None:
|
||
"""裸模型名在本地 model/ 目录下解析（存在 model.bin 时）。"""
|
||
# 数据：构造 <模型目录>/<名称>/model.bin。
|
||
models_root = tmp_path / "model"
|
||
target = models_root / "my-model"
|
||
target.mkdir(parents=True)
|
||
(target / "model.bin").write_bytes(b"weights")
|
||
candidates = [models_root / "faster-whisper-large-v2"]
|
||
|
||
# 测试过程
|
||
resolved = resolve_model_path({"model_path": "my-model"}, env={}, candidates=candidates)
|
||
|
||
# 验证结果：解析到本地目录。
|
||
assert resolved == str(target)
|
||
|
||
|
||
def test_resolve_model_path_env_used_when_no_param(tmp_path: Path) -> None:
|
||
"""无参数时使用 WHISPER_MODEL_PATH 环境变量。"""
|
||
# 数据：环境变量指向真实存在的模型目录。
|
||
model_dir = tmp_path / "env-model"
|
||
model_dir.mkdir()
|
||
(model_dir / "model.bin").write_bytes(b"w")
|
||
|
||
# 测试过程与验证结果
|
||
assert resolve_model_path({}, env={"WHISPER_MODEL_PATH": str(model_dir)}) == str(model_dir)
|
||
|
||
|
||
def test_resolve_model_path_prefers_complete_local_candidate(tmp_path: Path) -> None:
|
||
"""无参数/环境变量时使用本地候选目录（含 model.bin 才算完整）。"""
|
||
# 数据：第一个候选缺失 model.bin，第二个完整。
|
||
broken = tmp_path / "broken"
|
||
broken.mkdir()
|
||
good = tmp_path / "good"
|
||
good.mkdir()
|
||
(good / "model.bin").write_bytes(b"w")
|
||
|
||
# 测试过程
|
||
resolved = resolve_model_path({}, env={}, candidates=[broken, good])
|
||
|
||
# 验证结果：跳过不完整候选，选中完整目录。
|
||
assert resolved == str(good)
|
||
|
||
|
||
def test_resolve_model_path_falls_back_to_remote_name(tmp_path: Path) -> None:
|
||
"""全部本地候选缺失时回退到可下载的模型名。"""
|
||
# 数据：空候选目录。
|
||
empty = tmp_path / "empty"
|
||
empty.mkdir()
|
||
|
||
# 测试过程与验证结果
|
||
assert resolve_model_path({}, env={}, candidates=[empty]) == "large-v2"
|
||
|
||
|
||
def test_local_model_candidates_are_platform_paths() -> None:
|
||
"""本地候选包含单体内置模型目录（跨平台用 pathlib 表达）。"""
|
||
# 数据：无。
|
||
# 测试过程
|
||
candidates = _local_model_candidates()
|
||
|
||
# 验证结果：非空且都是 Path。
|
||
assert candidates
|
||
assert all(isinstance(p, Path) for p in candidates)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 时间戳与 WAV 时长
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_format_timestamp_pads_and_handles_hours() -> None:
|
||
"""时间戳格式化为 HH:MM:SS,mmm，毫秒与小时均正确。"""
|
||
# 数据：0、1.5、3661.004 秒。
|
||
# 测试过程与验证结果
|
||
assert format_timestamp(0) == "00:00:00,000"
|
||
assert format_timestamp(1.5) == "00:00:01,500"
|
||
assert format_timestamp(3661.004) == "01:01:01,004"
|
||
|
||
|
||
def test_wav_duration_from_real_header(tmp_path: Path) -> None:
|
||
"""WAV 时长按文件头精确计算（分块偏移依赖它，不能用假设块长）。"""
|
||
# 数据：3 秒合法 WAV。
|
||
path = tmp_path / "3s.wav"
|
||
with wave.open(str(path), "wb") as wav:
|
||
wav.setnchannels(1)
|
||
wav.setsampwidth(2)
|
||
wav.setframerate(16000)
|
||
wav.writeframes(b"\x00\x00" * 16000 * 3)
|
||
|
||
# 测试过程
|
||
duration = _wav_duration_seconds(path, fallback=99.0)
|
||
|
||
# 验证结果
|
||
assert duration == pytest.approx(3.0, abs=0.01)
|
||
|
||
|
||
def test_wav_duration_falls_back_on_invalid_file(tmp_path: Path) -> None:
|
||
"""非法 WAV 时回退到给定默认值（不抛异常中断整片转写）。"""
|
||
# 数据：非 WAV 内容。
|
||
path = tmp_path / "broken.wav"
|
||
path.write_bytes(b"not a wav")
|
||
|
||
# 测试过程与验证结果
|
||
assert _wav_duration_seconds(path, fallback=42.0) == 42.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SRT 行追加与时间轴偏移
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_append_srt_lines_applies_offset_and_index() -> None:
|
||
"""分块转写按块偏移平移时间轴，并延续 SRT 序号。"""
|
||
# 数据：一段本地时间 0~2 秒的分段，偏移 60 秒，起始序号 5。
|
||
segments = [FakeSegment(0.0, 2.0, "你好")]
|
||
|
||
# 测试过程
|
||
lines: list[str] = []
|
||
added = _append_srt_lines(lines, segments, offset=60.0, start_index=5)
|
||
|
||
# 验证结果：序号为 5，时间轴为 60~62 秒，返回本段新增条数 1。
|
||
assert added == 1
|
||
assert lines[0] == "5"
|
||
assert lines[1] == "00:01:00,000 --> 00:01:02,000"
|
||
assert lines[2] == "你好"
|
||
|
||
|
||
def test_append_srt_lines_strips_segment_text() -> None:
|
||
"""分段文本两端空白被去除（避免 SRT 正文带多余空格）。"""
|
||
# 数据：一条文本带首尾空格。
|
||
segments = [FakeSegment(1.0, 2.0, " 正常 ")]
|
||
|
||
# 测试过程
|
||
lines: list[str] = []
|
||
_append_srt_lines(lines, segments, offset=0.0, start_index=1)
|
||
|
||
# 验证结果：正文为去空白后的文本，序号为 1。
|
||
assert lines[0] == "1"
|
||
assert lines[2] == "正常"
|
||
|
||
|
||
def test_append_srt_lines_continues_numbering_across_chunks() -> None:
|
||
"""跨块调用时序号连续（调用方按上一块返回的条数累加）。"""
|
||
# 数据：两块各一段，第二块起始序号 = 1 + 第一块条数。
|
||
first: list[str] = []
|
||
added = _append_srt_lines(first, [FakeSegment(0, 1, "甲")], 0.0, 1)
|
||
second: list[str] = []
|
||
_append_srt_lines(second, [FakeSegment(0, 1, "乙")], 60.0, 1 + added)
|
||
|
||
# 验证结果：第二块序号为 2，时间轴带 60 秒偏移。
|
||
assert second[0] == "2"
|
||
assert second[1].startswith("00:01:00,000")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 平台分支
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_is_windows_flag_matches_platform() -> None:
|
||
"""_is_windows 反映当前平台（测试需同时可在 Windows 与 Linux 运行）。"""
|
||
# 数据：当前运行平台。
|
||
# 测试过程与验证结果
|
||
assert _is_windows() == (os.name == "nt")
|
||
|
||
|
||
def test_load_cuda_libraries_is_noop_off_windows(tmp_path: Path, monkeypatch) -> None:
|
||
"""非 Windows 平台加载 CUDA 库为无操作（Linux 由系统/venv 提供）。"""
|
||
# 数据：强制 _is_windows 为 False。
|
||
monkeypatch.setattr("nodes.whisper._is_windows", lambda: False)
|
||
|
||
# 测试过程与验证结果：不抛异常。
|
||
_load_cuda_libraries()
|
||
|
||
|
||
def test_load_cuda_libraries_scans_site_packages_on_windows(tmp_path: Path, monkeypatch) -> None:
|
||
"""Windows 下扫描 site-packages/nvidia/*/bin 并注册 DLL 搜索目录。"""
|
||
# 数据：伪造含 cublas/cudnn/cuda_nvrtc 三个厂商 bin 目录的 site-packages。
|
||
site = tmp_path / "site-packages"
|
||
vendors = ("cublas", "cudnn", "cuda_nvrtc")
|
||
for package in vendors:
|
||
bin_dir = site / "nvidia" / package / "bin"
|
||
bin_dir.mkdir(parents=True)
|
||
(bin_dir / f"{package}.dll").write_bytes(b"dll")
|
||
added: list[str] = []
|
||
monkeypatch.setattr("nodes.whisper._is_windows", lambda: True)
|
||
monkeypatch.setattr("nodes.whisper.sysconfig.get_paths", lambda: {"purelib": str(site)})
|
||
monkeypatch.setattr("os.add_dll_directory", added.append, raising=False)
|
||
|
||
# 测试过程
|
||
_load_cuda_libraries()
|
||
|
||
# 验证结果：三个厂商的 bin 目录都被加入 DLL 搜索路径。
|
||
assert len(added) == len(vendors)
|
||
assert all("nvidia" in path for path in added)
|
||
|
||
|
||
def test_load_cuda_libraries_skips_missing_vendor_dirs(tmp_path: Path, monkeypatch) -> None:
|
||
"""厂商目录不存在时跳过，不报错（部分轮子未安装）。"""
|
||
# 数据：只有 cublas 一个厂商目录。
|
||
site = tmp_path / "site-packages"
|
||
(site / "nvidia" / "cublas" / "bin").mkdir(parents=True)
|
||
added: list[str] = []
|
||
monkeypatch.setattr("nodes.whisper._is_windows", lambda: True)
|
||
monkeypatch.setattr("nodes.whisper.sysconfig.get_paths", lambda: {"purelib": str(site)})
|
||
monkeypatch.setattr("os.add_dll_directory", added.append, raising=False)
|
||
|
||
# 测试过程
|
||
_load_cuda_libraries()
|
||
|
||
# 验证结果：只注册存在的那一个。
|
||
assert len(added) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# invoke：分块转写与合并（假模型）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_invoke_transcribes_with_fake_model_and_writes_srt(tmp_path: Path, monkeypatch) -> None:
|
||
"""invoke 调用模型转写并写出 SRT（结构真实的分段替身）。"""
|
||
# 数据：真实 WAV 输入 + 假模型返回两段。
|
||
assert SPEECH_WAV.is_file(), f"缺少测试素材 {SPEECH_WAV}"
|
||
model = FakeModel([FakeSegment(0.0, 2.0, "第一句"), FakeSegment(2.5, 4.0, "第二句")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0))
|
||
|
||
# 验证结果：产物存在且含两段文本与时间轴。
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
assert "第一句" in content and "第二句" in content
|
||
assert "00:00:00,000 --> 00:00:02,000" in content
|
||
assert content.index("第一句") < content.index("第二句")
|
||
|
||
|
||
def test_invoke_passes_params_to_model(tmp_path: Path, monkeypatch) -> None:
|
||
"""节点参数透传到模型调用（language/task/beam_size 等）。"""
|
||
# 数据：指定语言与任务。
|
||
model = FakeModel([FakeSegment(0.0, 1.0, "x")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
invoke(_request(
|
||
tmp_path, SPEECH_WAV, chunk_seconds=0,
|
||
language="ja", task="translate", beam_size=3, vad_filter=False,
|
||
condition_on_previous_text=False,
|
||
))
|
||
|
||
# 验证结果：模型收到对应参数。
|
||
call = model.calls[0]
|
||
assert call["language"] == "ja"
|
||
assert call["task"] == "translate"
|
||
assert call["beam_size"] == 3
|
||
assert call["vad_filter"] is False
|
||
|
||
|
||
def test_invoke_default_vad_and_condition_flags(tmp_path: Path, monkeypatch) -> None:
|
||
"""默认 vad_filter=True 且 condition_on_previous_text=False（防重复）。"""
|
||
# 数据：不传相关参数。
|
||
model = FakeModel([FakeSegment(0.0, 1.0, "x")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0))
|
||
|
||
# 验证结果
|
||
assert model.calls[0]["vad_filter"] is True
|
||
assert model.calls[0]["condition_on_previous_text"] is False
|
||
|
||
|
||
def test_invoke_fails_without_audio_uri(tmp_path: Path) -> None:
|
||
"""缺少 audio_uri 时失败。"""
|
||
# 数据：空输入。
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, None))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "audio_uri" in (response.error or "")
|
||
|
||
|
||
def test_invoke_fails_when_audio_missing(tmp_path: Path) -> None:
|
||
"""音频文件不存在时失败。"""
|
||
# 数据：不存在的路径。
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, tmp_path / "nope.wav"))
|
||
|
||
# 验证结果
|
||
assert response.status == "failed"
|
||
assert "not found" in (response.error or "")
|
||
|
||
|
||
def test_invoke_cleans_japanese_hallucination_in_decode_full(tmp_path: Path, monkeypatch) -> None:
|
||
"""decode_full 模式下，长时日语寒暄幻觉整条删除（不留下 '-' 占位）。"""
|
||
# 数据：假模型返回一段 30 秒的"おやすみなさい"。
|
||
model = FakeModel([FakeSegment(0.0, 30.0, "おやすみなさい")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0, decode_full=True))
|
||
|
||
# 验证结果：幻觉被删除，产物无该文本。
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
assert "おやすみなさい" not in content
|
||
|
||
|
||
def test_invoke_filters_short_moan_in_decode_full(tmp_path: Path, monkeypatch) -> None:
|
||
"""decode_full 模式下短呻吟碎片被过滤，真实短对话保留。"""
|
||
# 数据：呻吟碎片 + 真实短对话。
|
||
model = FakeModel([
|
||
FakeSegment(0.0, 1.0, "あ…"),
|
||
FakeSegment(1.5, 3.0, "そこ、だめ"),
|
||
])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
response = invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0, decode_full=True))
|
||
|
||
# 验证结果：呻吟被删，真实对话保留。
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
assert "そこ、だめ" in content
|
||
assert "あ…" not in content
|
||
|
||
|
||
def test_invoke_keeps_moan_when_filter_disabled(tmp_path: Path, monkeypatch) -> None:
|
||
"""short_moan_max_chars=0 时关闭呻吟过滤。"""
|
||
# 数据：一条呻吟。
|
||
model = FakeModel([FakeSegment(0.0, 1.0, "あ…")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
response = invoke(_request(
|
||
tmp_path, SPEECH_WAV, chunk_seconds=0, decode_full=True, short_moan_max_chars=0,
|
||
))
|
||
|
||
# 验证结果：呻吟保留。
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
assert "あ…" in content
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 真实模型集成
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# 已废弃的模型目录：全系统只用 V2，V3 权重不得进入测试。
|
||
_DEPRECATED_MODEL_DIRS = ("faster-whisper-large-v3",)
|
||
|
||
|
||
def _v2_model_candidates() -> list[Path]:
|
||
"""返回可用的 V2 权重目录（排除已废弃的 V3）。
|
||
|
||
解析顺序：
|
||
1. `nodes.whisper` 文档化的默认候选（通用 V2 转写模型）；
|
||
2. `model/` 下其它已下载的 V2 权重（例如中文直出模型）。
|
||
V3 权重已全面停用（全系统只用 V2），即使留在盘上也不得被测试使用，
|
||
否则测的不是线上实际运行的模型。
|
||
"""
|
||
candidates = [p for p in _local_model_candidates() if p.name not in _DEPRECATED_MODEL_DIRS]
|
||
model_root = Path(__file__).resolve().parents[3] / "model"
|
||
if model_root.is_dir():
|
||
for path in sorted(model_root.iterdir()):
|
||
if not path.is_dir() or path.name in _DEPRECATED_MODEL_DIRS:
|
||
continue
|
||
# V3 的判据：preprocessor_config.json 的 feature_size == 128。
|
||
if _is_v3_weights(path):
|
||
continue
|
||
candidates.append(path)
|
||
return candidates
|
||
|
||
|
||
def _is_v3_weights(model_dir: Path) -> bool:
|
||
"""按 preprocessor_config.json 的 feature_size 判断是否为 V3 权重。
|
||
|
||
Whisper V2 的 mel 特征维度是 80，V3 是 128；这是区分两代权重的稳定判据
|
||
（目录名可能被人工改名，不能只靠名字判断）。
|
||
"""
|
||
import json
|
||
|
||
config = model_dir / "preprocessor_config.json"
|
||
if not config.is_file():
|
||
return False
|
||
try:
|
||
return int(json.loads(config.read_text(encoding="utf-8")).get("feature_size", 80)) == 128
|
||
except (ValueError, TypeError, OSError):
|
||
return False
|
||
|
||
|
||
def _real_model_available() -> Path | None:
|
||
"""返回一个可用的 V2 权重目录（缺失则返回 None 供跳过）。"""
|
||
for candidate in _v2_model_candidates():
|
||
if candidate.is_dir() and (candidate / "model.bin").is_file():
|
||
return candidate
|
||
return None
|
||
|
||
|
||
@pytest.mark.integration
|
||
def test_real_whisper_transcribes_real_speech(tmp_path: Path) -> None:
|
||
"""真实 faster-whisper 模型 + 真实语音：端到端转写产出可用 SRT。
|
||
|
||
本地无模型或素材时跳过；有则必须执行，作为假模型单测的校准。
|
||
"""
|
||
# 数据：模块 data/ 下的真实 60 秒语音。
|
||
if not SPEECH_WAV.is_file():
|
||
pytest.skip(f"缺少测试素材 {SPEECH_WAV}")
|
||
model_dir = _real_model_available()
|
||
if model_dir is None:
|
||
pytest.skip("本地没有完整 whisper 权重，跳过真实模型集成测试")
|
||
# 显存不足时跳过（真实模型推理需要显存，属外部环境状态）。
|
||
from tests.shared.gpu_memory import (
|
||
fits_with_margin,
|
||
require_gpu_memory,
|
||
require_node_result,
|
||
)
|
||
|
||
require_gpu_memory(model_dir)
|
||
# 分块路径（生产默认）会产生更多分配峰值，在临界显存卡上易触发 CUDA OOM；
|
||
# 显存充裕时走分块覆盖该路径，否则退化为整段单次推理。
|
||
chunk_seconds = 20 if fits_with_margin(model_dir) else 0
|
||
|
||
# 测试过程：真实模型转写真实语音。
|
||
response = invoke(_request(
|
||
tmp_path, SPEECH_WAV, chunk_seconds=chunk_seconds, language="ja",
|
||
model_path=str(model_dir),
|
||
))
|
||
|
||
# 验证结果：成功、产物为合法 SRT、时间轴递增且不超音频时长。
|
||
require_node_result(response, model_dir)
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
timelines = [line for line in content.splitlines() if "-->" in line]
|
||
assert timelines, "真实转写应产出至少一条字幕"
|
||
from tests.shared.srt_entries import parse_srt_entries
|
||
|
||
entries = parse_srt_entries(content)
|
||
starts = [e["start"] for e in entries]
|
||
assert starts == sorted(starts)
|
||
assert max(starts) <= 62.0
|
||
|
||
|
||
def test_invoke_drops_repetition_artifact_in_decode_full(tmp_path: Path, monkeypatch) -> None:
|
||
"""decode_full 下删除 30 秒重复伪影：它会带着翻译层一起进重复循环。
|
||
|
||
数据：假模型返回一段 30 秒窗口被同一单元填满的伪影 + 一条真实台词。
|
||
过程：调用 invoke（decode_full=True）。
|
||
验证：伪影整条删除、真实台词保留，产物里不再出现超长重复正文。
|
||
"""
|
||
artifact = "チン" * 111
|
||
model = FakeModel([
|
||
FakeSegment(0.0, 30.0, artifact),
|
||
FakeSegment(30.0, 33.0, "そこ、だめ"),
|
||
])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
response = invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0, decode_full=True))
|
||
|
||
assert response.status == "completed", response.error
|
||
content = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||
assert artifact not in content
|
||
assert "そこ、だめ" in content
|
||
|
||
|
||
def test_invoke_passes_hallucination_guard_params(tmp_path: Path, monkeypatch) -> None:
|
||
"""数据：工作流传入幻觉抑制参数（HST + word_timestamps + 两个阈值）。
|
||
|
||
过程：调用 invoke。
|
||
|
||
验证：参数原样传给 faster-whisper——静音段幻觉只能靠这些参数抑制：
|
||
`hallucination_silence_threshold` 需要 word_timestamps 才生效。
|
||
"""
|
||
# 数据：无语音段也能"编"出字幕的假模型 + 显式传入的抑制参数。
|
||
model = FakeModel([FakeSegment(0.0, 2.0, "こんにちは")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
invoke(_request(
|
||
tmp_path, SPEECH_WAV, chunk_seconds=0, language="ja",
|
||
word_timestamps=True, hallucination_silence_threshold=2.0,
|
||
no_speech_threshold=0.3, log_prob_threshold=-1.2,
|
||
compression_ratio_threshold=2.4,
|
||
))
|
||
|
||
# 验证结果
|
||
call = model.calls[0]
|
||
assert call["word_timestamps"] is True
|
||
assert call["hallucination_silence_threshold"] == 2.0
|
||
assert call["no_speech_threshold"] == 0.3
|
||
assert call["log_prob_threshold"] == -1.2
|
||
assert call["compression_ratio_threshold"] == 2.4
|
||
|
||
|
||
def test_invoke_omits_guard_params_by_default(tmp_path: Path, monkeypatch) -> None:
|
||
"""数据：不传抑制参数（默认工作流）。
|
||
|
||
过程：调用 invoke。
|
||
|
||
验证：不透传这些键，保持 faster-whisper 自身默认值，行为与改造前一致。
|
||
"""
|
||
# 数据：普通假模型。
|
||
model = FakeModel([FakeSegment(0.0, 1.0, "x")])
|
||
_inject_model(monkeypatch, model)
|
||
|
||
# 测试过程
|
||
invoke(_request(tmp_path, SPEECH_WAV, chunk_seconds=0))
|
||
|
||
# 验证结果
|
||
call = model.calls[0]
|
||
assert "hallucination_silence_threshold" not in call
|
||
assert "no_speech_threshold" not in call
|
||
assert "word_timestamps" not in call
|