cli.py

from __future__ import annotations

import contextlib
import importlib
import json
import math
import os
import signal
import subprocess
import sys
import tempfile
import threading
from dataclasses import dataclass
from pathlib import Path
from types import FrameType
from typing import Any, Callable, Optional

import typer

from core.external_client import ClientRequest, run_client_transcribe
from core.external_daemon import DaemonConfig, run_daemon
from core.stt_providers import (
    STT_PROVIDER_CHOICES,
    STT_PROVIDER_GIGAAM_CTC,
    STT_PROVIDER_WHISPER,
    normalize_stt_provider,
    resolve_gigaam_device,
)

app = typer.Typer(
    help=(
        "AskVLM CLI for local transcription, subtitles, and machine-friendly "
        "external transcription."
    )
)


def _load_cli_runtime() -> dict[str, Any]:
    """Load heavy runtime modules lazily so `--help` stays lightweight."""
    return {
        "LocalPipeline": importlib.import_module("core.pipelines").LocalPipeline,
        "burn_subtitles": importlib.import_module("core.ffmpeg").burn_subtitles,
        "export_document": importlib.import_module("utils.exporters").export_document,
        "export_srt_with_rules": importlib.import_module(
            "utils.exporters"
        ).export_srt_with_rules,
        "SubtitleRules": importlib.import_module("utils.exporters").SubtitleRules,
    }


def _collect_files(input_path: Path, *, recursive: bool) -> list[Path]:
    if input_path.is_dir():
        if recursive:
            return [path for path in input_path.rglob("*") if path.is_file()]
        return [path for path in input_path.iterdir() if path.is_file()]
    return [input_path]


def _create_local_pipeline(  # noqa: PLR0913
    *,
    whisper_model: str,
    engine: str,
    diarization: bool,
    dialog_blocks: bool,
    language: Optional[str],
    device: str,
    compute_type: str,
    stt_provider: str = STT_PROVIDER_WHISPER,
) -> Any:
    runtime = _load_cli_runtime()
    local_pipeline_cls = runtime["LocalPipeline"]
    return local_pipeline_cls(
        whisper_model=whisper_model,
        engine=engine,
        stt_provider=stt_provider,
        enable_diarization=diarization,
        enable_dialog_blocks=dialog_blocks,
        language=language,
        device=device,
        compute_type=compute_type,
    )


def _write_plain_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


@dataclass(frozen=True)
class _ExternalChildAttemptResult:
    success_text: Optional[str]
    return_code: int
    stdout: str
    stderr: str


def _normalize_transcript_text(text: Optional[str]) -> str:
    value = text or ""
    return value if value.strip() else ""


def _write_json_atomic(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        delete=False,
        dir=path.parent,
        prefix=f"{path.name}.",
        suffix=".tmp",
    ) as temp_file:
        json.dump(payload, temp_file)
        temp_file.flush()
        os.fsync(temp_file.fileno())
        temp_path = Path(temp_file.name)
    temp_path.replace(path)


def _read_child_success_result(path: Path) -> Optional[str]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None
    if not isinstance(data, dict):
        return None
    if data.get("status") != "ok":
        return None
    text = data.get("text")
    if not isinstance(text, str):
        return None
    return _normalize_transcript_text(text)


def _resolve_external_device(*, stt_provider: str, device: str) -> str:
    """Resolve ``--device`` for the selected STT provider.

    Args:
        stt_provider: Canonical provider id.
        device: Raw ``--device`` token from the CLI.

    Returns:
        Device token safe for the selected provider.

    Raises:
        typer.BadParameter: When GigaAM is requested with a non-CPU device.

    """
    if stt_provider != STT_PROVIDER_GIGAAM_CTC:
        return device
    try:
        return resolve_gigaam_device(device)
    except ValueError as exc:
        raise typer.BadParameter(str(exc)) from exc


