path: root/app/ui/hub.py

#!/usr/bin/env python3
"""The TUI main menu for the audiobook generator (run via ``audiobook.py``).

The hub is the single entry point for the whole workflow: it detects which
backends are already set up and offers to convert the input directory with
one of them, set up a new backend, or configure an existing one.
Each backend's setup wizard runs in its own curses session, so the hub
collects a "command" inside its own wrapper, returns to the plain terminal,
and then dispatches — no nested curses sessions.

Esc on the main menu quits the hub ('q' mirrors Esc on every screen).
Esc inside a sub-menu falls back to the main menu.
"""

import json
import re
import shutil
import urllib.parse
from pathlib import Path
from typing import Optional, Tuple

import audiobook
from backends import (
    REGISTRY,
    BackendStatus,
    ServerSpec,
    common,
    detect_all,
    get,
    servers,
)
from backends import audiocpp as audiocpp_backend
from backends import faster as faster_backend
from converter import config
from converter.converter import AUDIO_FORMATS
from converter.tts import (
    AUDIOCPP_FAMILY_QWEN3_TTS,
    BACKEND_AUDIOCPP,
    BACKEND_FASTER,
    BACKEND_QWEN,
    normalize_language,
)
from ui import tui

_GO_BACK = object()


def run() -> int:
    """Run the hub menu loop until the user quits. Returns exit code."""
    import curses
    while True:
        try:
            command = curses.wrapper(_hub_menu)
        except tui.WizardCancelled:
            return 0
        except KeyboardInterrupt:
            return 130
        if command is None:
            return 0
        kind = command[0]
        if kind == "quit":
            return 0
        if kind == "setup":
            info = get(command[1])
            if info is not None:
                info.setup_tui()
        elif kind == "configure":
            info = get(command[1])
            if info is not None and command[2] < len(info.configure_actions):
                info.configure_actions[command[2]].run()
        elif kind == "convert":
            _run_conversion(command[1], command[2])
        elif kind == "server":
            _run_server_action(command[1], command[2])


def _hub_menu(stdscr) -> Optional[tuple]:
    """Show the main menu; return a command tuple, or None to quit."""
    while True:
        statuses = detect_all()
        options = [("Set up a backend", "setup")]
        # Converting works against an external (remote) server too, but
        # configuring one and starting/stopping its servers need it on
        # this machine.
        if any(st.installed or st.running for st in statuses):
            options.insert(0, ("Convert books", "convert"))
        if any(st.installed for st in statuses):
            options.append(("Configure a backend", "configure"))
            options.append(("Start/Stop Backend Servers", "server"))
        options.append(("Settings", "settings"))
        options.append(("Quit", "quit"))
        choice = tui.menu(
            stdscr, "tts-audiobook-generator", options,
            table_title="Backend status", table_rows=_status_rows(statuses),
            notice_lines=_notice_lines())
        if choice is None or choice == "quit":
            return None
        if choice == "convert":
            cmd = _convert_menu(stdscr, statuses)
            if cmd is not None:
                return cmd
        elif choice == "setup":
            cmd = _setup_menu(stdscr, statuses)
            if cmd is not None:
                return cmd
        elif choice == "configure":
            cmd = _configure_menu(stdscr, statuses)
            if cmd is not None:
                return cmd
        elif choice == "server":
            cmd = _server_menu(stdscr, statuses)
            if cmd is not None:
                return cmd
        elif choice == "settings":
            _settings_menu(stdscr)


def _setup_menu(stdscr, statuses) -> Optional[tuple]:
    """Pick a backend to set up. Returns ("setup", key) or None to go back."""
    options = [(info.label, info.key) for info in REGISTRY]
    choice = tui.menu(stdscr, "Set up a backend", options,
                      back_value=_GO_BACK,
                      help_lines=["Clone/build/install a backend so you can "
                                  "convert with it."],
                      table_title="Backend status",
                      table_rows=_status_rows(statuses),
                      notice_lines=_notice_lines())
    if choice is _GO_BACK or choice is None:
        return None
    return ("setup", choice)


