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 Callable, 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 backends import qwen as qwen_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).

    A backend is 'running' (green/ok) when it is usable either locally — a
    server this tool started (``status.managed``) — or remotely — a server
    found by probing its remote URL (``status.remote``); the text names
    which, e.g. "running [local]", "running [remote]", or
    "running [local, remote]". 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 multi-model backend (qwen) also names which models
    answered in parentheses, e.g. "running [local, 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:
        tags = []
        if status.managed:
            tags.append("local")
        if status.remote:
            tags.append("remote")
        text = "running"
        if tags:
            text += " [" + ", ".join(tags) + "]"
        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]:
    """Collect run settings on one form: pick a backend, then its options.

    The first field is the Backend picker; the remaining fields are that
    backend's options (audio.cpp: model/voice/instructions; qwen:
    speaker or clone .wav; faster: voice), plus the shared output
    settings. A backend appears once as a managed entry ("audio.cpp") when
    it is installed+configured here, and once as a remote entry
    ("audio.cpp [remote]") when a running server was found at its remote
    URL. Managed entries read the local server.json / voices.json; remote
    entries query the remote server live. Each available entry's data is
    prepared up front so the Backend field lists only backends whose
    options could be gathered — an entry whose data is unavailable (e.g.
    an unreachable remote audio.cpp server) is dropped here.
    """
    entries = []
    for st in statuses:
        if st.ready:
            entries.append((st.key, st.label, st, False))
        if st.remote:
            entries.append((f"{st.key}-remote", f"{st.label} [remote]",
                            st, True))
    if not entries:
        tui.flash(stdscr, "No backend is ready to convert with yet — use "
                  "'Set up a backend' first.")
        return None
    builders = {}
    for key, _label, st, remote in entries:
        if remote:
            if st.key == BACKEND_AUDIOCPP:
                built = _audiocpp_fields(
                    stdscr, api_url=st.remote_urls.get("audiocpp"))
            elif st.key == BACKEND_QWEN:
                built = _qwen_fields(remote_modes=st.remote_models,
                                     urls=st.remote_urls)
            elif st.key == BACKEND_FASTER:
                built = _faster_fields(
                    stdscr, api_url=st.remote_urls.get("faster"))
            else:
                continue
        else:
            if st.key == BACKEND_AUDIOCPP:
                built = _audiocpp_fields(stdscr)
            elif st.key == BACKEND_QWEN:
                built = _qwen_fields()
            elif st.key == BACKEND_FASTER:
                built = _faster_fields(stdscr)
            else:
                continue
        if built is not None:
            builders[key] = built
    if not builders:
        return None
    choices = [(label, key) for key, label, _st, _remote in entries
               if key in builders]
    default = config.BACKEND if config.BACKEND in builders \
        else choices[0][1]
    fields = [{
        "key": "backend", "label": "Backend", "kind": "choice",
        "value": default, "choices": choices,
    }]
    for key, _label, _st, _remote in entries:
        if key not in builders:
            continue
        backend_fields, _ = builders[key]
        for field in backend_fields:
            field["visible"] = _gate_backend(field, key)
        fields += backend_fields
    fields += _common_fields()

    result = _show_convert_form(stdscr, "Convert books", fields)
    if result is None:
        return None
    _, mapper = builders[result["backend"]]
    cmd = mapper(result)
    if cmd is None:
        return None
    _add_autostart(cmd, statuses)
    return cmd


def _gate_backend(field: dict, key: str) -> Callable:
    """A visible() that shows FIELD only when the Backend field is KEY.

    Composes with any ``visible`` callable the field already carries
    (audio.cpp's task-driven Voice field, qwen's mode-driven fields), so
    both the backend gate and the field's own rule must pass.
    """
    base = field.get("visible", True)

    def visible(fields) -> bool:
        if _field_value(fields, "backend") != key:
            return False
        if callable(base):
            return base(fields)
        return bool(base)

    return visible


# Sentinel value the audio.cpp Voice field uses for "no --voice" (the
# built-in CustomVoice speaker); mapped to None when the form returns.
_AUDIOCPP_BUILTIN_SPEAKER = "(built-in speaker)"


def _field_value(fields, key: str, default=None):
    """Current value of the field named KEY, or DEFAULT when absent."""
    for field in fields:
        if field.get("key") == key:
            return field["value"]
    return default


