cat-shark
34f0052d61
`tests/shared/data/alignment/` 里的媒体素材按规则不入库（.gitignore 掉 mp4/wav）， 本机只拷了参考字幕时该用例会判红。按"外部环境状态缺失应跳过"的规则改为：目录在 但一个媒体文件都没有时 skip；有媒体但配对不全仍然失败（那才是真回归）。
140 lines
6.0 KiB
Python
140 lines
6.0 KiB
Python
"""真实语音时间对齐集成测试（数据 → 测试过程 → 验证结果）。
|
|
|
|
覆盖模块：`nodes/whisper.py`（真实转写）+ `tests/shared/realdata_contract.py`
|
|
（对齐量化工具）。这是模块级直测的补充：用真实模型 + 真实音频 + 人工校对
|
|
参考字幕，量化字幕时间轴与"说话真实发生时间"的偏差。
|
|
|
|
数据契约：`tests/shared/data/alignment/` 下 `<name>.wav|.mp4` 与其同名
|
|
`<name>.reference.srt`（人工校对）。素材缺失或本地无模型时跳过。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nodes.whisper import _local_model_candidates, invoke
|
|
from tests.shared.realdata_contract import (
|
|
TIER1_TOLERANCE_SECONDS,
|
|
align_report,
|
|
alignment_candidates,
|
|
clean_reference,
|
|
)
|
|
from tests.shared.srt_entries import parse_srt_entries
|
|
from wov_sdk.models import InvokeRequest
|
|
|
|
|
|
def _model_available() -> Path | None:
|
|
"""返回可用的 V2 权重目录（缺失则返回 None 供跳过）。
|
|
|
|
只接受 V2 权重：V3 已全面停用（均改用 V2），留在盘上的 V3 目录不得被
|
|
测试使用，否则测的不是线上实际运行的模型。
|
|
"""
|
|
from tests.nodes.test_whisper.test_transcribe import _v2_model_candidates
|
|
|
|
for candidate in _v2_model_candidates():
|
|
if candidate.is_dir() and (candidate / "model.bin").is_file():
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _sample_dir() -> Path | None:
|
|
"""返回包含对齐素材的目录（共享数据目录；缺失时返回 None）。"""
|
|
directory = Path(__file__).resolve().parents[2] / "shared" / "data" / "alignment"
|
|
return directory if directory.is_dir() else None
|
|
|
|
|
|
MEDIA_SUFFIXES = {
|
|
".wav", ".mp3", ".flac", ".m4a", ".aac", ".ogg",
|
|
".mp4", ".mkv", ".mov", ".webm", ".ts",
|
|
}
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_real_alignment_candidates_discovered() -> None:
|
|
"""对齐素材探查：能找到成对的媒体与参考字幕（数据契约可用）。"""
|
|
# 数据：共享数据目录下的真实素材。
|
|
directory = _sample_dir()
|
|
if directory is None:
|
|
pytest.skip("缺少 tests/shared/data/alignment 目录")
|
|
# 真实媒体素材不入库（.gitignore 掉的 mp4/wav），没拷过素材属环境状态：跳过
|
|
# 而不是失败；只要有一个媒体文件就说明素材已就位，此时配对不全才是真回归。
|
|
has_media = any(p.is_file() and p.suffix.lower() in MEDIA_SUFFIXES
|
|
for p in directory.iterdir())
|
|
if not has_media:
|
|
pytest.skip("缺少真实媒体素材（按需从媒体库复制，不入库）")
|
|
|
|
# 测试过程
|
|
candidates = alignment_candidates(directory)
|
|
|
|
# 验证结果：至少一对，且每对都有同名 reference.srt。
|
|
assert candidates, "应至少有一对素材 + 参考字幕"
|
|
for media in candidates:
|
|
assert media.with_suffix(".reference.srt").is_file()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_real_whisper_alignment_within_tolerance(tmp_path) -> None:
|
|
"""真实转写产物 vs 人工参考字幕：时间偏差在容差内且无系统性偏移。
|
|
|
|
这验证"字幕时间与说话时刻对齐"这一核心功能，使用真实模型与真实音频。
|
|
"""
|
|
# 数据：最小的真实对齐素材（避免长视频拖慢测试）。
|
|
directory = _sample_dir()
|
|
model_dir = _model_available()
|
|
if directory is None or model_dir is None:
|
|
pytest.skip("缺少对齐素材或本地 whisper 权重，跳过")
|
|
# 显存不足时跳过（转写整段真实音频需要显存，属外部环境状态）。
|
|
from tests.shared.gpu_memory import require_gpu_memory, require_node_result
|
|
|
|
require_gpu_memory(model_dir)
|
|
candidates = alignment_candidates(directory)
|
|
if not candidates:
|
|
pytest.skip("没有成对的对齐素材")
|
|
media = min(candidates, key=lambda p: p.stat().st_size)
|
|
reference = clean_reference(
|
|
parse_srt_entries(media.with_suffix(".reference.srt").read_text(encoding="utf-8"))
|
|
)
|
|
if not reference:
|
|
pytest.skip("参考字幕为空")
|
|
|
|
# 测试过程：走真实生产链路——先提音（视频/任意容器 → 16kHz 单声道 WAV），
|
|
# 再转写。whisper 只接受 WAV，直接喂 mp4 会解析失败（对齐素材多为 mp4）。
|
|
from nodes.ffmpeg import invoke as extract_audio
|
|
|
|
audio = extract_audio(InvokeRequest(
|
|
run_id="alignment", node_instance_id="ffmpeg-1", params={},
|
|
inputs={"video_uri": str(media)}, output_dir=str(tmp_path / "audio"),
|
|
))
|
|
require_node_result(audio, model_dir)
|
|
assert audio.status == "completed", audio.error
|
|
response = invoke(InvokeRequest(
|
|
run_id="alignment", node_instance_id="whisper-1",
|
|
params={
|
|
"language": "ja",
|
|
# 本用例目标是"时间轴对齐"：整段单次解码（chunk_seconds=0）避免
|
|
# 分块上下文带来的额外显存与上下文损失；分块路径本身已由
|
|
# whisper 模块测试覆盖。实测 6GB 卡上该配置稳定完成。
|
|
"chunk_seconds": 0,
|
|
"vad_filter": True,
|
|
"model_path": str(model_dir),
|
|
},
|
|
inputs={"audio_uri": audio.outputs["audio_uri"]},
|
|
output_dir=str(tmp_path / "out"),
|
|
))
|
|
require_node_result(response, model_dir)
|
|
assert response.status == "completed", response.error
|
|
produced = parse_srt_entries(open(response.outputs["srt_uri"], encoding="utf-8").read())
|
|
report = align_report(media.stem, True, produced, reference)
|
|
|
|
# 验证结果（基线由 V2 模型实测标定，见 docs/testing.md）：
|
|
# - 产出条数与参考规模同量级（不丢整段、不大量幻觉）；
|
|
# - 平均绝对偏差在真实转写抖动范围内；
|
|
# - 中位偏差不构成系统性偏早/偏晚。
|
|
assert produced, "真实转写应产出字幕"
|
|
assert len(produced) >= len(reference) * 0.5, report.format_summary()
|
|
assert report.mean_abs_error <= 2.0, report.format_summary()
|
|
assert not report.consistently_early, report.format_summary()
|
|
assert not report.consistently_late, report.format_summary()
|