def _run_external_transcribe_once(  # noqa: PLR0913
    *,
    input_path: Path,
    whisper_model: str,
    language: Optional[str],
    device: str,
    compute_type: str,
    diarization: bool,
    dialog_blocks: bool,
    work_dir: Optional[Path],
    stt_provider: str = STT_PROVIDER_WHISPER,
    before_close: Optional[Callable[[str], None]] = None,
) -> str:
    pipeline = _create_local_pipeline(
        whisper_model=whisper_model,
        engine="whisperx",
        diarization=diarization,
        dialog_blocks=dialog_blocks,
        language=language,
        device=device,
        compute_type=compute_type,
        stt_provider=stt_provider,
    )
    try:
        if work_dir is None:
            with tempfile.TemporaryDirectory(prefix="askvlm-cli-") as temp_dir:
                text = pipeline.process(input_path, Path(temp_dir)).get_full_text()
        else:
            work_dir.mkdir(parents=True, exist_ok=True)
            text = pipeline.process(input_path, work_dir).get_full_text()
        normalized_text = _normalize_transcript_text(text)
        if before_close is not None:
            before_close(normalized_text)
        return normalized_text
    finally:
        pipeline.close(aggressive=False)


def _build_external_child_command(  # noqa: PLR0913
    *,
    input_path: Path,
    whisper_model: str,
    language: Optional[str],
    device: str,
    compute_type: str,
    diarization: bool,
    dialog_blocks: bool,
    work_dir: Optional[Path],
    child_result_file: Path,
    stt_provider: str = STT_PROVIDER_WHISPER,
) -> list[str]:
    command = [
        sys.executable,
        str(Path(__file__).resolve()),
        "external-transcribe",
        str(input_path),
        "--stt-provider",
        stt_provider,
        "--whisper-model",
        whisper_model,
        "--device",
        device,
        "--compute-type",
        compute_type,
        "--_internal-child-mode",
        f"--_internal-result-file={child_result_file}",
    ]
    if language is not None:
        command.extend(["--language", language])
    if diarization:
        command.append("--diarization")
    if dialog_blocks:
        command.append("--dialog-blocks")
    if work_dir is not None:
        command.extend(["--work-dir", str(work_dir)])
    return command


def _run_external_transcribe_isolated_attempt(  # noqa: PLR0913
    *,
    input_path: Path,
    whisper_model: str,
    language: Optional[str],
    device: str,
    compute_type: str,
    diarization: bool,
    dialog_blocks: bool,
    work_dir: Optional[Path],
    stt_provider: str = STT_PROVIDER_WHISPER,
) -> _ExternalChildAttemptResult:
    with tempfile.TemporaryDirectory(prefix="askvlm-ext-ipc-") as ipc_dir:
        child_result_file = Path(ipc_dir) / "child_result.json"
        command = _build_external_child_command(
            input_path=input_path,
            whisper_model=whisper_model,
            language=language,
            device=device,
            compute_type=compute_type,
            diarization=diarization,
            dialog_blocks=dialog_blocks,
            work_dir=work_dir,
            child_result_file=child_result_file,
            stt_provider=stt_provider,
        )
        completed = subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
            env={**os.environ, "_ASKVLM_CHILD_RESULT_FILE": str(child_result_file)},
        )
        return _ExternalChildAttemptResult(
            success_text=_read_child_success_result(child_result_file),
            return_code=completed.returncode,
            stdout=completed.stdout or "",
            stderr=completed.stderr or "",
        )


def _is_windows_crash_like_return_code(return_code: int) -> bool:
    if return_code < 0:
        return True
    return (return_code & 0xFFFFFFFF) >= 0xC0000000


_INTERNAL_CHILD_IPC_ERROR = "Internal child mode requires"


def _is_internal_child_ipc_error(stderr: str) -> bool:
    """Return True when child stderr signals a missing IPC result-file path.

    Args:
        stderr: The captured stderr string from the child process, or None.

    Returns:
        True if the IPC setup error marker is present in stderr.
    """
    return _INTERNAL_CHILD_IPC_ERROR in (stderr or "")


def _raise_external_transcribe_error(attempt: _ExternalChildAttemptResult) -> None:
    details = (attempt.stderr or "").strip() or (attempt.stdout or "").strip()
    if details:
        typer.secho(details, err=True)
    raise typer.Exit(code=1)


def _emit_external_transcribe_outputs(
    *, text: str, output_file: Optional[Path], stdout: bool
) -> None:
    if output_file is not None:
        _write_plain_text(output_file, text)
    if stdout and text:
        typer.echo(text)