def _common_fields() -> list:
    """Field dicts for output format, speed, single-file, and debug.

    The single-file field is hidden for m4b (always a single file with
    embedded chapter markers), so its "visible" callable reads the live
    output-format value from the field list.
    """
    fmt_default = config.AUDIO_FORMAT \
        if config.AUDIO_FORMAT in AUDIO_FORMATS else AUDIO_FORMATS[0]
    return [
        {"key": "output_format", "label": "Output format", "kind": "choice",
         "value": fmt_default, "choices": list(AUDIO_FORMATS)},
        {"key": "speed", "label": "Speed", "kind": "text", "value": "1.0",
         "validate": lambda s: None if (_is_float(s) and float(s) > 0)
         else "Enter a positive number, e.g. 1.0"},
        {"key": "single_file", "label": "Combine all chapters",
         "kind": "bool", "value": False,
         "visible": lambda fs: _field_value(fs, "output_format") != "m4b"},
        {"key": "debug", "label": "Debug", "kind": "bool", "value": False},
    ]


def _common_kwargs(values: dict) -> dict:
    """Map the common form fields to converter keyword arguments."""
    output_format = values["output_format"]
    return {
        "output_format": output_format,
        "speed": float(values["speed"]),
        "single_file": bool(values["single_file"])
        and output_format != "m4b",
        "debug": bool(values["debug"]),
    }


def _show_convert_form(stdscr, title: str, fields: list) -> Optional[dict]:
    """Show the single conversion form (Generate!/Cancel, focus on
    Generate!) and return its values, or None/_GO_BACK to go back."""
    result = tui.form(stdscr, title, fields,
                      buttons=("Generate!", "Cancel"),
                      start_on_buttons=True, back_value=_GO_BACK)
    if result is None or result is _GO_BACK:
        return None
    return result


def _audiocpp_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]:
    """audio.cpp-specific fields and a result mapper for the Convert form.

    Returns ``(fields, mapper)`` where FIELDS are the audio.cpp options
    (Model / Voice / Instructions) and MAPPER turns a submitted form
    values dict into the audio.cpp converter kwargs. Returns None when
    the model list cannot be gathered (a flash explains why), so the
    caller drops audio.cpp from the Backend choices.

    With API_URL None (the managed entry) the model list is fed from the
    local checkout's server.json — the config of the server this tool
    manages. With API_URL set (the "[remote]" entry) the models and voices
    are queried live from that server instead (the same GET /v1/models and
    GET /v1/audio/voices endpoints the converter resolves at run time).
    """
    if api_url is None:
        checkout = audiocpp_backend.find_local_checkout()
        server_json = checkout / "server.json" if checkout else None
        if not (server_json and server_json.exists()):
            tui.flash(stdscr, "No audio.cpp server.json found — run "
                      "'Set up a backend' first.")
            return None
        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
        local = True
        url = config.AUDIOCPP_API_URL
    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.
        url = api_url
        local = False
        data = {}
        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

    # Normalize each entry so the form logic sees a family/task always.
    models = [dict(m) for m in models]
    for entry in models:
        entry["family"] = entry.get("family") or ""
        entry["task"] = entry.get("task") or "tts"
        if not local and not entry["family"]:
            # Servers predating the family field omit it; mirror the
            # converter's default: unknown family means qwen3_tts.
            entry["family"] = AUDIOCPP_FAMILY_QWEN3_TTS

    local_voices = _list_voices(data.get("voice_dir")) \
        if data.get("voice_dir") else []
    voice_cache: dict = {}  # model id -> voices (local: shared list)

    def voices_for(model_id: str) -> list:
        if local:
            return local_voices
        if model_id not in voice_cache:
            fetched = audiocpp_backend.fetch_server_voices(url, model_id)
            voice_cache[model_id] = fetched or []
        return voice_cache[model_id]

    def model_entry(fields):
        model_id = _field_value(fields, "model_id")
        return next((m for m in models if m.get("id") == model_id),
                    models[0])

    def model_task(fields) -> str:
        return model_entry(fields).get("task", "tts")

    def model_family(fields) -> str:
        return model_entry(fields).get("family") or ""

    def reset_voice(fields) -> None:
        """Re-point the Voice field at the newly selected model's voice."""
        voice_field = next(f for f in fields
                           if f.get("key") == "audiocpp_voice")
        if model_task(fields) == "vdes":
            voice_field["value"] = None
        elif model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS:
            voice_field["value"] = _AUDIOCPP_BUILTIN_SPEAKER
        else:
            voices = voices_for(_field_value(fields, "model_id"))
            voice_field["value"] = voices[0] if voices else ""

    def voice_choices(fields) -> list:
        if model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS:
            return [(_AUDIOCPP_BUILTIN_SPEAKER, _AUDIOCPP_BUILTIN_SPEAKER)] \
                + [(v, v) for v in voices_for(_field_value(fields,
                                                           "model_id"))]
        return [(v, v) for v in voices_for(_field_value(fields, "model_id"))]

    model_ids = [m.get("id") for m in models]
    default_model = config.AUDIOCPP_MODEL_ID \
        if config.AUDIOCPP_MODEL_ID in model_ids else model_ids[0]
    default_entry = next((m for m in models if m.get("id") == default_model),
                         models[0])
    initial_voice = _AUDIOCPP_BUILTIN_SPEAKER
    if default_entry.get("task") == "vdes":
        initial_voice = None
    elif default_entry.get("family") != AUDIOCPP_FAMILY_QWEN3_TTS:
        initial = voices_for(default_model)
        initial_voice = initial[0] if initial else ""

    fields = [
        {"key": "model_id", "label": "Model", "kind": "choice",
         "value": default_model,
         "choices": [(f"{m.get('id')}  ({m.get('family') or '?'}, "
                      f"{m.get('task') or 'tts'})", m.get("id"))
                     for m in models],
         "on_change": reset_voice},
        {"key": "audiocpp_voice", "label": "Voice", "kind": "choice",
         "value": initial_voice,
         "choices": lambda fs: voice_choices(fs),
         "visible": lambda fs: model_task(fs) != "vdes",
         "validate": lambda value: None
         if (model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS or value)
         else "This model needs a voice — pick one or switch models"},
        {"key": "instructions", "label": "Instructions", "kind": "text",
         "value": config.AUDIOCPP_INSTRUCTIONS,
         "validate": lambda value: None
         if (model_task(fields) != "vdes" or str(value).strip())
         else "Describe the voice, e.g. 'A warm female narrator'"},
    ]

    def mapper(result) -> Optional[tuple]:
        model_id = result["model_id"]
        voice = result["audiocpp_voice"]
        if voice == _AUDIOCPP_BUILTIN_SPEAKER or not voice:
            voice = None
        entry = next((m for m in models if m.get("id") == model_id), {})
        if entry.get("task") == "vdes":
            voice = None
        instructions = (result["instructions"] or "").strip() or None
        kwargs = {
            "model_id": model_id, "voice": voice,
            "instructions": instructions,
            **_common_kwargs(result),
        }
        if api_url is not None:
            kwargs["api_url"] = api_url
        return ("convert", BACKEND_AUDIOCPP, kwargs)

    return fields, mapper


