"""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