path: root/app/backends/common.py

"""Shared helpers for the backend setup wizards.

Every TTS backend setup wizard (audio.cpp, qwen, faster) lives in its own
module under ``backends``; this module holds the pieces more than one of
them needs: .wav discovery, path normalization, and the regex edit that
keeps ``app/converter/config.py`` in sync with the choices made in a wizard.
It deliberately imports nothing from the other backend modules (or the
TUI) so it can be reused without pulling curses into a non-interactive
run.
"""

import os
import re
import sys
import time
import urllib.parse
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

# Messages queued while the TUI is on screen, printed to the real console
# after the curses session ends (see ui.hub.run). Build/setup steps that
# fail inside the TUI record here so the user gets a copy-pastable command
# and a log path once the TUI exits, instead of losing the output.
_POST_TUI_NOTICES: List[str] = []

# The tts-audiobook-generator checkout root (where audiobook.py lives).
# Everything non-user-facing lives under ./app: the source packages
# (backends, converter, ui), the generated dirs (envs, chunks, logs, debug),
# and the backend checkouts (app/audio.cpp, app/faster-qwen3-tts).
TTS_ROOT = Path(__file__).resolve().parent.parent.parent

# The single "everything else" directory under TTS_ROOT.
APP_DIR = TTS_ROOT / "app"

# app/logs — build/server/conversion logs (already gitignored).
LOG_DIR = APP_DIR / "logs"

# The project's sample-voice directory: .wav files dropped here are offered
# as the default source when a setup/configure wizard asks for a wav
# directory (both the TUI browser start and the --wavs flag default).
VOICES_DIR = TTS_ROOT / "voices"

# app/converter/config.py — rewritten in place by update_config_value so the
# converter picks up the host/port/voice a wizard configured.
CONFIG_PATH = APP_DIR / "converter" / "config.py"

# Output directory of tts-audiobook-generator; never offered as a .wav
# source by detect_wav_dir.
TTS_OUTPUT_DIR = "output"

# The voice-transcript mapping file audio.cpp reads from its voice_dir.
# (The faster backend uses voices.json instead; see backends.faster.)
PROMPT_TEXT_FILENAME = "prompt_text"


def record_post_tui_notice(text: str) -> None:
    """Queue a message to print to the console after the TUI session ends.

    The TUI runs in a curses session, so ``print`` during it does not reach
    the real terminal. Steps that fail inside the TUI (e.g. the audio.cpp
    build) record a copy-pastable command and a log path here; ``ui.hub.run``
    drains the queue after the session ends.
    """
    _POST_TUI_NOTICES.append(text)


def drain_post_tui_notices() -> List[str]:
    """Return and clear the queued post-TUI messages."""
    notices = list(_POST_TUI_NOTICES)
    _POST_TUI_NOTICES.clear()
    return notices


def cancel_requested(cancel) -> bool:
    """True when CANCEL (a ``threading.Event``) is given and set.

    Shared guard for the multi-phase uninstall actions: cancellation is
    honored only between phases (stop servers / pip / delete files), so a
    phase that already started always runs to completion and an uninstall
    never tears halfway. Callers return 130 when this fires before a
    pending phase.
    """
    return cancel is not None and cancel.is_set()


def normalize_dir_arg(value: str) -> Path:
    """Normalize a user-supplied path argument.

    Strips surrounding quotes (a common copy-paste artifact), expands a
    leading ``~``, and resolves the result to an absolute path so relative
    paths are always validated against the current working directory.
    """
    cleaned = value.strip()
    if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"'":
        cleaned = cleaned[1:-1]
    return Path(os.path.expanduser(cleaned)).resolve()


def resolve_wav_dir_arg(value: str) -> Path:
    """Normalize a user-supplied wav directory argument."""
    return normalize_dir_arg(value)


def find_wav_files(input_dir: Path) -> List[Path]:
    """Return the .wav files in INPUT_DIR, sorted alphabetically by name.

    A missing or unreadable directory yields [] so callers can treat it
    like an empty directory (matching ``count_wavs``).
    """
    try:
        return sorted(
            (path for path in input_dir.iterdir()
             if path.is_file() and path.suffix.lower() == ".wav"),
            key=lambda path: path.name.lower(),
        )
    except OSError:
        return []