def _qwen_fields(remote_modes: Optional[list] = None,
                 urls: Optional[dict] = None) -> Optional[tuple]:
    """qwen-specific fields and a result mapper for the Convert form.

    Returns ``(fields, mapper)`` where FIELDS are the qwen options
    (Voice mode / Speaker / Clone .wav path) and MAPPER turns a
    submitted form values dict into the qwen converter kwargs. qwen
    always has options to offer, so it never signals unavailability.

    For the managed entry REMOTE_MODES/URLS are None and the mode picker
    offers both modes, targeting the configured local URLs. For a
    "[remote]" entry REMOTE_MODES names which demos answered remotely
    ("CustomVoice" and/or "Base") and URLS maps "qwen-custom"/"qwen-clone"
    to their URLs: the mode picker is limited to the available demos, and
    the mapper passes the matching remote URL as ``api_url``.
    """
    remote_modes = list(remote_modes or [])
    urls = dict(urls or {})
    mode_choices = []
    if urls.get("qwen-custom") or not remote_modes:
        mode_choices.append(("Built-in speaker", "custom"))
    if urls.get("qwen-clone") or not remote_modes:
        mode_choices.append(("Clone from a .wav file", "clone"))
    default_mode = mode_choices[0][1] if mode_choices else "custom"
    speakers = list(qwen_backend.QWEN_SPEAKERS)
    default_speaker = config.SPEAKER if config.SPEAKER in speakers \
        else speakers[0]
    fields = [
        {"key": "mode", "label": "Voice mode", "kind": "choice",
         "value": default_mode, "choices": mode_choices},
        {"key": "speaker", "label": "Speaker", "kind": "choice",
         "value": default_speaker, "choices": speakers,
         "visible": lambda fs: _field_value(fs, "mode") == "custom"},
        {"key": "clone", "label": "Clone .wav path", "kind": "text",
         "value": "",
         "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",
         "visible": lambda fs: _field_value(fs, "mode") == "clone"},
    ]

    def mapper(result) -> Optional[tuple]:
        clone = result["clone"].strip() if result["mode"] == "clone" else None
        speaker = result["speaker"]
        if result["mode"] == "custom" and speaker != config.SPEAKER:
            # Persist the speaker choice for this and future runs (mirrors
            # the qwen setup wizard), so the converter picks it up at
            # request time.
            common.update_config_value("SPEAKER", speaker)
            config.SPEAKER = speaker
        kwargs = {"clone": clone, **_common_kwargs(result)}
        if urls:
            api_url = urls.get("qwen-clone") if result["mode"] == "clone" \
                else urls.get("qwen-custom")
            if api_url:
                kwargs["api_url"] = api_url
        return ("convert", BACKEND_QWEN, kwargs)

    return fields, mapper