# * Distinct exit codes let a caller (for example the Telegram bot) react to a
# * busy/unavailable transcription service without parsing free-form stderr.
EXTERNAL_TRANSCRIBE_TIMEOUT_EXIT = 10
EXTERNAL_TRANSCRIBE_UNAVAILABLE_EXIT = 11
_CLIENT_TIMEOUT_MARKER = "ASKVLM_CLIENT_TIMEOUT"
_DAEMON_UNAVAILABLE_MARKER = "ASKVLM_DAEMON_UNAVAILABLE"
_CPU_FALLBACK_FAILED_MARKER = "ASKVLM_CPU_FALLBACK_FAILED"

# * Daemon outcomes that mean the service was reachable-but-degraded (the resident
# * GPU model could not deliver a transcript in time, or no daemon answered) rather
# * than the media being unprocessable — eligible for a bounded CPU recovery pass.
_DAEMON_DEGRADED_STATUSES = frozenset({"timeout", "unavailable"})


def _device_prefers_gpu(device: str) -> bool:
    """Return True when *device* would run on GPU, so a CPU fallback differs.

    Args:
        device: The requested ``--device`` token (``auto``/``cuda``/``cpu``/...).

    Returns:
        True for everything except an explicit CPU request.
    """
    return device.strip().lower() != "cpu"


def _try_cpu_fallback_transcribe(cpu_fallback: Callable[[], str]) -> Optional[str]:
    """Run the bounded CPU fallback, swallowing failures.

    Args:
        cpu_fallback: Callable performing one in-process CPU transcription.

    Returns:
        The recovered transcript text (possibly empty), or ``None`` when the
        fallback itself failed so the caller can surface the degraded exit code.
    """
    try:
        return cpu_fallback()
    except Exception as exc:  # noqa: BLE001 - best-effort recovery; original status still surfaced
        typer.secho(
            f"{_CPU_FALLBACK_FAILED_MARKER}: {type(exc).__name__}: {exc}".strip(),
            err=True,
        )
        return None


def _run_external_transcribe_via_daemon(  # noqa: PLR0913
    *,
    input_path: Path,
    language: Optional[str],
    device: str,
    compute_type: str,
    whisper_model: str,
    client_timeout: float,
    daemon_workers: int,
    output_file: Optional[Path],
    stdout: bool,
    stt_provider: str = STT_PROVIDER_WHISPER,
    cpu_fallback: Optional[Callable[[], str]] = None,
) -> None:
    """Transcribe through the shared daemon and map the outcome to CLI rules.

    When the daemon is reachable-but-degraded (client timeout or no daemon) on a
    GPU-seeded Whisper request, a bounded in-process CPU fallback is attempted
    before the degraded exit code is surfaced. GigaAM CTC is CPU-only and is
    never used as a CUDA fallback path.

    Args:
        input_path: Media file to transcribe.
        language: Optional language hint or ``None``.
        device: Preferred device for a freshly spawned daemon.
        compute_type: Preferred faster-whisper compute type.
        whisper_model: Preferred Whisper model name.
        client_timeout: Seconds to wait before giving up and signalling a drop.
        daemon_workers: Worker count requested when spawning a new daemon.
        output_file: Optional plain-text file to also write.
        stdout: Whether to echo the transcript to stdout.
        stt_provider: STT backend id (``whisper`` or ``gigaam-ctc``).
        cpu_fallback: Optional callable performing one in-process CPU
            transcription, used only on a degraded GPU-seeded Whisper outcome.

    Raises:
        typer.Exit: With code ``0`` on success, ``1`` on error,
            :data:`EXTERNAL_TRANSCRIBE_TIMEOUT_EXIT` on a client timeout, or
            :data:`EXTERNAL_TRANSCRIBE_UNAVAILABLE_EXIT` when no daemon is
            reachable.
    """
    outcome = run_client_transcribe(
        ClientRequest(
            input_path=input_path,
            language=language,
            device=device,
            compute_type=compute_type,
            whisper_model=whisper_model,
            client_timeout_s=client_timeout,
            daemon_workers=daemon_workers,
            stt_provider=stt_provider,
        )
    )
    if outcome.status in {"ok", "empty"}:
        _emit_external_transcribe_outputs(
            text=outcome.text, output_file=output_file, stdout=stdout
        )
        return
    # * Whisper-only CUDA→CPU recovery; never route GigaAM into this path.
    if (
        stt_provider == STT_PROVIDER_WHISPER
        and outcome.status in _DAEMON_DEGRADED_STATUSES
        and cpu_fallback is not None
        and _device_prefers_gpu(device)
    ):
        recovered = _try_cpu_fallback_transcribe(cpu_fallback)
        if recovered is not None:
            _emit_external_transcribe_outputs(
                text=recovered, output_file=output_file, stdout=stdout
            )
            return
    if outcome.status == "timeout":
        typer.secho(f"{_CLIENT_TIMEOUT_MARKER}: {outcome.detail or ''}".strip(), err=True)
        raise typer.Exit(code=EXTERNAL_TRANSCRIBE_TIMEOUT_EXIT)
    if outcome.status == "unavailable":
        typer.secho(
            f"{_DAEMON_UNAVAILABLE_MARKER}: {outcome.detail or ''}".strip(), err=True
        )
        raise typer.Exit(code=EXTERNAL_TRANSCRIBE_UNAVAILABLE_EXIT)
    typer.secho(outcome.detail or "transcription failed", err=True)
    raise typer.Exit(code=1)