def count_wavs(directory: Path) -> int:
    """Count the .wav files in DIRECTORY (0 when it cannot be read)."""
    try:
        return sum(1 for path in directory.iterdir()
                   if path.is_file() and path.suffix.lower() == ".wav")
    except OSError:
        return 0


def detect_wav_dir(audiocpp_dir: Path, tts_root: Path) -> Optional[Path]:
    """Find a unique directory that directly contains .wav files.

    Looks shallowly (the root itself and its immediate subdirectories) in
    both the audio.cpp checkout and the tts-audiobook-generator root (where
    audiobook.py lives), since clone reference .wavs commonly live in
    either. The tts-audiobook-generator ``output/`` directory is excluded.
    When exactly one candidate is found it is returned (as a starting
    directory for the .wav browser); when none or several are found None is
    returned so the caller falls back to its default start location.
    """
    candidates: List[Path] = []
    seen: Set[Path] = set()

    def consider(directory: Path) -> None:
        try:
            resolved = directory.resolve()
        except OSError:
            return
        if resolved in seen:
            return
        seen.add(resolved)
        if count_wavs(directory) > 0:
            candidates.append(directory)

    for root in (audiocpp_dir, tts_root):
        if not root.is_dir():
            continue
        consider(root)
        try:
            children = sorted(root.iterdir(), key=lambda p: p.name.lower())
        except OSError:
            continue
        for child in children:
            if not child.is_dir() or child.name.startswith("."):
                continue
            if root == tts_root and child.name == TTS_OUTPUT_DIR:
                continue
            consider(child)

    if len(candidates) == 1:
        return candidates[0]
    return None


def wav_dir_info(directory: Path) -> Tuple[str, str]:
    """TUI status describing the directory listed in the wav browser."""
    count = count_wavs(directory)
    if count:
        wavs = ".wav" if count == 1 else ".wavs"
        return (f"{count} {wavs} found in this directory. Press Enter.",
                "ok")
    return ("No .wav files found in this directory", "warn")


def wav_dir_preview(directory: Path) -> Tuple[str, str]:
    """TUI status describing a highlighted subdirectory in the wav browser."""
    count = count_wavs(directory)
    if count:
        wavs = ".wav" if count == 1 else ".wavs"
        return (f"{count} {wavs}", "ok")
    return ("no .wav files", "info")


def url_with_port(url: str, port: int) -> str:
    """Return URL with its port replaced/inserted as PORT."""
    parts = urllib.parse.urlsplit(url)
    host = parts.hostname or "127.0.0.1"
    return urllib.parse.urlunsplit(
        (parts.scheme or "http", f"{host}:{port}", parts.path, "", ""))


def normalize_remote_url(value: str) -> str:
    """Normalize a user-supplied remote server URL, or '' for "disabled".

    Accepts a bare ``host[:port]`` (a scheme of ``http`` is assumed), a full
    ``http(s)://host[:port][/path]`` URL, or the empty string (no remote
    server configured). Returns the normalized URL (bare host:port becomes
    ``http://host:port``). Raises ValueError for anything else — a missing
    host, a host containing whitespace, or a non-numeric port.
    """
    cleaned = value.strip()
    if not cleaned:
        return ""
    parts = urllib.parse.urlsplit(cleaned)
    if not parts.scheme:
        # Bare host[:port] — add the default scheme so netloc/host/port
        # parse cleanly. An explicit scheme is kept as-is (so "http://"
        # with no host fails the host check below).
        parts = urllib.parse.urlsplit(f"http://{cleaned}")
    host = parts.hostname
    if not host or any(ch.isspace() for ch in host):
        raise ValueError(
            "Enter a host:port (e.g. 10.20.30.40:8000) or a full URL "
            f"(e.g. http://10.20.30.40:8000); got {value!r}")
    try:
        parts.port  # raises ValueError for a non-numeric port
    except ValueError as exc:
        raise ValueError(
            f"Invalid port in remote URL {value!r}: {exc}") from exc
    return urllib.parse.urlunsplit(
        (parts.scheme or "http", parts.netloc, parts.path, "", ""))