def _faster_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]:
    """faster-specific fields and a result mapper for the Convert form.

    Returns ``(fields, mapper)`` where FIELDS are the faster options
    (Voice, as a picker when a local voices.json lists them, else typed
    free text) and MAPPER turns a submitted form values dict into the
    faster converter kwargs. Returns None when a local voices.json
    exists but cannot be read/used (a flash explains why), so the caller
    drops faster from the Backend choices.

    With API_URL None (the managed entry) a local checkout's voices.json
    drives the picker. With API_URL set (the "[remote]" entry) 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.
    """
    voices = None
    if api_url is None:
        checkout = faster_backend._checkout()
        voices_json = checkout / "voices.json"
        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.
        fields = [
            {"key": "faster_voice", "label": "Voice", "kind": "text",
             "value": config.FASTER_VOICE,
             "validate": lambda s: None if s.strip() else "Enter a voice name"},
        ]
    else:
        default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \
            next(iter(voices))
        fields = [
            {"key": "faster_voice", "label": "Voice", "kind": "choice",
             "value": default, "choices": [(k, k) for k in voices]},
        ]

    def mapper(result) -> Optional[tuple]:
        voice = result["faster_voice"].strip() \
            if isinstance(result["faster_voice"], str) \
            else result["faster_voice"]
        kwargs = {
            "voice": voice or None,
            **_common_kwargs(result),
        }
        if api_url is not None:
            kwargs["api_url"] = api_url
        return ("convert", BACKEND_FASTER, kwargs)

    return fields, mapper


# ---------------------------------------------------------------------------
# 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 (local instances)"},
        {"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},
        {"key": "audiocpp_remote_url", "label": "audio.cpp remote URL",
         "kind": "text",
         "value": config.AUDIOCPP_REMOTE_URL,
         "validate": _validate_remote_url,
         "note": "Remote (externally-run) servers. The hub probes each URL and "
                 "offers a \"[remote]\" backend entry when one answers. "
                 "Empty disables probing."},
        {"key": "faster_remote_url", "label": "faster-qwen3-tts remote URL",
         "kind": "text",
         "value": config.FASTER_REMOTE_URL,
         "validate": _validate_remote_url},
        {"key": "qwen_custom_remote_url", "label": "qwen-tts CustomVoice remote URL",
         "kind": "text",
         "value": config.QWEN_REMOTE_URL,
         "validate": _validate_remote_url},
        {"key": "qwen_clone_remote_url", "label": "qwen-tts Base remote URL",
         "kind": "text",
         "value": config.CLONE_REMOTE_URL,
         "validate": _validate_remote_url},
    ]
    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 _validate_remote_url(value: str) -> Optional[str]:
    """Error message for an invalid remote URL, or None to accept it."""
    try:
        common.normalize_remote_url(value)
        return None
    except ValueError as exc:
        return str(exc)


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"),
    }
    remote_urls = {
        "QWEN_REMOTE_URL": common.normalize_remote_url(
            values.get("qwen_custom_remote_url", "")),
        "CLONE_REMOTE_URL": common.normalize_remote_url(
            values.get("qwen_clone_remote_url", "")),
        "FASTER_REMOTE_URL": common.normalize_remote_url(
            values.get("faster_remote_url", "")),
        "AUDIOCPP_REMOTE_URL": common.normalize_remote_url(
            values.get("audiocpp_remote_url", "")),
    }
    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"]),
        **remote_urls,
    }
    _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).

    A remote conversion (``api_url`` in the kwargs) targets an externally-run
    server, so no autostart is attempted and the managed instance's setup
    state is irrelevant. Otherwise, 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; a managed server
    whose port is already occupied by a server this tool did not start is
    left alone but warned about. After the conversion, offer to stop a
    server we started.
    """
    autostart = kwargs.pop("autostart", None)
    api_url = kwargs.get("api_url")
    if api_url:
        print(f"[INFO] Converting against remote server at {api_url}")
    else:
        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 status.servers:
            # The managed server's port may be held by a server we did not
            # start (its pid file is absent); the conversion would silently
            # talk to that server, so call it out.
            spec = _select_spec(status, kwargs)
            if spec is not None and common.server_running(spec.url) \
                    and not servers.alive(spec.name):
                print(f"[WARNING] A server this tool did not start is already "
                      f"running at {spec.url}; the conversion will talk to it. "
                      f"Stop it (or change the port) to use the managed "
                      f"{status.label} instance.")
            elif 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(cmd: tuple, statuses) -> None:
    """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. The user already accepted the run on the
    Generate! screen, so no start-server prompt is asked here — the server
    is simply started. Mode-aware for qwen (custom vs clone). Remote
    conversions (a ``api_url`` in the kwargs) never autostart: the server
    is external to this tool.
    """
    _, key, kwargs = cmd
    if kwargs.get("api_url"):
        return
    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
    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