@app.command()
def transcribe(
    input_path: Path = typer.Argument(
        ..., exists=True, readable=True, help="Input media file or directory"
    ),
    output_dir: Path = typer.Option(
        Path("transcriptions"), "--output", "-o", help="Output directory"
    ),
    whisper_model: str = typer.Option(
        "large-v3",
        help="Whisper model name for batch transcription (default: large-v3)",
    ),
    engine: str = typer.Option(
        "whisper",
        help=(
            "Backend compatibility hint: whisper | whisperx | auto. The current "
            "local batch pipeline uses the Whisper/Faster-Whisper path."
        ),
    ),
    language: Optional[str] = typer.Option(None, help="Language code (optional)"),
    diarization: bool = typer.Option(True, help="Enable speaker diarization"),
    dialog_blocks: bool = typer.Option(False, help="Format text with LLM"),
    export: str = typer.Option("txt", help="Export format: txt|srt|vtt|json"),
    recursive: bool = typer.Option(
        False, "--recursive", "-r", help="Process directories recursively"
    ),
    overwrite: bool = typer.Option(
        False, "--overwrite", help="Overwrite existing output files"
    ),
    device: str = typer.Option(
        "auto", help="Device: auto|cuda|cpu (passed to engines where applicable)"
    ),
    compute_type: str = typer.Option(
        "float16",
        help=(
            "Compute type for faster-whisper: float16|int8|int8_float16|auto. "
            "Default float16 (extreme profile) for best quality on 8+ GiB VRAM."
        ),
    ),
) -> None:
    """Transcribe a file or directory and export results."""
    runtime = _load_cli_runtime()
    export_document = runtime["export_document"]
    output_dir.mkdir(parents=True, exist_ok=True)
    # * Normalize engine when auto is requested
    chosen_engine = "whisperx" if engine == "auto" else engine
    pipeline = _create_local_pipeline(
        whisper_model=whisper_model,
        engine=chosen_engine,
        diarization=diarization,
        dialog_blocks=dialog_blocks,
        language=language,
        device=device,
        compute_type=compute_type,
    )

    try:
        for media in _collect_files(input_path, recursive=recursive):
            out_file = output_dir / f"{media.stem}.{export.lower()}"
            if out_file.exists() and not overwrite:
                typer.echo(f"Exists, skip: {out_file}")
                continue
            typer.echo(f"Processing {media}...")
            doc = pipeline.process(media, output_dir)
            export_document(doc, export, out_file)
            typer.echo(f"Saved to {out_file}")
    finally:
        with contextlib.suppress(Exception):
            pipeline.close(aggressive=True)