def server_running(url: str, timeout: float = 0.3) -> bool:
    """True when something accepts TCP connections at URL's host:port.

    A protocol-agnostic socket connect: an HTTP TTS server that is up will
    accept the connection (we do not need to speak HTTP to know it is
    listening). Returns False on any parse or connection error, so a
    misconfigured URL never blocks the hub — it just reports the backend
    as not running. Used by each backend's ``detect()`` to set
    ``BackendStatus.running``.
    """
    import socket
    try:
        parts = urllib.parse.urlsplit(url)
        host = parts.hostname or "127.0.0.1"
        port = parts.port or (443 if (parts.scheme or "http") == "https"
                              else 80)
    except ValueError:
        return False
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def update_config_value(key: str, value,
                        config_path: Optional[Path] = None) -> bool:
    """Set ``KEY`` to VALUE in app/converter/config.py and in memory.

    Only the value of the named assignment changes: indentation and any
    trailing comment are preserved. Strings render double-quoted; other
    literals (ints, booleans) render bare. After a successful write (or
    when the file already holds VALUE) the new value is mirrored onto the
    imported ``converter.config`` module, so a wizard's change takes
    effect immediately instead of only after the next process start.
    Returns True when the file now holds VALUE, False when it could not
    be read or written (or KEY has no line in it).
    """
    path = Path(config_path) if config_path is not None else CONFIG_PATH
    rendered = f'"{value}"' if isinstance(value, str) else str(value)
    try:
        text = path.read_text(encoding="utf-8")
        match = re.search(
            rf'(?m)^(\s*{re.escape(key)}\s*=\s*)("[^"]*"|\S+)(\s*(?:#.*)?)$',
            text)
        if match is None:
            return False
        if match.group(2) != rendered:
            text = text[:match.start(2)] + rendered + text[match.end(2):]
            path.write_text(text, encoding="utf-8")
    except OSError:
        return False
    # Local import: this module must stay importable before the venv
    # exists (backends.envs bootstraps from it), and converter.config is
    # stdlib-only constants, safe to load whenever a wizard runs.
    from converter import config as _config
    setattr(_config, key, value)
    return True


def read_prompt_text(prompt_path: Path) -> Dict[str, str]:
    """Parse a prompt_text file into a stem -> transcript mapping.

    Lines are ``<name>|<transcript>``; blank lines are skipped and a line
    without a ``|`` separator is treated as a name with an empty transcript.
    Returns an empty mapping when the file does not exist.
    """
    if not prompt_path.exists():
        return {}
    mapping: Dict[str, str] = {}
    for line in prompt_path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        if "|" in line:
            name, _, text = line.partition("|")
        else:
            name, text = line, ""
        mapping[name.strip()] = text
    return mapping


def write_prompt_text(wav_dir: Path,
                      transcripts: Dict[str, str]) -> Path:
    """Write the voice_dir prompt_text mapping into WAV_DIR.

    One ``<basename-without-extension>|<transcript>`` line per voice.
    Returns the path of the written file.
    """
    prompt_path = wav_dir / PROMPT_TEXT_FILENAME
    lines = [f"{name}|{text}" for name, text in transcripts.items()]
    prompt_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return prompt_path


def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
                           *, emit=None, cancel=None, on_cancel=None) -> int:
    """Run a subprocess, streaming output to the console or to EMIT.

    With EMIT None the child inherits the real terminal and its output
    appears normally (the non-interactive CLI paths). With EMIT given (a
    ``callable(str)``) the child's stdout/stderr are merged, read line by
    line (splitting on both ``\\n`` and ``\\r`` so carriage-return progress
    updates like git's or tqdm's surface as lines), and each line is passed
    to EMIT — the in-TUI task view path.

    CANCEL is an optional ``threading.Event``: once set, ON_CANCEL (if given)
    is called (e.g. to touch a ``--cancel-file``), then the child's process
    group is terminated (SIGTERM, escalating to SIGKILL after a grace
    period) and 130 is returned. Returns the process exit code.
    """
    import subprocess
    if emit is None:
        try:
            result = subprocess.run(argv,
                                    cwd=str(cwd) if cwd is not None else None)
        except OSError as exc:
            print(f"[ERROR] Could not run {' '.join(argv)}: {exc}")
            return 1
        return result.returncode

    popen_kwargs = {"stdout": subprocess.PIPE, "stderr": subprocess.STDOUT}
    if cwd is not None:
        popen_kwargs["cwd"] = str(cwd)
    if sys.platform == "win32":
        popen_kwargs["creationflags"] = \
            subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore[attr-defined]
    else:
        popen_kwargs["start_new_session"] = True
    try:
        proc = subprocess.Popen(argv, **popen_kwargs)
    except OSError as exc:
        emit(f"[ERROR] Could not run {' '.join(argv)}: {exc}")
        return 1

    cancelled = False

    def _reader() -> None:
        try:
            for raw in iter(proc.stdout.readline, b""):
                if not raw:
                    break
                text = raw.decode("utf-8", errors="replace")
                for line in text.splitlines():
                    if line:
                        emit(line)
        except (OSError, ValueError):
            pass

    reader = _spawn_reader(_reader)
    while True:
        if cancel is not None and cancel.is_set():
            cancelled = True
            if on_cancel is not None:
                try:
                    on_cancel()
                except Exception:
                    pass
                # Give a graceful-cancel hook (e.g. a --cancel-file) a
                # moment to let the child exit cleanly before forcing it.
                grace_end = time.time() + 3
                while time.time() < grace_end:
                    if proc.poll() is not None:
                        break
                    time.sleep(0.1)
            if proc.poll() is None:
                _terminate_process_group(proc)
            break
        if proc.poll() is not None:
            break
        time.sleep(0.1)
    try:
        reader.join(timeout=5)
    finally:
        if reader.is_alive():
            reader.join(timeout=0)
    if cancelled:
        return 130
    return proc.returncode


