"""Transcribe the first audio stream of one local video with faster-whisper."""

from __future__ import annotations

import argparse
from contextlib import contextmanager
import json
import math
import os
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence


MODEL_NAME = "large-v3"
DEVICE = "cuda"
COMPUTE_TYPE = "float16"
SHORT_AUDIO_LIMIT_SECONDS = 1200.0
CHUNK_CORE_SECONDS = 1200.0
CHUNK_OVERLAP_SECONDS = 5.0
MEDIA_DECODE_THREADS = 4
MEDIA_FILTER_THREADS = 2
MEDIA_MUTEX_NAME = r"Local\MBXMediaHeavyTaskV1"


class MediaTranscriptionError(RuntimeError):
    """A user-facing error that should stop the command without a traceback."""


class MediaHostGuardError(RuntimeError):
    """The shared Windows heavy-media guard could not be acquired safely."""


@dataclass(frozen=True)
class OutputPaths:
    directory: Path
    audio: Path
    text: Path
    srt: Path
    json: Path

    @property
    def files(self) -> tuple[Path, Path, Path, Path]:
        return (self.audio, self.text, self.srt, self.json)


@dataclass(frozen=True)
class RuntimeInfo:
    cuda_device_count: int
    cuda_compute_types: tuple[str, ...]


@dataclass(frozen=True)
class TranscriptSegment:
    start: float
    end: float
    text: str


@dataclass(frozen=True)
class TranscriptResult:
    language: str | None
    language_probability: float | None
    segments: tuple[TranscriptSegment, ...]


@dataclass(frozen=True)
class AudioChunk:
    index: int
    total: int
    core_start: float
    core_end: float
    audio_start: float
    audio_end: float

    @property
    def audio_duration(self) -> float:
        return self.audio_end - self.audio_start


ProcessRunner = Callable[[Sequence[str]], subprocess.CompletedProcess[str]]
ModelLoader = Callable[[], tuple[Any, RuntimeInfo]]


def _is_windows() -> bool:
    return os.name == "nt"


def _subprocess_creation_flags() -> int:
    if not _is_windows():
        return 0
    return int(getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0x00004000))


def _windows_kernel32() -> Any:
    import ctypes
    from ctypes import wintypes

    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel32.CreateMutexW.argtypes = (ctypes.c_void_p, wintypes.BOOL, wintypes.LPCWSTR)
    kernel32.CreateMutexW.restype = wintypes.HANDLE
    kernel32.WaitForSingleObject.argtypes = (wintypes.HANDLE, wintypes.DWORD)
    kernel32.WaitForSingleObject.restype = wintypes.DWORD
    kernel32.ReleaseMutex.argtypes = (wintypes.HANDLE,)
    kernel32.ReleaseMutex.restype = wintypes.BOOL
    kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
    kernel32.CloseHandle.restype = wintypes.BOOL
    kernel32.GetCurrentProcess.restype = wintypes.HANDLE
    kernel32.GetPriorityClass.argtypes = (wintypes.HANDLE,)
    kernel32.GetPriorityClass.restype = wintypes.DWORD
    kernel32.SetPriorityClass.argtypes = (wintypes.HANDLE, wintypes.DWORD)
    kernel32.SetPriorityClass.restype = wintypes.BOOL
    return kernel32


def _win32_last_error() -> int:
    import ctypes

    return int(ctypes.get_last_error())