@app.command()
def subtitle(
    input_path: Path = typer.Argument(
        ..., exists=True, readable=True, help="Input media file or directory"
    ),
    output_dir: Path = typer.Option(
        Path("transcriptions"), "--output", "-o", help="Output directory"
    ),
    whisper_model: str = typer.Option(
        "large-v3",
        help="Whisper model name for subtitle generation (default: large-v3)",
    ),
    language: Optional[str] = typer.Option(None, help="Language code (optional)"),
    device: str = typer.Option("auto", help="Device: auto|cuda|cpu"),
    compute_type: str = typer.Option(
        "float16",
        help="Compute type for faster-whisper (default float16; consider int8_float16 on 8–12 GiB if OOM)",
    ),
    diarization: bool = typer.Option(False, help="Enable speaker diarization"),
    burn_in: bool = typer.Option(True, help="Burn subtitles into the video"),
    save_srt: bool = typer.Option(True, help="Always save .srt sidecar"),
    format: str = typer.Option("srt", help="Subtitle format: srt|vtt|ass (srt only for burn)"),
    max_cps: float = typer.Option(18.0, help="Max characters per second"),
    max_line_chars: int = typer.Option(42, help="Max characters per line"),
    max_lines: int = typer.Option(2, help="Max lines per cue"),
    min_duration: float = typer.Option(1.2, help="Minimum cue duration (s)"),
    max_duration: float = typer.Option(6.0, help="Maximum cue duration (s)"),
) -> None:
    """Generate subtitles with readability rules and optionally burn them into the video."""
    runtime = _load_cli_runtime()
    export_srt_with_rules = runtime["export_srt_with_rules"]
    subtitle_rules_cls = runtime["SubtitleRules"]
    burn_subtitles = runtime["burn_subtitles"]
    output_dir.mkdir(parents=True, exist_ok=True)
    pipeline = _create_local_pipeline(
        whisper_model=whisper_model,
        engine="whisperx",
        diarization=diarization,
        dialog_blocks=False,
        language=language,
        device=device,
        compute_type=compute_type,
    )
    files = _collect_files(input_path, recursive=False)

    rules = subtitle_rules_cls(
        max_line_chars=max_line_chars,
        max_lines=max_lines,
        min_duration=min_duration,
        max_duration=max_duration,
        max_cps=max_cps,
    )

    try:
        for media in files:
            typer.echo(f"Processing {media}...")
            doc = pipeline.process(media, output_dir)
            srt_path = output_dir / f"{media.stem}.srt"
            # Export srt with rules
            srt_text = export_srt_with_rules(doc, rules)
            srt_path.write_text(srt_text, encoding="utf-8")
            if save_srt:
                typer.echo(f"Saved SRT: {srt_path}")
            if burn_in and media.suffix.lower() in {".mp4", ".mov", ".mkv", ".avi"}:
                out_video = output_dir / f"{media.stem}_subbed.mp4"
                burn_subtitles(media, srt_path, out_video)
                typer.echo(f"Burned-in video: {out_video}")
    finally:
        with contextlib.suppress(Exception):
            pipeline.close(aggressive=True)