def _spawn_reader(target):
    import threading
    thread = threading.Thread(target=target, daemon=True)
    thread.start()
    return thread


def _terminate_process_group(proc) -> None:
    """Terminate PROC's process group (SIGTERM, then SIGKILL after a grace).

    Death is detected with ``proc.poll()`` (which reaps the zombie) rather
    than a ``killpg(pgid, 0)`` probe — the latter still succeeds on a
    zombie, so it would always wait the full grace period.
    """
    import signal
    if sys.platform == "win32":
        try:
            proc.terminate()
        except OSError:
            pass
        deadline = time.time() + 10
        while time.time() < deadline:
            if proc.poll() is not None:
                return
            time.sleep(0.1)
        try:
            proc.kill()
        except OSError:
            pass
        return
    try:
        pgid = os.getpgid(proc.pid)
    except (ProcessLookupError, OSError):
        return
    try:
        os.killpg(pgid, signal.SIGTERM)
    except (ProcessLookupError, OSError):
        return
    deadline = time.time() + 10
    while time.time() < deadline:
        if proc.poll() is not None:
            return
        time.sleep(0.1)
    try:
        os.killpg(pgid, signal.SIGKILL)
    except (ProcessLookupError, OSError):
        pass
    proc.wait()


def git_clone(url: str, target: Path, *, emit=None, cancel=None) -> int:
    """Clone URL into TARGET, streaming to the console or to EMIT. Returns
    the exit code."""
    if emit is None:
        print(f"[INFO] Cloning {url} into {target}...")
        return run_console_subprocess(["git", "clone", url, str(target)])
    emit(f"[INFO] Cloning {url} into {target}...")
    # --progress makes git report percentage updates even though stderr is
    # piped (it normally only does so on a terminal), feeding the task view.
    return run_console_subprocess(
        ["git", "clone", "--progress", url, str(target)],
        emit=emit, cancel=cancel)


def pip_install(packages: List[str], *, emit=None, cancel=None) -> int:
    """pip install PACKAGES into the managed venv (``envs/tts``). Returns exit code.

    Delegates to ``backends.envs.pip_install`` so backend TTS packages are
    installed alongside the app requirements in the tool-managed environment
    rather than into whatever interpreter happens to be running the wizard.
    With EMIT given (the in-TUI task view) pip runs with its output streamed
    into EMIT; CANCEL aborts it. The import is local to avoid a circular
    import (envs imports this module).
    """
    from backends import envs
    return envs.pip_install(packages, emit=emit, cancel=cancel)


def pip_uninstall(packages: List[str], *, emit=None) -> int:
    """pip uninstall PACKAGES from the managed venv. Returns exit code.

    Delegates to ``backends.envs.pip_uninstall`` (local import to avoid a
    circular import). Used by the backends' ``uninstall`` action. With EMIT
    given (the in-TUI task view) pip runs piped, streaming into EMIT, so
    its output never touches the terminal behind curses.
    """
    from backends import envs
    return envs.pip_uninstall(packages, emit=emit)