feat: whisper 节点自动接入自适应 VAD + 方案文档

- nodes/whisper.py：invoke 检测到 vad_filter=true 且未显式传 vad_parameters
  且 WOV_AUTO_VAD=1 时，自动调用 vad_profiler.vad_parameters_for_audio 按
  本音频信号动态确定 VAD 参数并传给 transcribe；分析失败回退默认不中断转写
  （防御性，不影响整段流程）。
- tests/test_nodes.py：新增自动 VAD 异常回退测试（patch profiler 抛错，
  验证 whisper 正常转写 + transcribe 收到 vad_parameters=None）。
- docs/adaptive_vad.md：完整方案文档（背景/可行性/三层架构/参数建议规则/
  评分器/实现落点/取舍/实现状态）。

docs/adaptive_vad.md

@@ -0,0 +1,107 @@
+# 每视频自适应 VAD 调参方案
+
+## 背景：为什么必须每视频单独调 VAD
+
+实测 CJOD-255 音频（2 小时成人视频）暴露了固定 VAD 配置的根本缺陷：
+
+- 全片 1s 能量 **56% 在 RMS<900（音乐/低语）**、仅 13% 是静音级(<300)；
+  全程有 BGM，**几乎没有纯静音**。
+- 简单能量阈值（RMS>700）会把**整片识别成 1 个 4120s 的巨型语音段**——
+  BGM 让能量始终高于阈值，VAD 无法分离出真实说话。
+- 结果是 whisper 全段解码：80% 字幕碎片化（啊/嗯/はい）、32% 参考句漏识别、
+  敏感段（口交等）丢失。
+
+不同视频的 BGM/静音/人声比例天差地别（测试片、音乐 MV、含 BGM 剧集、
+纯语音播客），**固定 VAD 参数必然在多数视频上偏差**。因此需要对**每个视频**
+先独立做信号分析，再动态决定 VAD 参数。
+
+## 可行性
+
+1. **VAD 是解码前的前置步骤**：先用 ffmpeg 提 WAV（已有），可独立做信号分析，
+   不占用模型推理资源。
+2. **faster-whisper 支持 `vad_parameters` 细参**：可传
+   `threshold`（说话判定阈值）、`min_silence_duration_ms`（断句静音时长）、
+   `speech_pad_ms`（语音段首尾缓冲）。当前节点只传 `vad_filter`，细参入口未用。
+3. **求解代价可控**：找最优参数在 1-2 分钟代表性片段上做网格搜索，
+   全片只跑一次确定后的参数。
+
+## 方案架构（三层）
+
+```
+每个视频进入 whisper 节点时：
+┌─────────────────────────────────────────────────────────┐
+│ ① 信号分析 profile_audio（秒级，无模型）                │
+│    - 1s 网格能量/RMS 分布 → 静音比例、BGM 底噪、语音疏密    │
+│    - 推断：threshold 候选、min_silence 候选、speech_pad    │
+│ ② 片段网格验证 pick_best_vad（90s 代表片段，3~5 组参数）    │
+│    - 若提供参考字幕 → 用"参考覆盖率+时间对齐误差"评分      │
+│    - 无参考 → 用"碎片率/连贯性/置信度"启发式评分            │
+│ ③ 全片用 ② 选出的 vad_parameters 跑一次                  │
+└─────────────────────────────────────────────────────────┘
+```
+
+## 信号分析 → VAD 参数推荐规则
+
+| 音频特征（从能量分布算出） | VAD 建议 |
+| --- | --- |
+| 静音比例高（>30%）、BGM 低 | threshold 高(0.6~0.7)、min_silence 2000，正常 VAD |
+| **BGM 覆盖广（如 CJOD 56%<900）** | **threshold 低(0.3~0.4)、min_silence 300~500、speech_pad 0**——避免把音乐当语音整段拼接 |
+| 纯语音（播客/采访）| threshold 0.5、min_silence 1000，普通 |
+| 长静音（停顿>2s 常见）| speech_pad 减小（0~200）防时间轴压缩漂移 |
+
+### 关键：BGM 掩盖音频的处理
+
+对整片被 BGM 覆盖的视频，纯能量 VAD 天然失效。三种手段按优先级：
+
+1. **降低 VAD threshold**（0.5→0.3）：让 silero 的语音概率模型在 BGM 中
+   更敏感地捕捉人声，避免把音乐段误判为静音而吞掉轻语。
+2. **降低 min_silence_duration_ms**（2000→300~500）：避免把停顿超过 2s 的
+   短句硬并成一条，缓解"碎片化/漏句"。
+3. **speech_pad_ms 减小**（400→0~200）：缓冲越大，Whisper 对片段起始时间
+   估计越偏早，减小可提升时间对齐精度。
+
+## 评分器（决定最优参数）
+
+### 有参考字幕（OCR 硬字幕 = ground truth）
+参考的第 k 条 vs 转录的对齐：
+- **覆盖率** = 能在 1.5s 内找到语义对应日文的参考条数 / 总参考条数（越高越好）
+- **时间误差** = 配对条目 |转录start - 参考start| 的平均（越低越好）
+- 综合分 = 覆盖率 - 0.3×时间误差（权重可调）
+
+### 无参考字幕（启发式）
+用转录自身质量：
+- **碎声率** = 纯单字/语气词条目占比（越低越好，如 <20%）
+- **连贯性** = 每条平均字数（适中为好，不碎不并）
+- **置信度** = whisper segment 的 avg_logprob
+
+## 实现落点
+
+- 新模块 `nodes/vad_profiler.py`：`profile_audio()`（信号分析）、
+  `pick_best_vad()`（片段网格 + 评分）、`vad_parameters_for_audio()`（总入口）
+- `nodes/whisper.py`：在 `invoke` 里检测到 `vad_filter=true` 且未显式传
+  `vad_parameters` 时，调用 profiler 生成 per-video 参数传给 `transcribe()`
+- 参考字幕可选：若工作流/请求提供 `reference_srt_uri`，走"有参考评分"
+- 门控：`WOV_AUTO_VAD=1`（默认开），可关；显式传 `vad_parameters` 时跳过
+
+## 取舍
+
+- **收益**：每个视频用最贴合其声学的 VAD，显著降低碎片化、漏句、时间错位，
+  尤其对 BGM 音频（如成人视频、综艺）。
+- **成本**：每视频额外 10~30s 信号分析 + 片段上 3~5 次短转写（分钟级），
+  相比全片转写可接受。
+- **风险**：片段网格选取的代表性片段若无说话/纯音乐，可能评为最差参数。
+  缓解：选 2 段（开头+中段）拼接，且阈值下限保护。
+## 实现状态（2026-09 已落地）
+
+- `nodes/vad_profiler.py`：`profile_audio()`（1s 能量分析）、
+  `suggest_vad_parameters()`（信号→VAD 建议）、`score_transcript()`（幻觉词
+  扣分启发式）、`vad_parameters_for_audio()`（总入口）、`_pick_representative_start()`
+  与 `_grid_search_vad()`（片段网格）。
+- `nodes/whisper.py`：`invoke` 检测到 `vad_filter=true` 且未显式传
+  `vad_parameters` 且 `WOV_AUTO_VAD=1` 时，自动调用 profiler 生成 per-video 参数
+  传给 `transcribe()`；分析失败回退默认不中断转写。
+- 测试 `tests/test_vad_profiler.py`（19 个）覆盖：信号分析、四个建议分支、
+  幻觉词/碎片评分、网格选优、异常回退、空音频/低采样率、候选网格展开。
+- 门控：`WOV_AUTO_VAD=0` 关闭自动调参；显式传 `vad_parameters` 时跳过。
+- 说明：当前接入默认 `whisper_invoke=None`（仅信号分析，秒级、不额外跑模型）；
+  若未来要启用片段网格验证，传入 whisper 回调并暴露配置即可，框架已就绪。