@app.command("external-transcribe")
def external_transcribe(  # noqa: PLR0913
    input_path: Path = typer.Argument(
        ...,
        exists=True,
        readable=True,
        file_okay=True,
        dir_okay=False,
        help="Single audio or video file to transcribe for an external caller.",
    ),
    output_file: Optional[Path] = typer.Option(
        None,
        "--output-file",
        "-o",
        help="Optional plain-text file to write alongside stdout output.",
    ),
    whisper_model: str = typer.Option(
        "small",
        help="Whisper model name for the external one-shot flow (default: small).",
    ),
    stt_provider: str = typer.Option(
        STT_PROVIDER_WHISPER,
        "--stt-provider",
        help=(
            "Speech-to-text provider: whisper (default) or gigaam-ctc "
            "(CPU-only; included in .[ml])."
        ),
    ),
    language: Optional[str] = typer.Option(
        None, help="Optional language code, for example: en, ru, de."
    ),
    device: str = typer.Option(
        "auto",
        help=(
            "Preferred device: auto|cuda|cpu. When CUDA memory is exhausted, "
            "Whisper retries on CPU automatically. GigaAM CTC accepts only cpu "
            "(auto resolves to cpu)."
        ),
    ),
    compute_type: str = typer.Option(
        "auto",
        help=(
            "Compute type: auto|float16|int8|int8_float16. 'auto' uses float16 "
            "on CUDA and int8 on CPU. Ignored for gigaam-ctc."
        ),
    ),
    diarization: bool = typer.Option(
        False,
        "--diarization/--no-diarization",
        help=(
            "Enable speaker diarization. Disabled by default for external calls; "
            "this can require additional GPU memory."
        ),
    ),
    dialog_blocks: bool = typer.Option(
        False,
        "--dialog-blocks/--no-dialog-blocks",
        help="Enable LLM-based text formatting. Disabled by default.",
    ),
    stdout: bool = typer.Option(
        True,
        "--stdout/--no-stdout",
        help="Write only the final transcript text to stdout. Enabled by default.",
    ),
    work_dir: Optional[Path] = typer.Option(
        None,
        "--work-dir",
        help=(
            "Optional directory for intermediate files. When omitted, AskVLM uses "
            "a temporary directory and deletes it after completion."
        ),
    ),
    no_daemon: bool = typer.Option(
        False,
        "--no-daemon/--daemon",
        help=(
            "Run the legacy in-process one-shot flow instead of routing the "
            "request through the shared transcription daemon."
        ),
    ),
    client_timeout: float = typer.Option(
        300.0,
        "--client-timeout",
        min=1.0,
        help=(
            "Daemon mode: seconds to wait for a transcript before giving up and "
            "signalling the daemon to drop the job."
        ),
    ),
    daemon_workers: int = typer.Option(
        1,
        "--daemon-workers",
        min=1,
        max=8,
        help=(
            "Daemon mode: worker count requested when a new daemon is spawned. "
            "Defaults to 1 to keep a single active model in memory."
        ),
    ),
    _internal_child_mode: bool = typer.Option(
        False,
        "--_internal-child-mode",
        hidden=True,
    ),
    _internal_result_file: Optional[Path] = typer.Option(
        None,
        "--_internal-result-file",
        hidden=True,
    ),
) -> None:
    """Transcribe one media file and return plain text for external applications."""
    if not stdout and output_file is None:
        raise typer.BadParameter(
            "Either keep --stdout enabled or provide --output-file."
        )
    try:
        provider = normalize_stt_provider(stt_provider)
    except ValueError as exc:
        raise typer.BadParameter(str(exc)) from exc
    if provider not in STT_PROVIDER_CHOICES:
        raise typer.BadParameter(
            f"Unsupported --stt-provider {stt_provider!r}; "
            f"expected one of: {', '.join(STT_PROVIDER_CHOICES)}"
        )
    device = _resolve_external_device(stt_provider=provider, device=device)
    # * Env-var fallback: recover IPC path if CLI arg was not parsed
    if _internal_child_mode and _internal_result_file is None:
        _env_path = os.environ.get("_ASKVLM_CHILD_RESULT_FILE")
        if _env_path:
            _internal_result_file = Path(_env_path)

    if _internal_child_mode:
        if _internal_result_file is None:
            raise typer.BadParameter(
                "Internal child mode requires --_internal-result-file."
            )
        try:
            _run_external_transcribe_once(
                input_path=input_path,
                whisper_model=whisper_model,
                language=language,
                device=device,
                compute_type=compute_type,
                diarization=diarization,
                dialog_blocks=dialog_blocks,
                work_dir=work_dir,
                stt_provider=provider,
                before_close=lambda text: _write_json_atomic(
                    _internal_result_file,
                    {"status": "ok", "text": text},
                ),
            )
            return
        except Exception as exc:
            if not _internal_result_file.exists():
                with contextlib.suppress(Exception):
                    _write_json_atomic(
                        _internal_result_file,
                        {
                            "status": "error",
                            "error_type": type(exc).__name__,
                            "message": str(exc),
                        },
                    )
            raise

    if not no_daemon:
        _run_external_transcribe_via_daemon(
            input_path=input_path,
            language=language,
            device=device,
            compute_type=compute_type,
            whisper_model=whisper_model,
            client_timeout=client_timeout,
            daemon_workers=daemon_workers,
            output_file=output_file,
            stdout=stdout,
            stt_provider=provider,
            # * Bounded Whisper CUDA→CPU recovery only. GigaAM is CPU-only and
            # * must not participate in CUDA fallbacks or Windows GPU isolation.
            cpu_fallback=(
                None
                if provider != STT_PROVIDER_WHISPER
                else lambda: _run_external_transcribe_once(
                    input_path=input_path,
                    whisper_model=whisper_model,
                    language=language,
                    device="cpu",
                    compute_type=compute_type,
                    diarization=diarization,
                    dialog_blocks=dialog_blocks,
                    work_dir=work_dir,
                    stt_provider=STT_PROVIDER_WHISPER,
                )
            ),
        )
        return

    # * Windows GPU child isolation is Whisper-only; GigaAM always runs in-process CPU.
    windows_non_cpu = (
        provider == STT_PROVIDER_WHISPER
        and sys.platform.startswith("win")
        and device.lower() != "cpu"
    )
    if windows_non_cpu:
        first_attempt = _run_external_transcribe_isolated_attempt(
            input_path=input_path,
            whisper_model=whisper_model,
            language=language,
            device=device,
            compute_type=compute_type,
            diarization=diarization,
            dialog_blocks=dialog_blocks,
            work_dir=work_dir,
            stt_provider=provider,
        )
        if first_attempt.success_text is not None:
            _emit_external_transcribe_outputs(
                text=first_attempt.success_text,
                output_file=output_file,
                stdout=stdout,
            )
            return

        if (
            not _is_windows_crash_like_return_code(first_attempt.return_code)
            and not _is_internal_child_ipc_error(first_attempt.stderr)
        ):
            _raise_external_transcribe_error(first_attempt)

        retry_attempt = _run_external_transcribe_isolated_attempt(
            input_path=input_path,
            whisper_model=whisper_model,
            language=language,
            device="cpu",
            compute_type=compute_type,
            diarization=diarization,
            dialog_blocks=dialog_blocks,
            work_dir=work_dir,
            stt_provider=provider,
        )
        if retry_attempt.success_text is not None:
            _emit_external_transcribe_outputs(
                text=retry_attempt.success_text,
                output_file=output_file,
                stdout=stdout,
            )
            return
        _raise_external_transcribe_error(retry_attempt)

    _run_external_transcribe_once(
        input_path=input_path,
        whisper_model=whisper_model,
        language=language,
        device=device,
        compute_type=compute_type,
        diarization=diarization,
        dialog_blocks=dialog_blocks,
        work_dir=work_dir,
        stt_provider=provider,
        before_close=lambda text: _emit_external_transcribe_outputs(
            text=text,
            output_file=output_file,
            stdout=stdout,
        ),
    )