def _configure_menu(stdscr, statuses) -> Optional[tuple]:
    """Pick an installed backend and one of its configure actions."""
    by_key = {st.key: st for st in statuses}
    installed = [info for info in REGISTRY
                 if by_key.get(info.key) is not None
                 and by_key[info.key].installed]
    if not installed:
        tui.flash(stdscr, "No backend is installed yet — use 'Set up a "
                  "backend' first.")
        return None
    options = [(info.label, info.key) for info in installed]
    key = tui.menu(stdscr, "Configure a backend", options,
                   back_value=_GO_BACK,
                   table_title="Backend status",
                   table_rows=_status_rows(statuses),
                   notice_lines=_notice_lines())
    if key is _GO_BACK or key is None:
        return None
    info = get(key)
    actions = info.configure_actions
    choice = tui.menu(
        stdscr, f"Configure {info.label}",
        [(action.label, index) for index, action in enumerate(actions)],
        back_value=_GO_BACK)
    if choice is _GO_BACK or choice is None:
        return None
    return ("configure", key, choice)


def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]:
    """Map a backend's state to (status_text, status_kind, name_kind).

    'running' (green/ok) takes priority — an external server is already up;
    otherwise 'installed' (orange/warn) when the backend is present on disk,
    or 'unavailable' (red/err). A backend that is neither installed nor
    running is unusable, so its name is dimmed (NAME_KIND).
    A running server the hub did not start itself (no live pid file for any
    of its specs — see ``servers.manages``) is tagged "[remote]"; a
    multi-model backend (qwen) also names which models answered in
    parentheses, e.g. "running [remote] (Base, CustomVoice)".
    CURSES has no true orange, so the theme's yellow 'warn' is used; it
    renders amber/orange on most terminals.
    """
    if status is not None and status.running:
        text = "running"
        if not status.managed:
            text += " [remote]"
        if status.running_models:
            text += " (" + ", ".join(status.running_models) + ")"
        return (text, "ok", "body")
    if status is not None and status.installed:
        return ("installed", "warn", "body")
    return ("unavailable", "err", "dim")


def _status_rows(statuses) -> list:
    """Status-table rows for tui.menu: (label, status, kind, name_kind).

    One row per detected backend, in detect order — the same table the
    main menu shows, reused on each flow's first picker screen so the
    backend states stay visible there.
    """
    return [(st.label, *_status_mark(st)) for st in statuses]


def _notice_lines() -> Optional[list]:
    """Warning lines shown above the status table, or None when all good."""
    if shutil.which("ffmpeg") is None:
        return [("Warning: ffmpeg not installed!", "err")]
    return None


def _convert_menu(stdscr, statuses) -> Optional[tuple]:
    """Pick an available backend and collect per-backend run settings."""
    available = [st for st in statuses if st.ready or st.running]
    if not available:
        tui.flash(stdscr, "No backend is ready to convert with yet — use "
                  "'Set up a backend' first.")
        return None
    options = [(st.label, st.key) for st in available]
    table = {"table_title": "Backend status",
             "table_rows": _status_rows(statuses),
             "notice_lines": _notice_lines()}
    key = tui.menu(stdscr, "Convert books with...", options,
                   back_value=_GO_BACK, **table)
    if key is _GO_BACK or key is None:
        return None
    if key == BACKEND_AUDIOCPP:
        cmd = _convert_audiocpp(stdscr, statuses)
    elif key == BACKEND_QWEN:
        cmd = _convert_qwen(stdscr)
    elif key == BACKEND_FASTER:
        cmd = _convert_faster(stdscr)
    else:
        return None
    if cmd is None:
        return None
    _add_autostart(stdscr, cmd, statuses)
    return cmd