nodes/vad_profiler.py

@@ -0,0 +1,311 @@
+"""每视频自适应 VAD 调参（信号分析 + 片段网格验证）。
+
+背景（实测 CJOD-255）：全程 BGM 覆盖的视频几乎没有纯静音，固定 VAD
+参数会把音乐当语音，导致 whisper 全段解码 → 碎片化(80%)、漏句(32%)、
+敏感段丢失。不同视频声学差异大，需**每视频独立分析后再定 VAD 参数**。
+
+方案（用户确认）：
+1. 信号分析（秒级，无模型）缩窄参数空间：1s 能量分布 → 静音比例、
+   BGM 底噪、语音疏密，推出 threshold/min_silence/speech_pad 候选；
+2. 片段小网格验证（90s 代表片段，3~5 组参数跑 whisper）：
+   - **不用参考字幕**（实际部署无参考）：纯转录质量启发式评分；
+   - **常出现的幻觉词作为扣分项**：幻觉越多 → VAD 把非语音当语音 →
+     该参数组合越差。
+3. 全片用选出的 vad_parameters 跑一次。
+
+门控：whisper 节点 detected 未显式传 vad_parameters 且 vad_filter=true 时
+自动调用本模块；WOV_AUTO_VAD=0 可关闭。
+"""
+
+from __future__ import annotations
+
+import math
+import os
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import wave  # 仅读 WAV 头/样本
+
+# 幻觉词扣分集合：VAD 若把音乐/静音当语音，whisper 常在这些段输出套话。
+# 每条命中扣 1 分；重复次数越高 → 该参数组合越差。
+HALLUCINATION_TOKENS = (
+    "ご視聴ありがとうございました",  # 感谢观看
+    "ご視聴ありがとうございました。",
+    "おやすみなさい",            # 晚安
+    "音楽",                      # 音乐
+    "また見てね",                # 下次再见
+    "Goodbye",
+    "ありがとうございました",      # 感谢
+    "エンドロール",
+    "お楽しみに",
+)
+
+# 单字/语气词（碎片化判定）：纯这些字符的条目视为"碎声"。
+_FRAGMENT_RE = re.compile(r"^[ぁ-んァ-ヶあいうえおっーんすよわ\s、。]+$")
+
+# 默认候选网格（片段验证时用；由信号分析缩小范围）。
+_THRESHOLDS = (0.3, 0.5, 0.7)
+_SILENCES_MS = (300, 1000, 2000)
+_PADS_MS = (0, 200, 400)
+
+
+@dataclass
+class AudioProfile:
+    """音频信号分析结果。"""
+
+    sample_rate: int = 16000
+    duration_seconds: float = 0.0
+    rms_bins: list[float] = field(default_factory=list)  # 每 1s RMS
+    silence_ratio: float = 0.0       # RMS<300（静音级）占比
+    lowish_ratio: float = 0.0        # RMS<900（音乐/低语）占比
+    voice_ratio: float = 0.0         # 高能（语音）占比
+    bgm_heavy: bool = False          # 是否 BGM 覆盖广（低能量占比高但非纯静音）
+    long_silence: bool = False       # 是否长停顿常见
+
+    @property
+    def median_rms(self) -> float:
+        if not self.rms_bins:
+            return 0.0
+        ordered = sorted(self.rms_bins)
+        return ordered[len(ordered) // 2]
+
+
+def _read_wav_samples(audio_path: Path, sample_rate: int) -> tuple[list[int], int]:
+    """读取单声道 WAV 的全部样本（16-bit 小端）。返回 (样本列表, 实际采样率)。"""
+    with wave.open(str(audio_path), "rb") as wav:
+        rate = wav.getframerate()
+        n = wav.getnframes()
+        data = wav.readframes(n)
+    import array
+
+    samples = array.array("h")
+    samples.frombytes(data)
+    return samples, rate
+
+
+def profile_audio(audio_path: Path, sample_rate: int = 16000) -> AudioProfile:
+    """做 1s 网格能量分析，返回 AudioProfile。
+
+    每秒计算 RMS；据此得出静音/音乐/语音占比，判断是否 BGM 覆盖、有无长静音。
+    """
+    samples, rate = _read_wav_samples(audio_path, sample_rate)
+    if rate != sample_rate:
+        # 采样率不一致时用 wave 实际 rate 近似（不重采样，仅用于比例判断）。
+        rate = rate or sample_rate
+    bin_frames = rate  # 1 秒一个 bin
+    bins: list[float] = []
+    n_bins = len(samples) // bin_frames
+    for i in range(n_bins):
+        seg = samples[i * bin_frames : (i + 1) * bin_frames]
+        rms = math.sqrt(sum(s * s for s in seg) / len(seg)) if seg else 0
+        bins.append(rms)
+
+    if not bins:
+        return AudioProfile(sample_rate=sample_rate, duration_seconds=0.0)
+
+    silence = sum(1 for r in bins if r < 300) / len(bins)
+    lowish = sum(1 for r in bins if r < 900) / len(bins)
+    voice = sum(1 for r in bins if r >= 900) / len(bins)
+
+    # BGM 覆盖：低能量占比高（音频不静也不高亢），说明有音乐铺底。
+    bgm_heavy = lowish > 0.4 and silence < 0.4
+    # 长静音：存在连续 >=5 个静音 bin（5s 以上停顿）。
+    long_silence = False
+    run = 0
+    for r in bins:
+        if r < 300:
+            run += 1
+            if run >= 5:
+                long_silence = True
+                break
+        else:
+            run = 0
+
+    return AudioProfile(
+        sample_rate=rate,
+        duration_seconds=len(samples) / rate,
+        rms_bins=bins,
+        silence_ratio=silence,
+        lowish_ratio=lowish,
+        voice_ratio=voice,
+        bgm_heavy=bgm_heavy,
+        long_silence=long_silence,
+    )
+
+
+def suggest_vad_parameters(profile: AudioProfile) -> dict:
+    """由信号分析推出候选 VAD 参数（不跑模型，秒级）。
+
+    返回 dict 含 threshold/min_silence_duration_ms/speech_pad_ms 候选。
+    规则见 docs/adaptive_vad.md。
+    """
+    if profile.bgm_heavy:
+        # BGM 覆盖广：降低 threshold 增强人声敏感、减 min_silence 防并句、
+        # 减 speech_pad 防时间偏移。
+        return {
+            "threshold": 0.3,
+            "min_silence_duration_ms": 300,
+            "speech_pad_ms": 0,
+        }
+    if profile.long_silence:
+        # 长停顿常见：speech_pad 减小防时间轴压缩漂移，threshold 正常。
+        return {
+            "threshold": 0.5,
+            "min_silence_duration_ms": 1000,
+            "speech_pad_ms": 200,
+        }
+    if profile.silence_ratio > 0.3:
+        # 静音占比高（正常带静音的音频）：threhold 略高剔除虚警。
+        return {
+            "threshold": 0.6,
+            "min_silence_duration_ms": 2000,
+            "speech_pad_ms": 400,
+        }
+    # 常规。
+    return {
+        "threshold": 0.5,
+        "min_silence_duration_ms": 1000,
+        "speech_pad_ms": 400,
+    }
+
+
+def _hits_hallucination(text: str) -> int:
+    """统计一条文本命中幻觉词的次数（多个套话可叠加）。"""
+    hits = 0
+    for token in HALLUCINATION_TOKENS:
+        if token in text:
+            hits += 1
+    return hits
+
+
+def score_transcript(segments: list) -> float:
+    """启发式评分转录质量（不用参考字幕）。
+
+    分数越高越好：
+    - 碎片率（纯单字/语气词条目占比）越低越好：penalty += 碎片数 * 2
+    - 幻觉词（感谢观看/晚安/音乐）越多越差：penalty += 幻觉命中次数 * 3
+    - 平均每条字数适中（不碎不并）：penalty += max(0, 平均字数-30) * 0.5
+    返回 100 - penalty。
+    """
+    if not segments:
+        return 0.0
+    n = len(segments)
+    frag = 0
+    hall = 0
+    total_chars = 0
+    for seg in segments:
+        text = getattr(seg, "text", "")
+        if not text:
+            continue
+        total_chars += len(text)
+        if _FRAGMENT_RE.match(text):
+            frag += 1
+        hall += _hits_hallucination(text)
+    avg_len = total_chars / n if n else 0
+    penalty = frag * 2.0 + hall * 3.0 + max(0.0, (avg_len - 20)) * 0.3
+    return max(0.0, 100.0 - penalty)
+
+
+def vad_parameters_for_audio(
+    audio_path: Path,
+    sample_rate: int = 16000,
+    whisper_invoke=None,
+    run_dir: Path | None = None,
+    chunk_seconds: int = 60,
+) -> dict:
+    """为单个音频决定最优 VAD 参数（信号分析 + 片段网格验证）。
+
+    若不提供 whisper_invoke（无法跑片段），退化为仅信号分析启发式。
+    """
+    profile = profile_audio(audio_path, sample_rate)
+    suggested = suggest_vad_parameters(profile)
+
+    if whisper_invoke is None or not profile.rms_bins:
+        # 无推理能力：直接用信号分析结论，不再网格验证。
+        return suggested
+
+    # 片段网格：取 1 段 90s 代表性片段（选能量中等段，避免纯音乐/纯静音）。
+    frag_start = _pick_representative_start(profile, 90)
+    fragment_params = _grid_search_vad(
+        whisper_invoke, audio_path, run_dir, chunk_seconds, frag_start, suggested
+    )
+    return fragment_params
+
+
+def _pick_representative_start(profile: AudioProfile, window: int) -> int:
+    """从 1s 能量里选一段 90s 窗口（平均能量接近中位、不纯静音）。
+
+    返回窗口起始秒；音频不足 window 秒时返回 0。
+    """
+    bins = profile.rms_bins
+    n = len(bins)
+    if n <= window:
+        return 0
+    median = profile.median_rms
+    best = 0
+    best_score = float("inf")
+    for start in range(0, n - window, max(1, window // 2)):
+        window_bins = bins[start : start + window]
+        avg = sum(window_bins) / len(window_bins)
+        # 分数 = 与中位能量差的绝对值 + 静音惩罚（窗口全静音不好）。
+        silence = sum(1 for r in window_bins if r < 300) / len(window_bins)
+        score = abs(avg - median) + (silence * 1000 if silence > 0.6 else 0)
+        if score < best_score:
+            best_score = score
+            best = start
+    return best
+
+
+def _grid_search_vad(
+    whisper_invoke,
+    audio_path: Path,
+    run_dir: Path | None,
+    chunk_seconds: int,
+    frag_start: int,
+    suggested: dict,
+) -> dict:
+    """对候选参数网格各跑一次片段 whisper，按启发式评分选最优。
+
+    候选 = 信号分析建议值 + 少量邻域组合（_candidate_params），
+    每组合在 60s 片段上转写，按 score_transcript 选最高分。
+    若无法运行 whisper（如缺模型）返回 suggested。
+    """
+    window_seconds = 60
+    cand = _candidate_params(suggested)
+    best = suggested
+    best_score = -1.0
+    for params in cand:
+        try:
+            segs = whisper_invoke(
+                audio_path=audio_path,
+                start_seconds=frag_start,
+                window_seconds=window_seconds,
+                vad_parameters=params,
+                output_dir=str(run_dir / "vad_probe") if run_dir else None,
+                chunk_seconds=chunk_seconds,
+            )
+        except Exception:
+            # 单组参数调用失败不致命，跳过继续评估其它组合。
+            continue
+        if not isinstance(segs, list):
+            continue
+        score = score_transcript(segs)
+        if score > best_score:
+            best_score = score
+            best = params
+    return best
+
+
+def _candidate_params(suggested: dict) -> list[dict]:
+    """由建议值扩展出候选网格（附近组合，避免过深搜索）。"""
+    base = dict(suggested)
+    thres = base.get("threshold", 0.5)
+    sil = base.get("min_silence_duration_ms", 1000)
+    pad = base.get("speech_pad_ms", 400)
+    cand = []
+    for t in (thres, round(thres - 0.1, 1), round(thres + 0.1, 1)):
+        for s in (sil, sil // 2, sil * 2):
+            for p in (0, pad // 2, pad):
+                cand.append({"threshold": t, "min_silence_duration_ms": s, "speech_pad_ms": p})
+    return cand

nodes/whisper.py

@@ -236,6 +236,27 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
         # 分块转写：默认每 1 分钟一块（chunk_seconds=60），切块失败自动回退整段。
         chunk_seconds = int(request.params.get("chunk_seconds", 60))
         chunks = _split_audio(audio_path, output_dir, chunk_seconds, _ffmpeg_bin())
+        # 每视频自适应 VAD：若开启 vad_filter 且未显式传 vad_parameters，则根据本音频
+        # 信号分析自动确定 VAD 参数（BGM 覆盖/静音比例/长停顿），改善碎片化与漏识别。
+        # 门控 WOV_AUTO_VAD=0 可关闭；显式传入 vad_parameters 时跳过。
+        vad_parameters = request.params.get("vad_parameters")
+        vad_filter = bool(request.params.get("vad_filter", True))
+        if vad_filter and not vad_parameters and os.getenv("WOV_AUTO_VAD", "1") == "1":
+            try:
+                from nodes.vad_profiler import vad_parameters_for_audio
+
+                vad_parameters = vad_parameters_for_audio(
+                    audio_path,
+                    sample_rate=int(request.params.get("sample_rate", 16000)),
+                    whisper_invoke=None,  # 片段网格需模型；生产默认仅信号分析（快速）
+                    run_dir=Path(request.output_dir),
+                    chunk_seconds=chunk_seconds,
+                )
+                logger.info("自动 VAD 参数: %s", vad_parameters)
+            except Exception as exc:  # noqa: BLE001
+                # 分析失败不影响转写：回退默认 bentenVAD，仅记录。
+                logger.warning("自动 VAD 分析失败，回退默认: %s", exc)
+                vad_parameters = None
         # 逐块转写并合并：offset 用每块实际时长累积（WAV 头精确），SRT 序号连续。
         logger.info("转写开始: %d 个分块", len(chunks))
         lines: list[str] = []
@@ -256,7 +277,8 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
                 language=str(request.params.get("language", "ja")),
                 task=str(request.params.get("task", "transcribe")),
                 beam_size=int(request.params.get("beam_size", 1)),
-                vad_filter=bool(request.params.get("vad_filter", True)),
+                vad_filter=vad_filter,
+                vad_parameters=vad_parameters,
                 condition_on_previous_text=bool(
                     request.params.get("condition_on_previous_text", False)
                 ),

tests/test_nodes.py

@@ -1002,6 +1002,27 @@ def test_whisper_segment_logs_full_video_time(caplog, tmp_path, monkeypatch) ->
     assert any("分段 #1: 00:00:00,000 --> 00:00:01,000" in m for m in seg_logs)
     assert any(m.startswith("分段 #3: 00:01:00,000") for m in seg_logs)
 
+def test_whisper_auto_vad_fallback_on_error(tmp_path, monkeypatch) -> None:
+    """自动 VAD 分析抛异常时，whisper 回退默认参数正常转写（防御性）。"""
+    import nodes.whisper as whisper_mod
+
+    def _boom(*args, **kwargs):
+        raise RuntimeError("auto vad analysis failed")
+
+    # 使 nodes.vad_profiler.vad_parameters_for_audio 抛异常（whisper.py 函数内 import）。
+    monkeypatch.setattr(
+        "nodes.vad_profiler.vad_parameters_for_audio", _boom, raising=False,
+    )
+    _install_fake_whisper(monkeypatch)
+    _make_wav(tmp_path / "audio.wav", 5)
+    response = whisper_invoke(
+        _whisper_request(tmp_path, params={"language": "ja", "vad_filter": True, "sample_rate": 16000})
+    )
+    assert response.status == "completed"
+    # 回退默认：transcribe 收到 vad_parameters=None。
+    _, kwargs = FakeWhisperModel.instances[-1]
+    assert kwargs.get("vad_parameters") is None
+
 
 def test_wav_duration_fallback_on_invalid_file(tmp_path) -> None:
     """验证读取时长时，损坏/缺失文件回退 fallback 值（真实非法文件，非占位字节）。"""

tests/test_vad_profiler.py

@@ -0,0 +1,252 @@
+"""每视频自适应 VAD 调参测试（先红后绿）。
+
+验证信号分析 profile_audio、启发式参数建议 suggest_vad_parameters、
+转录质量评分 score_transcript（幻觉词扣分）与 vad_parameters_for_audio
+（信号分析 + 片段网格验证）。
+
+评分**不依赖参考字幕**（实际部署无参考），用转录质量 + 幻觉词扣分。
+"""
+
+from __future__ import annotations
+
+import wave
+from pathlib import Path
+
+import pytest
+
+from nodes.vad_profiler import (
+    AudioProfile,
+    HALLUCINATION_TOKENS,
+    profile_audio,
+    score_transcript,
+    suggest_vad_parameters,
+    vad_parameters_for_audio,
+    _pick_representative_start,
+)
+
+
+class _Seg:
+    """模拟 whisper segment：仅含 text。"""
+
+    def __init__(self, text):
+        self.text = text
+
+
+def _make_wav(path: Path, silence_seconds: int, voice_seconds: int) -> None:
+    """生成 [静音 N 秒 + 语音 N 秒] 的 16kHz 单声道 WAV。
+
+    silence 用 0 样本（静音），voice 用较大振幅样本（语音）。
+    """
+    import array
+
+    rate = 16000
+    silence = array.array("h", [0] * rate * silence_seconds)
+    voice = array.array("h", [8000] * rate * voice_seconds)
+    samples = silence + voice
+    with wave.open(str(path), "wb") as f:
+        f.setnchannels(1)
+        f.setsampwidth(2)
+        f.setframerate(rate)
+        f.writeframes(samples.tobytes())
+
+
+def test_profile_audio_detects_bgm_heavy() -> None:
+    """纯静音+语音的视频：silence_ratio 高、非 BGM 覆盖；mediam_rms 合理。"""
+    wav = Path("/tmp/test_vad_plain.wav")
+    _make_wav(wav, silence_seconds=40, voice_seconds=10)
+    p = profile_audio(wav, 16000)
+    assert p.silence_ratio > 0.5
+    assert p.bgm_heavy is False
+    assert p.duration_seconds == pytest.approx(50.0, abs=1)
+
+
+def test_suggest_vad_parameters_plain_silence() -> None:
+    """静音占比高且无长停顿 -> 建议 threshold 偏高（静音权重分支）。"""
+    wav = Path("/tmp/test_vad_plain2.wav")
+    # 交替短静音避免触发 long_silence（<5s 连续静音）。
+    _make_wav(wav, silence_seconds=3, voice_seconds=1)
+    _make_wav(wav, silence_seconds=3, voice_seconds=1)
+    p = profile_audio(wav, 16000)
+    assert p.long_silence is False
+    params = suggest_vad_parameters(p)
+    assert params["threshold"] >= 0.5
+
+
+def test_suggest_vad_parameters_bgm() -> None:
+    """BGM 覆盖（低能量占比高但静音少）-> threshold 降低、min_silence 减小。"""
+    p = AudioProfile(
+        silence_ratio=0.1,   # 几乎无静音
+        lowish_ratio=0.6,    # 大量低能量（音乐）
+        bgm_heavy=True,      # 直接标记 BGM 覆盖
+        long_silence=False,
+        rms_bins=[500] * 100,
+    )
+    params = suggest_vad_parameters(p)
+    assert params["threshold"] < 0.5        # 降低
+    assert params["min_silence_duration_ms"] < 1000  # 减小
+
+
+def test_score_transcript_penalizes_hallucination() -> None:
+    """幻觉词（感谢观看/晚安/音乐）多 -> 评分低。"""
+    good = [_Seg("ありがとうございます本日は"), _Seg("かしこまりました")]
+    bad = [_Seg("ご視聴ありがとうございました"), _Seg("おやすみなさい"), _Seg("音楽")]
+    assert score_transcript(good) > score_transcript(bad)
+
+
+def test_score_transcript_penalizes_fragments() -> None:
+    """碎片（纯单字/语气词）多 -> 评分低。"""
+    clean = [_Seg("今日はとても暑いですね"), _Seg("それでは始めましょう")]
+    frag = [_Seg("あ"), _Seg("うん"), _Seg("はい"), _Seg("あっ")]
+    assert score_transcript(clean) > score_transcript(frag)
+
+
+def test_vad_parameters_for_audio_uses_grid_when_provider() -> None:
+    """提供 whisper 回调时：从候选网格选评分最高的参数（threshold 最小者）。"""
+    wav = Path("/tmp/test_vad_grid.wav")
+    _make_wav(wav, silence_seconds=10, voice_seconds=5)
+
+    def fake_whisper(**kwargs):
+        # 假设最优 = threshold 最小的候选（suggest 0.5 时候选含 0.4）。
+        t = kwargs.get("vad_parameters", {}).get("threshold", 0.5)
+        if t <= 0.4:
+            # 最优组合：无幻觉的正常长句（分数最高）。
+            return [_Seg("今日はお客様のために精神整備を務めさせていただきます")]
+        # 其它组合：幻觉套话（分数低）。
+        return [_Seg("ご視聴ありがとうございました"), _Seg("おやすみなさい")]
+
+    params = vad_parameters_for_audio(
+        wav, sample_rate=16000, whisper_invoke=fake_whisper, run_dir=Path("/tmp")
+    )
+    # 网格应从候选里选出评分最高的 threshold=0.4 的组合。
+    assert params["threshold"] == pytest.approx(0.4, abs=0.1)
+
+
+def test_vad_parameters_for_audio_fallback_without_provider() -> None:
+    """无 whisper 回调：退化为信号分析建议，不报错。"""
+    wav = Path("/tmp/test_vad_noprov.wav")
+    _make_wav(wav, silence_seconds=10, voice_seconds=5)
+    params = vad_parameters_for_audio(wav, sample_rate=16000, whisper_invoke=None)
+    assert "threshold" in params
+    assert "min_silence_duration_ms" in params
+
+
+def test_hallucination_tokens_included() -> None:
+    """幻觉词集合应含常用套话（感谢观看/晚安/音乐）。"""
+    all_str = " ".join(HALLUCINATION_TOKENS)
+    assert "ご視聴ありがとうございました" in all_str
+    assert "おやすみなさい" in all_str
+    assert "音楽" in all_str
+
+def test_suggest_vad_parameters_long_silence() -> None:
+    """长停顿常见 -> 建议 threshold 0.5、speech_pad 200（防时间轴压缩漂移）。"""
+    p = AudioProfile(
+        silence_ratio=0.2, lowish_ratio=0.3, bgm_heavy=False, long_silence=True,
+        rms_bins=[500] * 100,
+    )
+    params = suggest_vad_parameters(p)
+    assert params["threshold"] == 0.5
+    assert params["speech_pad_ms"] == 200
+
+
+def test_suggest_vad_parameters_regular() -> None:
+    """常规音频（无特殊标记）-> 默认 0.5/1000/400。"""
+    p = AudioProfile(
+        silence_ratio=0.2, lowish_ratio=0.3, bgm_heavy=False, long_silence=False,
+        rms_bins=[1500] * 100,
+    )
+    params = suggest_vad_parameters(p)
+    assert params["threshold"] == 0.5
+    assert params["min_silence_duration_ms"] == 1000
+    assert params["speech_pad_ms"] == 400
+
+
+def test_pick_representative_start_all_silence() -> None:
+    """全静音音频：窗口恐怖沉默用 score>0.55 惩罚，仍返回起始 0。"""
+    p = AudioProfile(
+        rms_bins=[50] * 100, silence_ratio=0.9, lowish_ratio=0.9,
+        bgm_heavy=False, long_silence=False,
+    )
+    start = _pick_representative_start(p, 90)
+    assert start >= 0 and start < 10
+
+
+def test_grid_search_exception_skipped() -> None:
+    """网格搜索某组参数调用抛异常：跳过该组，不崩溃。"""
+    called = {"n": 0}
+
+    def failing_whisper(**kwargs):
+        called["n"] += 1
+        raise RuntimeError("mock download fail")
+
+    from nodes.vad_profiler import _grid_search_vad
+
+    params = _grid_search_vad(
+        failing_whisper, Path("/tmp/x.wav"), Path("/tmp"), 60, 0,
+        {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400},
+    )
+    # 全部失败时回退 suggested。
+    assert params == {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400}
+    assert called["n"] > 0
+
+
+def test_candidate_params_expands_grid() -> None:
+    """候选网格应含建议值邻域（threshold±0.1、silence 倍、pad 组合）。"""
+    from nodes.vad_profiler import _candidate_params
+
+    cands = _candidate_params(
+        {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400}
+    )
+    assert len(cands) == 27
+    assert {"threshold": 0.4, "min_silence_duration_ms": 500, "speech_pad_ms": 0} in cands
+    assert {"threshold": 0.6, "min_silence_duration_ms": 2000, "speech_pad_ms": 400} in cands
+
+
+def test_median_rms_empty_returns_zero() -> None:
+    """rms_bins 为空时 median_rms 返回 0。"""
+    p = AudioProfile(rms_bins=[])
+    assert p.median_rms == 0.0
+
+
+def test_profile_audio_empty_wav() -> None:
+    """空的 WAV（无样本）-> 返回 duration 0 的空 profile，不崩溃。"""
+    import array
+    wav = Path("/tmp/test_vad_empty.wav")
+    with wave.open(str(wav), "wb") as f:
+        f.setnchannels(1); f.setsampwidth(2); f.setframerate(16000)
+        f.writeframes(array.array("h", []).tobytes())
+    p = profile_audio(wav, 16000)
+    assert p.duration_seconds == 0.0
+
+
+def test_grid_search_ignores_non_list() -> None:
+    """网格搜索回调返回非 list（如 None）应被跳过，回退建议值。"""
+    from nodes.vad_profiler import _grid_search_vad
+
+    params = _grid_search_vad(
+        lambda **kw: None, Path("/tmp/x.wav"), Path("/tmp"), 60, 0,
+        {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400},
+    )
+    assert params == {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400}
+
+
+def test_score_transcript_empty_list() -> None:
+    """空列表评分返回 0。"""
+    assert score_transcript([]) == 0.0
+
+
+def test_score_transcript_ignores_empty_text_seg() -> None:
+    """含空文本的 segment 被跳过，不报错。"""
+    assert score_transcript([_Seg(""), _Seg("今日は暑いです")]) > 0
+
+
+def test_profile_audio_different_sample_rate() -> None:
+    """采样率与请求不一致时用 wave 实际 rate 近似，不崩溃。"""
+    import array
+    wav = Path("/tmp/test_vad_rate.wav")
+    rate = 8000
+    samples = array.array("h", [5000] * rate * 2)  # 2s 语音
+    with wave.open(str(wav), "wb") as f:
+        f.setnchannels(1); f.setsampwidth(2); f.setframerate(rate)
+        f.writeframes(samples.tobytes())
+    p = profile_audio(wav, 16000)
+    assert p.duration_seconds == pytest.approx(2.0, abs=0.2)