def _install_stop_handlers(stop_event: threading.Event) -> None:
    """Wire available stop signals to request a graceful daemon shutdown.

    Args:
        stop_event: Flag set when a stop signal is received.
    """

    def _handle(_signum: int, _frame: FrameType | None) -> None:
        stop_event.set()

    for sig_name in ("SIGINT", "SIGTERM", "SIGBREAK"):
        sig = getattr(signal, sig_name, None)
        if sig is None:
            continue
        with contextlib.suppress(ValueError, OSError):
            signal.signal(sig, _handle)


@app.command("external-transcribe-daemon")
def external_transcribe_daemon(
    workers: int = typer.Option(
        1, "--workers", min=1, max=8, help="Concurrent resident workers (default: 1)."
    ),
    stt_provider: str = typer.Option(
        STT_PROVIDER_WHISPER,
        "--stt-provider",
        help="Resident STT provider: whisper (default) or gigaam-ctc (CPU-only).",
    ),
    whisper_model: str = typer.Option(
        "small", "--whisper-model", help="Resident Whisper model (default: small)."
    ),
    device: str = typer.Option(
        "auto",
        "--device",
        help=(
            "Device for the resident model: auto|cuda|cpu. "
            "GigaAM CTC accepts only cpu (auto→cpu)."
        ),
    ),
    compute_type: str = typer.Option(
        "auto", "--compute-type", help="faster-whisper compute type (default: auto)."
    ),
    language: Optional[str] = typer.Option(
        None, "--language", help="Optional fixed language hint for the resident model."
    ),
    queue_dir: Optional[Path] = typer.Option(
        None, "--queue-dir", help="Queue root to serve (default: project cache)."
    ),
    idle_shutdown: float = typer.Option(
        600.0,
        "--idle-shutdown",
        min=1.0,
        help="Exit after this many idle seconds to release VRAM (default: 600).",
    ),
) -> None:
    """Run the singleton transcription daemon that serves the shared queue.

    Only one daemon serves a given queue; a second invocation exits immediately
    when the queue is already owned. The daemon loads the model once and serves
    every ``external-transcribe`` client through the file-based queue. A live
    daemon with a different ``--stt-provider`` is a mismatch: clients must
    restart the singleton rather than silently using the wrong resident model.
    """
    try:
        provider = normalize_stt_provider(stt_provider)
    except ValueError as exc:
        raise typer.BadParameter(str(exc)) from exc
    device = _resolve_external_device(stt_provider=provider, device=device)
    config = DaemonConfig(
        max_workers=workers,
        whisper_model=whisper_model,
        device=device,
        compute_type=compute_type,
        language=language,
        idle_shutdown_s=idle_shutdown,
        stt_provider=provider,
    )
    stop_event = threading.Event()
    _install_stop_handlers(stop_event)
    code = run_daemon(queue_dir, config=config, stop_event=stop_event)
    raise typer.Exit(code=code)