@contextmanager
def media_host_guard() -> Iterable[None]:
    """Serialize heavy media CLIs and lower the current Windows process priority."""
    if not _is_windows():
        yield
        return

    kernel32 = _windows_kernel32()

    wait_object_0 = 0x00000000
    wait_abandoned = 0x00000080
    wait_timeout = 0x00000102
    below_normal_priority_class = 0x00004000
    handle = kernel32.CreateMutexW(None, False, MEDIA_MUTEX_NAME)
    if not handle:
        raise MediaHostGuardError(
            f"无法创建重型媒体互斥（Win32={_win32_last_error()}）。"
        )
    acquired = False
    process_handle = kernel32.GetCurrentProcess()
    original_priority = 0
    priority_lowered = False
    try:
        wait_result = int(kernel32.WaitForSingleObject(handle, 0))
        if wait_result == wait_timeout:
            raise MediaHostGuardError(
                "已有重型媒体任务运行，拒绝并行启动；请等待其结束后重试。"
            )
        if wait_result not in (wait_object_0, wait_abandoned):
            raise MediaHostGuardError(
                f"无法获取重型媒体互斥（Win32 wait={wait_result}）。"
            )
        acquired = True
        original_priority = int(kernel32.GetPriorityClass(process_handle))
        if original_priority == 0:
            raise MediaHostGuardError(
                f"无法读取当前进程优先级（Win32={_win32_last_error()}）。"
            )
        if not kernel32.SetPriorityClass(process_handle, below_normal_priority_class):
            raise MediaHostGuardError(
                f"无法将当前进程设为低于正常优先级（Win32={_win32_last_error()}）。"
            )
        priority_lowered = True
        yield
    finally:
        if priority_lowered and not kernel32.SetPriorityClass(process_handle, original_priority):
            print(
                "警告：重型媒体任务结束后无法恢复原进程优先级"
                f"（Win32={_win32_last_error()}）。",
                file=sys.stderr,
            )
        if acquired and not kernel32.ReleaseMutex(handle):
            print(
                f"警告：释放重型媒体互斥失败（Win32={_win32_last_error()}）。",
                file=sys.stderr,
            )
        if not kernel32.CloseHandle(handle):
            print(
                f"警告：关闭重型媒体互斥句柄失败（Win32={_win32_last_error()}）。",
                file=sys.stderr,
            )