def _convert_audiocpp(stdscr, statuses) -> Optional[tuple]:
    """Collect audio.cpp run settings for a managed or remote server.

    With a local checkout configured (its server.json), the menus are fed
    from that file — the config of the server this tool manages. Without
    one, the running server is external and nothing is known about it
    locally, so its model and voice lists are queried live instead (the
    same GET /v1/models and GET /v1/audio/voices endpoints the converter
    resolves at run time).
    """
    checkout = audiocpp_backend.find_local_checkout()
    server_json = checkout / "server.json" if checkout else None
    local = bool(server_json and server_json.exists())
    url = config.AUDIOCPP_API_URL

    if local:
        try:
            data = json.loads(server_json.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            tui.flash(stdscr, f"Could not read {server_json}.")
            return None
        models = data.get("models") or []
        if not models:
            tui.flash(stdscr, "No model entries in server.json. Reconfigure "
                      "audio.cpp first.")
            return None
    else:
        # Remote flow: the backend only reaches the convert menu while a
        # server is running, so query it — the local config says nothing
        # about an external server.
        models = audiocpp_backend.fetch_server_models(url)
        if models is None:
            tui.flash(stdscr, f"Could not list models from the audio.cpp "
                      f"server at {url}. Is an audiocpp_server answering "
                      "there?")
            return None
        if not models:
            tui.flash(stdscr, f"The audio.cpp server at {url} hosts no "
                      "model entries.")
            return None
        data = {}

    model_options = [(f"{m.get('id')}  ({m.get('family') or '?'}, "
                      f"{m.get('task') or 'tts'})", m.get("id"))
                     for m in models]
    model_id = tui.menu(stdscr, "Select the audio.cpp model to use",
                        model_options, back_value=_GO_BACK)
    if model_id is _GO_BACK or model_id is None:
        return None
    entry = next((m for m in models if m.get("id") == model_id), {})
    family = entry.get("family")
    task = entry.get("task", "tts")
    if not local:
        # Servers predating the family/task fields omit them; mirror the
        # converter's defaults (_resolve_family/_resolve_task): unknown
        # family means qwen3_tts, a missing task plain tts.
        family = family or AUDIOCPP_FAMILY_QWEN3_TTS
        task = task or "tts"

    def _voices() -> Optional[list]:
        """Voice names for MODEL_ID, or None when a remote server's voice
        list cannot be queried. Locally: the voice_dir's .wav stems;
        remotely: GET /v1/audio/voices."""
        if local:
            voice_dir = data.get("voice_dir")
            return _list_voices(voice_dir) if voice_dir else []
        return audiocpp_backend.fetch_server_voices(url, model_id)

    # Voice: optional for qwen3_tts (built-in speaker), required otherwise.
    voice = None
    if task == "vdes":
        # Voice design: no voice, instructions required.
        pass
    elif family == AUDIOCPP_FAMILY_QWEN3_TTS:
        # Speaker mode available; voice optional.
        voices = _voices() or []
        if voices:
            opts = [("(built-in speaker)", None)] + [(v, v) for v in voices]
            voice = tui.menu(stdscr, "Voice", opts, back_value=_GO_BACK)
            if voice is _GO_BACK:
                return None
        else:
            voice = None
    else:
        voices = _voices()
        if voices is None:
            tui.flash(stdscr, f"Could not list voices from the audio.cpp "
                      f"server at {url}.")
            return None
        if not voices:
            if local:
                tui.flash(stdscr, f"This model needs a --voice but voice_dir "
                          f"{data.get('voice_dir')} has no .wav voices. "
                          "Reconfigure audio.cpp or add voices.")
            else:
                tui.flash(stdscr, f"This model needs a --voice but the "
                          f"server at {url} lists none for '{model_id}'. "
                          "Configure voice presets or a voice_dir on the "
                          "server.")
            return None
        voice = tui.menu(stdscr, "Select the voice to clone", [(v, v) for v in voices],
                         back_value=_GO_BACK)
        if voice is _GO_BACK or voice is None:
            return None

    # Instructions: required for vdes, optional otherwise.
    instructions = None
    if task == "vdes":
        instructions = tui.line_edit(
            stdscr, "Voice design instructions (required for this model)",
            config.AUDIOCPP_INSTRUCTIONS,
            validate=lambda s: None if s.strip()
            else "Describe the voice, e.g. 'A warm female narrator'",
            back_value=_GO_BACK)
        if instructions is _GO_BACK:
            return None
    else:
        instructions = tui.line_edit(
            stdscr, "Style instructions (optional, blank for none)",
            config.AUDIOCPP_INSTRUCTIONS, back_value=_GO_BACK)
        if instructions is _GO_BACK:
            return None
        if not instructions.strip():
            instructions = None

    common_kw = _common_options(stdscr)
    if common_kw is None:
        return None
    return ("convert", BACKEND_AUDIOCPP, {
        "model_id": model_id, "voice": voice, "instructions": instructions,
        **common_kw,
    })


def _convert_qwen(stdscr) -> Optional[tuple]:
    """Collect qwen run settings: built-in speaker or clone a .wav."""
    mode = tui.menu(
        stdscr, "qwen-tts mode",
        [("Custom voice (built-in speaker)", "custom"),
         ("Voice clone from a .wav file", "clone")],
        back_value=_GO_BACK,
        help_lines=[f"Speaker: {config.SPEAKER} (change it via Configure "
                    "qwen-tts)"])
    if mode is _GO_BACK or mode is None:
        return None
    clone = None
    if mode == "clone":
        clone = tui.line_edit(
            stdscr, "Path to a reference .wav (10-15s is ideal)",
            "",
            validate=lambda s: None if (s and Path(s).is_file()
                                        and s.lower().endswith(".wav"))
            else "Enter the path to an existing .wav file",
            back_value=_GO_BACK)
        if clone is _GO_BACK:
            return None
    common_kw = _common_options(stdscr)
    if common_kw is None:
        return None
    return ("convert", BACKEND_QWEN, {"clone": clone, **common_kw})


def _convert_faster(stdscr) -> Optional[tuple]:
    """Collect faster run settings: pick or type a voice name.

    With a local checkout's voices.json the picker lists it (the config of
    the server this tool manages). Without one, the running server was
    configured elsewhere and its voice names are unknown here, so the name
    is typed instead — safe for any value, since the server falls back to
    its first configured voice when the name is not defined.
    """
    checkout = faster_backend._checkout()
    voices_json = checkout / "voices.json"
    voices = None
    if voices_json.exists():
        try:
            voices = json.loads(voices_json.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            tui.flash(stdscr, f"Could not read {voices_json}.")
            return None
        if not voices:
            tui.flash(stdscr, "voices.json has no voices. Reconfigure faster.")
            return None
    if voices is None:
        # No local voices.json: prompt for a server-side voice name.
        voice_text = tui.line_edit(
            stdscr, "Server-side voice to clone with "
            "(blank uses the server's first voice)",
            config.FASTER_VOICE,
            validate=lambda s: None if s.strip() else "Enter a voice name",
            back_value=_GO_BACK)
        if voice_text is _GO_BACK:
            return None
        voice = voice_text.strip()
    else:
        default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \
            next(iter(voices))
        voice = tui.menu(
            stdscr, "Select the voice to clone",
            [(k, k) for k in voices],
            default_index=list(voices).index(default), back_value=_GO_BACK)
        if voice is _GO_BACK or voice is None:
            return None
    common_kw = _common_options(stdscr)
    if common_kw is None:
        return None
    return ("convert", BACKEND_FASTER, {"voice": voice, **common_kw})


def _common_options(stdscr) -> Optional[dict]:
    """Collect output format, speed, single-file, chunk, debug."""
    fmt_options = [(f, f) for f in AUDIO_FORMATS]
    fmt_default = AUDIO_FORMATS.index(config.AUDIO_FORMAT) \
        if config.AUDIO_FORMAT in AUDIO_FORMATS else 0
    output_format = tui.menu(stdscr, "Output format", fmt_options,
                             default_index=fmt_default, back_value=_GO_BACK)
    if output_format is _GO_BACK or output_format is None:
        return None
    speed_text = tui.line_edit(
        stdscr, "Playback speed (1.0 = normal)", "1.0",
        validate=lambda s: None if (_is_float(s) and float(s) > 0)
        else "Enter a positive number, e.g. 1.0",
        back_value=_GO_BACK)
    if speed_text is _GO_BACK:
        return None
    single_file = tui.confirm(stdscr, "Combine all chapters into one file?",
                              default=False, cancel_value=_GO_BACK)
    if single_file is _GO_BACK:
        return None
    chunk = tui.confirm(stdscr, "Force client-side chunking (--chunk)?",
                        default=False, cancel_value=_GO_BACK)
    if chunk is _GO_BACK:
        return None
    debug = tui.confirm(stdscr, "Debug mode (dump per-chunk audio/text)?",
                        default=False, cancel_value=_GO_BACK)
    if debug is _GO_BACK:
        return None
    return {
        "output_format": output_format,
        "speed": float(speed_text),
        "single_file": single_file,
        "chunk": chunk,
        "debug": debug,
    }


# ---------------------------------------------------------------------------
# Settings menu (global output options -> app/converter/config.py)
# ---------------------------------------------------------------------------

def _settings_menu(stdscr) -> None:
    """Edit the global output settings; Save writes them back to config.py."""
    fields = [
        {"key": "audio_format", "label": "Audio format", "kind": "choice",
         "value": config.AUDIO_FORMAT, "choices": list(AUDIO_FORMATS)},
        {"key": "audio_bitrate", "label": "Audio bitrate", "kind": "text",
         "value": config.AUDIO_BITRATE,
         "validate": _validate_bitrate},
        {"key": "language", "label": "Language", "kind": "text",
         "value": config.LANGUAGE, "validate": _validate_language},
        {"key": "chunk_size", "label": "Chunk size (words)", "kind": "text",
         "value": str(config.CHUNK_SIZE), "validate": _validate_chunk_size},
        {"key": "audiocpp_port", "label": "audio.cpp port",
         "kind": "text",
         "value": str(_port_from_url(config.AUDIOCPP_API_URL, 8080)),
         "validate": _validate_port,
         "note": "Ports apply to servers this tool starts and detecting "
                 "local servers"},
        {"key": "faster_port", "label": "faster-qwen3-tts port",
         "kind": "text",
         "value": str(_port_from_url(config.FASTER_API_URL, 8000)),
         "validate": _validate_port},
        {"key": "qwen_custom_port", "label": "qwen-tts CustomVoice port",
         "kind": "text",
         "value": str(_port_from_url(config.QWEN_API_URL, 7860)),
         "validate": _validate_port},
        {"key": "qwen_clone_port", "label": "qwen-tts Base port",
         "kind": "text",
         "value": str(_port_from_url(config.CLONE_API_URL, 7861)),
         "validate": _validate_port},
    ]
    result = tui.form(stdscr, "Settings", fields, back_value=_GO_BACK)
    if result is None or result is _GO_BACK:
        return
    try:
        _apply_settings(result)
    except ValueError as exc:
        tui.flash(stdscr, str(exc), "err")
        return
    tui.flash(stdscr, "Settings saved.", "ok")


def _validate_bitrate(value: str) -> Optional[str]:
    """Error message for a blank audio bitrate, or None to accept it."""
    if value.strip():
        return None
    return "Audio bitrate must not be empty"


def _validate_language(value: str) -> Optional[str]:
    """Error message for an unrecognized LANGUAGE, or None to accept it."""
    try:
        normalize_language(value)
        return None
    except ValueError as exc:
        return str(exc)


def _validate_chunk_size(value: str) -> Optional[str]:
    """Error message for an invalid CHUNK_SIZE, or None to accept it."""
    try:
        number = int(value.strip())
    except ValueError:
        return "Enter a whole number of words, e.g. 250"
    if number < 1:
        return "Chunk size must be at least 1"
    return None


def _validate_port(value: str) -> Optional[str]:
    """Error message for an invalid port, or None to accept it."""
    try:
        number = int(value.strip())
    except ValueError:
        return "Enter a port number, e.g. 8080"
    if not 1 <= number <= 65535:
        return "Port must be between 1 and 65535"
    return None


def _port_from_url(url: str, default: int) -> int:
    """Return the port in URL, or DEFAULT when it has none/unparsable."""
    try:
        return urllib.parse.urlsplit(url).port or default
    except ValueError:
        return default


def _apply_settings(values: dict) -> None:
    """Write VALUES to app/converter/config.py and reload them in-memory."""
    chunk_size = int(values["chunk_size"].strip())
    if chunk_size < 1:
        raise ValueError("Chunk size must be at least 1")
    bitrate = values["audio_bitrate"].strip()
    if not bitrate:
        raise ValueError("Audio bitrate must not be empty")
    if values["audio_format"] not in AUDIO_FORMATS:
        raise ValueError(f"Unsupported audio format: {values['audio_format']}")

    ports = {
        "qwen_custom_port": _read_port(values, "qwen_custom_port"),
        "qwen_clone_port": _read_port(values, "qwen_clone_port"),
        "faster_port": _read_port(values, "faster_port"),
        "audiocpp_port": _read_port(values, "audiocpp_port"),
    }
    updates = {
        "AUDIO_FORMAT": values["audio_format"],
        "AUDIO_BITRATE": bitrate,
        "LANGUAGE": normalize_language(values["language"]),
        "CHUNK_SIZE": chunk_size,
        "QWEN_API_URL": common.url_with_port(
            config.QWEN_API_URL, ports["qwen_custom_port"]),
        "CLONE_API_URL": common.url_with_port(
            config.CLONE_API_URL, ports["qwen_clone_port"]),
        "FASTER_API_URL": common.url_with_port(
            config.FASTER_API_URL, ports["faster_port"]),
        "AUDIOCPP_API_URL": common.url_with_port(
            config.AUDIOCPP_API_URL, ports["audiocpp_port"]),
    }
    _write_config(updates)
    for name, value in updates.items():
        setattr(config, name, value)

    _sync_audiocpp_server_port(ports["audiocpp_port"])


def _read_port(values: dict, key: str) -> int:
    """Parse a port field value, raising ValueError on a bad number."""
    try:
        number = int(values[key].strip())
    except (KeyError, ValueError):
        raise ValueError(f"Enter a valid port for {key}")
    if not 1 <= number <= 65535:
        raise ValueError("Port must be between 1 and 65535")
    return number


def _sync_audiocpp_server_port(port: int) -> None:
    """Rewrite the audio.cpp server.json 'port' to PORT when it exists.

    A missing checkout/server.json is a no-op (the config URL still
    changes; the file is regenerated on reconfigure). An existing
    server.json that cannot be updated raises, so the save is not
    reported as successful while the two are out of sync.
    """
    checkout = audiocpp_backend.find_local_checkout()
    if checkout is None:
        return
    server_json = checkout / "server.json"
    if not server_json.exists():
        return
    if not audiocpp_backend.update_server_config_port(port):
        raise ValueError(
            f"Could not update {server_json}; the audio.cpp port was "
            "left as-is")


def _write_config(updates: dict) -> None:
    """Rewrite the ``NAME = value`` lines for UPDATES in app/converter/config.py.

    Only the value of each named assignment changes: the indentation, the
    quotes (double, matching the file's style) and any trailing comment on
    the line are preserved. Every other line is left untouched.
    """
    path = Path(config.__file__).resolve()
    text = path.read_text(encoding="utf-8")
    for name, value in updates.items():
        rendered = str(value) if isinstance(value, int) else f'"{value}"'
        pattern = re.compile(
            rf"^(\s*{re.escape(name)}\s*=\s*)(\S*)(\s*(#.*))?$",
            re.MULTILINE)
        text, count = pattern.subn(
            lambda m, rendered=rendered:
            f"{m.group(1)}{rendered}{m.group(3) or ''}", text)
        if count != 1:
            raise ValueError(f"Could not find {name} in {path}")
    path.write_text(text, encoding="utf-8")


def _run_conversion(backend: str, kwargs: dict) -> None:
    """Run a conversion in the plain console (after the TUI returns).

    When the convert menu recorded an ``autostart`` server (the user opted to
    have the hub start it), spawn it now and abort the conversion if it does
    not come up. After the conversion, offer to stop a server we started.
    """
    autostart = kwargs.pop("autostart", None)
    status = next((s for s in detect_all() if s.key == backend), None)
    if status is not None and not status.ready and not status.running:
        print(f"[WARNING] {status.label} is not fully set up.")
    if autostart:
        spec = _find_spec(autostart)
        if spec is None:
            print(f"[WARNING] no server named '{autostart}'; continuing")
        elif not servers.start(spec):
            print("[ERROR] could not start the server; aborting conversion.")
            if status is not None and status.launch_hint:
                print("Start it manually and run the conversion again:")
                print(f"  {status.launch_hint}")
            return
    elif status is not None and not status.running and status.launch_hint:
        print("[INFO] Make sure the server is running. Start it with:")
        print(f"  {status.launch_hint}")
    try:
        audiobook.convert(backend=backend, **kwargs)
    finally:
        if autostart:
            _maybe_stop_server(autostart)


def _maybe_stop_server(name: str) -> None:
    """Ask (in the plain console) whether to stop a server we auto-started."""
    try:
        ans = input(f"\n[?] Stop the '{name}' server now? [y/N] ").strip().lower()
    except EOFError:
        return
    if ans in ("y", "yes"):
        servers.stop(name)


def _add_autostart(stdscr, cmd: tuple, statuses) -> None:
    """Offer to auto-start the conversion's target server when it isn't running.

    Records the chosen server spec name as ``kwargs['autostart']`` for
    ``_run_conversion`` to act on. Mode-aware for qwen (custom vs clone).
    """
    _, key, kwargs = cmd
    status = next((s for s in statuses if s.key == key), None)
    if status is None or not status.servers:
        return
    spec = _select_spec(status, kwargs)
    if spec is None:
        return
    if common.server_running(spec.url):
        return
    choice = tui.confirm(stdscr, f"The {status.label} server is not running. "
                         "Start it automatically?", default=True,
                         cancel_value=False)
    if choice is True:
        kwargs["autostart"] = spec.name


def _select_spec(status, kwargs) -> Optional[ServerSpec]:
    """The server spec this conversion needs (mode-aware for qwen)."""
    if status.key == BACKEND_QWEN:
        wanted = "qwen-clone" if kwargs.get("clone") else "qwen-custom"
        return next((s for s in status.servers if s.name == wanted), None)
    return status.servers[0] if status.servers else None


def _find_spec(name: str) -> Optional[ServerSpec]:
    """Look up a server spec by name across every backend's detect()."""
    for st in detect_all():
        for spec in st.servers:
            if spec.name == name:
                return spec
    return None


def _run_server_action(spec_name: str, action: str) -> None:
    """Run a Start/Stop action in the plain console (after the TUI returns)."""
    if action == "start":
        spec = _find_spec(spec_name)
        if spec is None:
            print(f"[ERROR] no server named '{spec_name}'")
            return
        servers.start(spec)
    elif action == "stop":
        servers.stop(spec_name)


def _server_menu(stdscr, statuses) -> Optional[tuple]:
    """Pick a backend, then one of its servers and a Start/Stop action."""
    # Only backends installed on this machine: a merely-running external
    # server cannot be stopped from here (stop() refuses without our pid
    # file), so listing it would dead-end.
    candidates = [st for st in statuses if st.installed]
    if not candidates:
        tui.flash(stdscr, "No backend is installed yet — use 'Set up a "
                  "backend' first.")
        return None
    options = [(st.label, st.key) for st in candidates]
    key = tui.menu(stdscr, "Start / Stop a server", options,
                   back_value=_GO_BACK,
                   table_title="Backend status",
                   table_rows=_status_rows(statuses),
                   notice_lines=_notice_lines())
    if key is _GO_BACK or key is None:
        return None
    status = next((s for s in statuses if s.key == key), None)
    if status is None:
        return None
    return _server_actions(stdscr, status)


def _server_actions(stdscr, status) -> Optional[tuple]:
    """Pick a server spec (qwen has two) and a Start or Stop action."""
    specs = status.servers
    if not specs:
        tui.flash(stdscr, f"{status.label} has no server configured. "
                  "Run 'Set up a backend' first.")
        return None
    if len(specs) == 1:
        spec = specs[0]
    else:
        options = [(f"{s.name}  ({'running' if common.server_running(s.url) else 'stopped'})",
                    s.name) for s in specs]
        name = tui.menu(stdscr, f"{status.label} server", options,
                        back_value=_GO_BACK)
        if name is _GO_BACK or name is None:
            return None
        spec = next((s for s in specs if s.name == name), None)
        if spec is None:
            return None
    running = common.server_running(spec.url)
    action = tui.menu(
        stdscr, f"{spec.name}  ({'running' if running else 'stopped'})",
        [("Start", "start"), ("Stop", "stop")], back_value=_GO_BACK)
    if action is _GO_BACK or action is None:
        return None
    return ("server", spec.name, action)


def _list_voices(voice_dir: str) -> list:
    """Return sorted .wav stems in VOICE_DIR (best-effort)."""
    try:
        path = Path(voice_dir)
        if not path.is_dir():
            return []
        return sorted(
            (p.stem for p in path.iterdir()
             if p.is_file() and p.suffix.lower() == ".wav"),
            key=str.lower,
        )
    except OSError:
        return []


def _is_float(value: str) -> bool:
    try:
        float(value)
        return True
    except ValueError:
        return False