@app.command("external-extract-frames")
def external_extract_frames(  # noqa: PLR0913
    input_path: Path = typer.Argument(
        ...,
        exists=True,
        readable=True,
        file_okay=True,
        dir_okay=False,
        help="Video file to extract frames from.",
    ),
    output_dir: Path = typer.Option(
        ...,
        "--output-dir",
        "-o",
        help="Directory where extracted frame images are written.",
    ),
    fps: float = typer.Option(
        0.5,
        "--fps",
        help="Target sampling rate in frames per second (default: 0.5).",
        min=0.001,
    ),
    fps_fallback: float = typer.Option(
        0.2,
        "--fps-fallback",
        help="Fallback FPS used when frame-budget would be exceeded (default: 0.2).",
        min=0.001,
    ),
    frame_budget: int = typer.Option(
        20,
        "--frame-budget",
        help=(
            "Maximum number of frames to extract. "
            "When the target FPS would produce more frames, fps-fallback is used instead. "
            "0 disables the cap."
        ),
        min=0,
    ),
    as_json: bool = typer.Option(
        False,
        "--json/--no-json",
        help="Output a JSON object instead of one path per line.",
    ),
) -> None:
    """Extract video frames at adaptive FPS for external vision pipelines.

    Writes frame images to OUTPUT_DIR. Prints extracted frame paths to stdout
    (one per line), or a JSON manifest when --json is used.

    Exit code 0 on success (even if the video has zero frames). Exit code 1 on
    any processing failure.
    """
    import json as _json

    from core.ffmpeg import extract_frames_for_span, get_media_duration_seconds

    output_dir.mkdir(parents=True, exist_ok=True)

    duration_s = get_media_duration_seconds(input_path)
    if duration_s <= 0.0:
        typer.secho(
            f"Warning: could not determine duration of {input_path}; defaulting to 0 frames.",
            err=True,
        )
        if as_json:
            typer.echo(
                _json.dumps({"frames": [], "fps_used": fps, "duration_s": 0.0})
            )
        return

    # * Select FPS: use target unless frame budget would be exceeded
    effective_fps = fps
    if frame_budget > 0:
        estimated = math.ceil(duration_s * fps)
        if estimated > frame_budget:
            effective_fps = fps_fallback
        # * Hard cap: when even the fallback FPS overshoots the budget, derive the
        # * rate from the budget itself so sampling stays uniform over the full
        # * duration instead of over-producing frames for the caller to truncate.
        if math.ceil(duration_s * effective_fps) > frame_budget:
            effective_fps = frame_budget / duration_s

    frame_paths = extract_frames_for_span(
        video_file=input_path,
        start_s=0.0,
        end_s=duration_s,
        output_pattern=output_dir / "frame-%06d.png",
        fps=effective_fps,
    )
    # * ffmpeg fps rounding may still emit one extra frame; enforce the cap exactly.
    if frame_budget > 0 and len(frame_paths) > frame_budget:
        frame_paths = frame_paths[:frame_budget]

    if as_json:
        typer.echo(
            _json.dumps(
                {
                    "frames": [str(p) for p in frame_paths],
                    "fps_used": effective_fps,
                    "duration_s": duration_s,
                }
            )
        )
    else:
        for p in frame_paths:
            typer.echo(str(p))


def main() -> None:
    app()


if __name__ == "__main__":
    main()