def _run_process(command: Sequence[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        list(command),
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        creationflags=_subprocess_creation_flags(),
    )


def _error_detail(result: subprocess.CompletedProcess[str]) -> str:
    detail = (result.stderr or result.stdout or "未返回错误详情").strip()
    return detail[-1200:]


def _require_executable(name: str) -> str:
    executable = shutil.which(name)
    if executable is None:
        raise MediaTranscriptionError(
            f"未找到 {name}，请先安装并确保它位于当前 PATH。"
        )
    return executable


def _resolve_video(video_path: str | Path) -> Path:
    video = Path(video_path).expanduser().resolve()
    if not video.exists():
        raise MediaTranscriptionError(f"视频文件不存在：{video}")
    if not video.is_file():
        raise MediaTranscriptionError(f"输入路径不是文件：{video}")
    return video


def _build_output_paths(video: Path, output_dir: str | Path | None) -> OutputPaths:
    if output_dir is None:
        directory = video.with_name(f"{video.stem}.transcript")
    else:
        directory = Path(output_dir).expanduser().resolve()

    stem = video.stem
    return OutputPaths(
        directory=directory,
        audio=directory / f"{stem}.audio.flac",
        text=directory / f"{stem}.txt",
        srt=directory / f"{stem}.srt",
        json=directory / f"{stem}.json",
    )


def _ensure_output_available(outputs: OutputPaths) -> None:
    if outputs.directory.exists() and not outputs.directory.is_dir():
        raise MediaTranscriptionError(
            f"输出路径已存在且不是目录：{outputs.directory}"
        )

    conflicts = [path for path in outputs.files if path.exists()]
    if conflicts:
        formatted = "\n".join(f"- {path}" for path in conflicts)
        raise MediaTranscriptionError(
            "输出文件已存在；默认不会覆盖。请改用新的输出目录或先人工处理已有结果：\n"
            f"{formatted}"
        )


def _probe_first_audio_stream(
    video: Path,
    ffprobe: str,
    runner: ProcessRunner,
) -> None:
    command = [
        ffprobe,
        "-v",
        "error",
        "-select_streams",
        "a:0",
        "-show_entries",
        "stream=index",
        "-of",
        "csv=p=0",
        str(video),
    ]
    result = runner(command)
    if result.returncode != 0:
        raise MediaTranscriptionError(
            f"FFprobe 无法检查视频音轨：{_error_detail(result)}"
        )
    if not result.stdout.strip():
        raise MediaTranscriptionError(f"视频不包含可用音轨：{video}")


def _extract_first_audio_stream(
    video: Path,
    audio_path: Path,
    ffmpeg: str,
    runner: ProcessRunner,
) -> None:
    command = [
        ffmpeg,
        "-hide_banner",
        "-loglevel",
        "error",
        "-nostdin",
        "-n",
        "-threads",
        str(MEDIA_DECODE_THREADS),
        "-i",
        str(video),
        "-map",
        "0:a:0",
        "-vn",
        "-ac",
        "1",
        "-ar",
        "16000",
        "-c:a",
        "flac",
        str(audio_path),
    ]
    result = runner(command)
    if result.returncode != 0:
        raise MediaTranscriptionError(
            f"FFmpeg 提取第一条音轨失败：{_error_detail(result)}"
        )
    if not audio_path.is_file() or audio_path.stat().st_size == 0:
        raise MediaTranscriptionError("FFmpeg 未生成有效的 FLAC 音频文件。")


def _probe_audio_duration(
    audio_path: Path,
    ffprobe: str,
    runner: ProcessRunner,
) -> float:
    command = [
        ffprobe,
        "-v",
        "error",
        "-show_entries",
        "format=duration",
        "-of",
        "default=noprint_wrappers=1:nokey=1",
        str(audio_path),
    ]
    result = runner(command)
    if result.returncode != 0:
        raise MediaTranscriptionError(
            f"FFprobe 无法读取 FLAC 时长：{_error_detail(result)}"
        )
    try:
        duration = float(result.stdout.strip())
    except ValueError as exc:
        raise MediaTranscriptionError(
            f"FFprobe 返回了无效的 FLAC 时长：{result.stdout.strip() or '空'}"
        ) from exc
    if not math.isfinite(duration) or duration <= 0:
        raise MediaTranscriptionError(f"FLAC 时长必须为有限正数，实际为：{duration}")
    return duration


def _plan_audio_chunks(duration: float) -> tuple[AudioChunk, ...]:
    if not math.isfinite(duration) or duration <= SHORT_AUDIO_LIMIT_SECONDS:
        return ()

    total = int(math.ceil(duration / CHUNK_CORE_SECONDS))
    chunks: list[AudioChunk] = []
    for offset in range(total):
        core_start = offset * CHUNK_CORE_SECONDS
        core_end = min(duration, core_start + CHUNK_CORE_SECONDS)
        chunks.append(
            AudioChunk(
                index=offset + 1,
                total=total,
                core_start=core_start,
                core_end=core_end,
                audio_start=max(0.0, core_start - CHUNK_OVERLAP_SECONDS),
                audio_end=min(duration, core_end + CHUNK_OVERLAP_SECONDS),
            )
        )
    return tuple(chunks)


def _extract_audio_chunk(
    full_audio: Path,
    chunk_path: Path,
    chunk: AudioChunk,
    ffmpeg: str,
    runner: ProcessRunner,
) -> None:
    if chunk_path.exists():
        raise MediaTranscriptionError(f"临时音频块已存在，拒绝覆盖：{chunk_path}")
    command = [
        ffmpeg,
        "-hide_banner",
        "-loglevel",
        "error",
        "-nostdin",
        "-n",
        "-ss",
        f"{chunk.audio_start:.6f}",
        "-threads",
        str(MEDIA_DECODE_THREADS),
        "-i",
        str(full_audio),
        "-t",
        f"{chunk.audio_duration:.6f}",
        "-ac",
        "1",
        "-ar",
        "16000",
        "-c:a",
        "flac",
        str(chunk_path),
    ]
    result = runner(command)
    if result.returncode != 0:
        raise MediaTranscriptionError(
            f"FFmpeg 提取转写分块 {chunk.index}/{chunk.total} 失败："
            f"{_error_detail(result)}"
        )
    if not chunk_path.is_file() or chunk_path.stat().st_size == 0:
        raise MediaTranscriptionError(
            f"FFmpeg 未生成有效的转写分块 {chunk.index}/{chunk.total}。"
        )


def _load_cuda_model() -> tuple[Any, RuntimeInfo]:
    try:
        import ctranslate2
        from faster_whisper import WhisperModel
    except ImportError as exc:
        raise MediaTranscriptionError(
            "缺少 faster-whisper/CTranslate2，请在项目隔离 Conda 环境中运行。"
        ) from exc

    try:
        device_count = int(ctranslate2.get_cuda_device_count())
        compute_types = tuple(
            sorted(str(item) for item in ctranslate2.get_supported_compute_types(DEVICE))
        )
    except Exception as exc:
        raise MediaTranscriptionError(
            f"CUDA GPU 自检失败；不会降级到 CPU。原始错误：{exc}"
        ) from exc

    if device_count < 1:
        raise MediaTranscriptionError("未检测到可用 CUDA GPU；不会降级到 CPU。")
    if COMPUTE_TYPE not in compute_types:
        raise MediaTranscriptionError(
            "当前 CUDA GPU 不支持 float16；不会降级到 CPU。"
            f"检测到的计算类型：{', '.join(compute_types) or '无'}"
        )

    try:
        model = WhisperModel(
            MODEL_NAME,
            device=DEVICE,
            compute_type=COMPUTE_TYPE,
        )
    except Exception as exc:
        raise MediaTranscriptionError(
            "large-v3 模型加载失败；请检查模型缓存/网络以及 CUDA 12、cuBLAS、"
            f"cuDNN 9。不会降级到 CPU。原始错误：{exc}"
        ) from exc

    return model, RuntimeInfo(device_count, compute_types)


def _collect_transcript(
    model: Any,
    audio_path: Path,
    language: str | None,
) -> TranscriptResult:
    options: dict[str, Any] = {"vad_filter": True}
    if language:
        options["language"] = language

    try:
        segment_iterator, info = model.transcribe(str(audio_path), **options)
        segments = tuple(_normalize_segments(segment_iterator))
    except Exception as exc:
        raise MediaTranscriptionError(
            "GPU 语音转写失败；请检查 CUDA/cuDNN/显存和输入音频。"
            f"不会降级到 CPU。原始错误：{exc}"
        ) from exc

    detected_language = getattr(info, "language", None)
    probability = getattr(info, "language_probability", None)
    if probability is not None:
        probability = float(probability)
    return TranscriptResult(detected_language, probability, segments)


def _normalize_segments(segments: Iterable[Any]) -> Iterable[TranscriptSegment]:
    for segment in segments:
        text = str(segment.text).strip()
        if not text:
            continue
        start = max(0.0, float(segment.start))
        end = max(start, float(segment.end))
        yield TranscriptSegment(start=start, end=end, text=text)


def _project_chunk_segments(
    segments: Iterable[TranscriptSegment],
    chunk: AudioChunk,
    audio_duration: float,
) -> tuple[TranscriptSegment, ...]:
    projected: list[TranscriptSegment] = []
    for segment in segments:
        start = min(audio_duration, max(0.0, chunk.audio_start + segment.start))
        end = min(audio_duration, max(start, chunk.audio_start + segment.end))
        midpoint = (start + end) / 2.0
        belongs = chunk.core_start <= midpoint < chunk.core_end
        if chunk.index == chunk.total:
            belongs = chunk.core_start <= midpoint <= chunk.core_end
        if belongs:
            projected.append(TranscriptSegment(start=start, end=end, text=segment.text))
    return tuple(projected)


def _collect_chunked_transcript(
    model: Any,
    full_audio: Path,
    audio_duration: float,
    language: str | None,
    ffmpeg: str,
    runner: ProcessRunner,
) -> TranscriptResult:
    chunks = _plan_audio_chunks(audio_duration)
    if not chunks:
        raise MediaTranscriptionError("内部错误：短音频不应进入分块转写路径。")

    chunk_path = full_audio.parent / ".transcribe-chunk.flac"
    active_language = language
    result_language = language
    language_probability: float | None = None
    projected: list[TranscriptSegment] = []

    for chunk in chunks:
        print(f"正在转写分块 {chunk.index}/{chunk.total}", flush=True)
        try:
            _extract_audio_chunk(full_audio, chunk_path, chunk, ffmpeg, runner)
            chunk_result = _collect_transcript(model, chunk_path, active_language)
            if chunk.index == 1:
                result_language = chunk_result.language or language
                language_probability = chunk_result.language_probability
                if language is None:
                    if not chunk_result.language:
                        raise MediaTranscriptionError(
                            "首个音频块未返回检测语言，无法为后续块固定语言。"
                        )
                    active_language = chunk_result.language
            projected.extend(
                _project_chunk_segments(chunk_result.segments, chunk, audio_duration)
            )
        finally:
            chunk_path.unlink(missing_ok=True)

    projected.sort(key=lambda item: (item.start, item.end, item.text))
    return TranscriptResult(
        language=result_language,
        language_probability=language_probability,
        segments=tuple(projected),
    )


def _format_txt_timestamp(seconds: float) -> str:
    total_seconds = max(0, int(seconds))
    hours, remainder = divmod(total_seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    return f"{hours:02d}:{minutes:02d}:{secs:02d}"


def _format_srt_timestamp(seconds: float) -> str:
    total_milliseconds = max(0, int(round(seconds * 1000)))
    hours, remainder = divmod(total_milliseconds, 3_600_000)
    minutes, remainder = divmod(remainder, 60_000)
    secs, milliseconds = divmod(remainder, 1000)
    return f"{hours:02d}:{minutes:02d}:{secs:02d},{milliseconds:03d}"


def _write_transcript_files(
    video: Path,
    staged: OutputPaths,
    result: TranscriptResult,
) -> None:
    text_content = "\n\n".join(
        f"[{_format_txt_timestamp(segment.start)}] {segment.text}"
        for segment in result.segments
    )
    if text_content:
        text_content += "\n"
    staged.text.write_text(text_content, encoding="utf-8")

    srt_blocks = []
    for index, segment in enumerate(result.segments, start=1):
        srt_blocks.append(
            "\n".join(
                [
                    str(index),
                    f"{_format_srt_timestamp(segment.start)} --> "
                    f"{_format_srt_timestamp(segment.end)}",
                    segment.text,
                ]
            )
        )
    srt_content = "\n\n".join(srt_blocks)
    if srt_content:
        srt_content += "\n"
    staged.srt.write_text(srt_content, encoding="utf-8")

    payload = {
        "source": video.name,
        "audio": staged.audio.name,
        "language": result.language,
        "language_probability": result.language_probability,
        "model": MODEL_NAME,
        "device": DEVICE,
        "compute_type": COMPUTE_TYPE,
        "vad_filter": True,
        "segments": [
            {
                "start": segment.start,
                "end": segment.end,
                "text": segment.text,
            }
            for segment in result.segments
        ],
    }
    staged.json.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def _staged_paths(staging_dir: Path, final: OutputPaths) -> OutputPaths:
    return OutputPaths(
        directory=staging_dir,
        audio=staging_dir / final.audio.name,
        text=staging_dir / final.text.name,
        srt=staging_dir / final.srt.name,
        json=staging_dir / final.json.name,
    )


def _commit_outputs(staged: OutputPaths, final: OutputPaths) -> None:
    _ensure_output_available(final)
    created_directory = False
    moved: list[Path] = []
    try:
        if not final.directory.exists():
            final.directory.mkdir()
            created_directory = True
        for source, destination in zip(staged.files, final.files, strict=True):
            try:
                source.replace(destination)
            except BaseException:
                if destination.exists() and not source.exists():
                    moved.append(destination)
                raise
            else:
                moved.append(destination)
    except BaseException as exc:
        rollback_failures: list[str] = []
        for path in reversed(moved):
            try:
                path.unlink(missing_ok=True)
            except BaseException as cleanup_exc:
                rollback_failures.append(f"{path}: {cleanup_exc}")
        if created_directory:
            try:
                final.directory.rmdir()
            except BaseException as cleanup_exc:
                rollback_failures.append(f"{final.directory}: {cleanup_exc}")

        if not isinstance(exc, Exception):
            if rollback_failures:
                try:
                    setattr(
                        exc,
                        "_media_commit_rollback_failures",
                        tuple(rollback_failures),
                    )
                except Exception:
                    pass
            raise

        rollback_detail = ""
        if rollback_failures:
            rollback_detail = "；回滚不完整：" + "；".join(rollback_failures)
        raise MediaTranscriptionError(
            f"提交输出文件失败，已回滚已写结果：{exc}{rollback_detail}"
        ) from exc


def transcribe_video(
    video_path: str | Path,
    output_dir: str | Path | None = None,
    language: str | None = None,
    *,
    runner: ProcessRunner | None = None,
    model_loader: ModelLoader | None = None,
) -> tuple[OutputPaths, RuntimeInfo]:
    runner = runner or _run_process
    model_loader = model_loader or _load_cuda_model
    language = language.strip().lower() if language else None

    video = _resolve_video(video_path)
    outputs = _build_output_paths(video, output_dir)
    _ensure_output_available(outputs)
    ffprobe = _require_executable("ffprobe")
    ffmpeg = _require_executable("ffmpeg")
    _probe_first_audio_stream(video, ffprobe, runner)

    outputs.directory.parent.mkdir(parents=True, exist_ok=True)
    staging_dir = Path(
        tempfile.mkdtemp(
            prefix=f".{video.stem}.transcribe-",
            dir=str(outputs.directory.parent),
        )
    )
    staged = _staged_paths(staging_dir, outputs)

    try:
        print("正在提取第一条音轨（16 kHz、单声道 FLAC）……")
        _extract_first_audio_stream(video, staged.audio, ffmpeg, runner)
        audio_duration = _probe_audio_duration(staged.audio, ffprobe, runner)
        print("正在加载 large-v3（CUDA、float16）……")
        model, runtime = model_loader()
        print("正在使用 GPU 和 VAD 转写……")
        if audio_duration <= SHORT_AUDIO_LIMIT_SECONDS:
            transcript = _collect_transcript(model, staged.audio, language)
        else:
            transcript = _collect_chunked_transcript(
                model,
                staged.audio,
                audio_duration,
                language,
                ffmpeg,
                runner,
            )
        _write_transcript_files(video, staged, transcript)
        _commit_outputs(staged, outputs)
    finally:
        shutil.rmtree(staging_dir, ignore_errors=True)

    return outputs, runtime


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="用 FFmpeg 和 faster-whisper GPU 转写单个本地视频。"
    )
    parser.add_argument("video", help="本地视频文件路径")
    parser.add_argument("--output", help="可选输出目录；默认在视频旁创建 <文件名>.transcript")
    parser.add_argument("--language", help="可选语言代码，例如 zh；默认自动检测")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = _build_parser().parse_args(argv)
    try:
        with media_host_guard():
            outputs, runtime = transcribe_video(args.video, args.output, args.language)
    except (MediaTranscriptionError, MediaHostGuardError) as exc:
        print(f"错误：{exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt as exc:
        rollback_failures = getattr(exc, "_media_commit_rollback_failures", ())
        if rollback_failures:
            detail = "\n".join(f"- {item}" for item in rollback_failures)
            print(
                "错误：用户中断；正式输出回滚不完整，请人工检查以下路径：\n"
                f"{detail}",
                file=sys.stderr,
            )
        else:
            print("错误：用户中断，未提交输出文件。", file=sys.stderr)
        return 130

    print(f"完成：{outputs.directory}")
    print(
        "GPU 自检："
        f"CUDA 设备数={runtime.cuda_device_count}，"
        f"计算类型={','.join(runtime.cuda_compute_types)}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())