path: root/app/ui/hub.py

#!/usr/bin/env python3
"""The TUI hub 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, or install/configure/remove a backend via the "Configure
backends" menu.

The entire hub runs in one curses session, driven by a single ``tui.Wizard``
stack of screens (the ``_Hub`` class below). Every menu/action is a screen
that returns the next screen, ``Wizard.BACK`` (Esc/q) to pop one screen, or
None to quit. Backend setup wizards and the conversion run view run as
opaque leaf screens on this same session (console tails under
``tui.suspend``); a leaf screen finishes by returning ``Wizard.BACK``, so
the stack lands back on the menu that launched it. Esc therefore steps back
exactly one screen everywhere — on the main menu (an empty stack) it quits.
'q' mirrors Esc on every screen that has no typed text.
"""

import contextlib
import functools
import io
import json
import re
import shutil
import urllib.parse
from datetime import datetime
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 probe as backend_probe
from backends import qwen as qwen_backend
from converter import config
from converter.converter import (
    AUDIO_FORMATS,
    AudiobookConverter,
    LOGS_FOLDER,
    voice_mode_for,
)
from converter.tts import (
    AUDIOCPP_FAMILY_QWEN3_TTS,
    BACKEND_AUDIOCPP,
    BACKEND_FASTER,
    BACKEND_QWEN,
    normalize_language,
)
from ui import runview, tui

_CANCEL = object()  # sentinel: a convert preflight confirm backed out


class _BackToForm(Exception):
    """Raised when Esc backs out of a preflight confirm (re-show the form)."""


def run() -> int:
    """Run the hub as one curses session; return the exit code."""
    import curses
    try:
        curses.wrapper(_app)
    except tui.WizardCancelled:
        return 0
    except KeyboardInterrupt:
        return 130
    return 0


def _app(stdscr) -> None:
    """Drive the whole hub as one ``tui.Wizard`` stack of screens."""
    _Hub(stdscr).run()


class _Hub:
    """The hub as a single ``tui.Wizard`` stack of screens.

    Every menu and action is a zero-argument bound method driven by
    ``tui.Wizard``: a screen returns the next screen (advance),
    ``Wizard.BACK`` (Esc/q — pop exactly one screen), or None (quit, only
    reached from the main menu). Backend setup wizards and the conversion
    run view run as opaque leaf screens on this same session; a leaf screen
    finishes by returning ``Wizard.BACK``, so the stack naturally lands back
    on the menu that launched it.
    """

    def __init__(self, stdscr):
        self.stdscr = stdscr

    def run(self) -> None:
        tui.Wizard().run(self.screen_main)

    # -- top level ------------------------------------------------------

    def screen_main(self):
        statuses = detect_all()
        options = [("Configure backends", "configure_backends")]
        # 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(("Start/Stop Backend Servers", "server"))
        options.append(("Settings", "settings"))
        options.append(("Quit", "quit"))
        choice = tui.menu(
            self.stdscr, "tts-audiobook-generator", options,
            back_value=tui.Wizard.BACK,
            table_title="Backend status", table_rows=_status_rows(statuses),
            notice_lines=_notice_lines())
        if choice is tui.Wizard.BACK or choice == "quit":
            return None
        if choice == "convert":
            return self.screen_convert
        if choice == "configure_backends":
            return self.screen_configure
        if choice == "server":
            return self.screen_server
        return self.screen_settings

    # -- configure / install / uninstall --------------------------------

    def screen_configure(self):
        """One flat menu of backend setup/configure/cleanup actions.

        Options are populated from the detected statuses: install (any
        uninstalled backend), configure each installed backend,
        download/delete audio.cpp models (when a server.json references
        models on/off disk), and uninstall. Selecting one pushes the next
        screen; Esc pops back to the main menu.
        """
        while True:
            statuses = detect_all()
            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]
            options = [(f"Configure {info.label}", ("configure", info.key))
                       for info in installed]
            if any(info.key not in by_key or not by_key[info.key].installed
                   for info in REGISTRY):
                options.append(("Install Backend", "install"))

            audiocpp_status = by_key.get("audiocpp")
            missing = []
            if audiocpp_status is not None and audiocpp_status.installed:
                checkout = audiocpp_backend.find_local_checkout()
                server_json = checkout / "server.json" if checkout else None
                if server_json is not None and server_json.exists():
                    missing = audiocpp_backend.missing_model_entries(
                        server_json)
            if missing:
                options.append(("Download Missing Models (audio.cpp)",
                                "download_models"))

            if installed:
                options.append(("Uninstall Backend", "uninstall"))

            choice = tui.menu(
                self.stdscr, "Configure backends", options,
                back_value=tui.Wizard.BACK,
                help_lines=["Install, configure, or remove a TTS backend."],
                table_title="Backend status",
                table_rows=_status_rows(statuses),
                notice_lines=_notice_lines())
            if choice is tui.Wizard.BACK:
                return tui.Wizard.BACK
            if choice == "install":
                return self.screen_install
            if choice == "uninstall":
                return self.screen_uninstall
            if choice == "download_models":
                _download_models_action(self.stdscr)
                continue  # an inline action: re-show this same menu
            _kind, key = choice
            info = get(key)
            if info is None:
                continue
            return self.screen_setup(info)

    def screen_setup(self, info):
        """Run one backend's setup wizard as a leaf screen of the stack.

        The wizard drives its own internal ``tui.Wizard`` on this screen;
        Esc on its first screen (or Ctrl-C) returns here and the hub pops
        back to the menu that launched it. A crash flashes and does the same.
        """
        def screen():
            try:
                info.setup_screen(self.stdscr)
            except tui.WizardCancelled:
                pass
            except Exception as exc:  # noqa: BLE001 - keep the hub alive
                tui.flash(self.stdscr, str(exc), "err")
            return tui.Wizard.BACK
        return screen

    def screen_install(self):
        info = self._pick_backend(installed_only=False)
        if info is None:
            return tui.Wizard.BACK
        return self.screen_setup(info)

    def screen_uninstall(self):
        info = self._pick_backend(installed_only=True)
        if info is None:
            return tui.Wizard.BACK
        with tui.suspend(self.stdscr):
            info.uninstall()
        return tui.Wizard.BACK

    def _pick_backend(self, installed_only: bool):
        """Pick a backend for the Install/Uninstall actions.

        With INSTALLED_ONLY False every backend is listed (the install
        list); with it True only the currently-installed ones are (the
        uninstall list). Returns a registry entry, or None to go back.
        """
        statuses = detect_all()
        by_key = {st.key: st for st in statuses}
        if installed_only:
            candidates = [info for info in REGISTRY
                          if by_key.get(info.key) is not None
                          and by_key[info.key].installed]
        else:
            candidates = [info for info in REGISTRY
                          if by_key.get(info.key) is None
                          or not by_key[info.key].installed]
        if not candidates:
            tui.flash(self.stdscr, "No backends to list here.")
            return None
        options = [(info.label, info.key) for info in candidates]
        title = "Uninstall Backend" if installed_only else "Install Backend"
        key = tui.menu(self.stdscr, title, options,
                       back_value=tui.Wizard.BACK,
                       table_title="Backend status",
                       table_rows=_status_rows(statuses),
                       notice_lines=_notice_lines())
        if key is tui.Wizard.BACK:
            return None
        return get(key)

    # -- convert --------------------------------------------------------

    def screen_convert(self):
        """Collect run settings, preflight, then run the conversion view.

        Esc on the form (or Cancel) pops back to the main menu; Esc on a
        preflight "overwrite?" confirm returns to the form (one screen).
        """
        prepared = _convert_form(self.stdscr)
        if prepared is None:
            return tui.Wizard.BACK
        fields, builders, statuses = prepared
        while True:
            result = tui.form(self.stdscr, "Convert books", fields,
                              buttons=("Generate!", "Cancel"),
                              start_on_buttons=True,
                              back_value=tui.Wizard.BACK)
            if result is tui.Wizard.BACK or result is None:
                return tui.Wizard.BACK
            _, mapper = builders[result["backend"]]
            cmd = mapper(result)
            if cmd is None:
                return tui.Wizard.BACK
            _add_autostart(cmd, statuses)
            try:
                ok = _preflight(self.stdscr, cmd)
            except _BackToForm:
                continue
            if not ok:
                return tui.Wizard.BACK
            self._run_conversion(cmd[1], cmd[2])
            return tui.Wizard.BACK

    def _run_conversion(self, backend: str, kwargs: dict) -> None:
        """Run a conversion in the full-screen run view on this session.

        A crash inside the view cancels the worker and flashes an error
        instead of taking the whole hub down; the timed getch the run view
        leaves behind is reset so the hub menus still block for keys.
        """
        run_config = _prepare_run_config(backend, kwargs)
        if run_config is None:
            return
        view = runview.RunView(self.stdscr, run_config)
        try:
            view.run()
        except tui.WizardCancelled:
            pass
        except KeyboardInterrupt:
            pass
        except Exception as exc:  # noqa: BLE001 - keep the hub alive
            view._cancel.set()
            view._worker.join(timeout=30)
            tui.flash(self.stdscr, f"The run view failed: {exc}", "err")
        finally:
            try:
                self.stdscr.timeout(-1)
            except Exception:
                pass

    # -- settings -------------------------------------------------------

    def screen_settings(self):
        result = tui.form(self.stdscr, "Settings", _settings_fields(),
                          back_value=tui.Wizard.BACK)
        if result is tui.Wizard.BACK or result is None:
            return tui.Wizard.BACK
        try:
            _apply_settings(result)
        except ValueError as exc:
            tui.flash(self.stdscr, str(exc), "err")
            return tui.Wizard.BACK
        tui.flash(self.stdscr, "Settings saved.", "ok")
        return tui.Wizard.BACK

    # -- servers --------------------------------------------------------

    def screen_server(self):
        statuses = detect_all()
        candidates = [st for st in statuses if st.installed]
        if not candidates:
            tui.flash(self.stdscr, "No backend is installed yet — use "
                      "'Configure backends' first.")
            return tui.Wizard.BACK
        options = [(st.label, st.key) for st in candidates]
        key = tui.menu(self.stdscr, "Start / Stop a server", options,
                       back_value=tui.Wizard.BACK,
                       table_title="Backend status",
                       table_rows=_status_rows(statuses),
                       notice_lines=_notice_lines())
        if key is tui.Wizard.BACK:
            return tui.Wizard.BACK
        status = next((s for s in statuses if s.key == key), None)
        if status is None:
            return tui.Wizard.BACK
        specs = status.servers
        if not specs:
            tui.flash(self.stdscr, f"{status.label} has no server "
                      "configured. Run 'Configure backends' first.")
            return tui.Wizard.BACK
        if len(specs) == 1:
            return functools.partial(self._server_action, specs[0])
        return functools.partial(self._server_spec, status, specs)

    def _server_spec(self, status, specs):
        options = [(f"{s.name}  ({'running' if common.server_running(s.url) else 'stopped'})",
                    s.name) for s in specs]
        name = tui.menu(self.stdscr, f"{status.label} server", options,
                        back_value=tui.Wizard.BACK)
        if name is tui.Wizard.BACK:
            return tui.Wizard.BACK
        spec = next((s for s in specs if s.name == name), None)
        if spec is None:
            return tui.Wizard.BACK
        return functools.partial(self._server_action, spec)

    def _server_action(self, spec):
        running = common.server_running(spec.url)
        action = tui.menu(
            self.stdscr,
            f"{spec.name}  ({'running' if running else 'stopped'})",
            [("Start", "start"), ("Stop", "stop")],
            back_value=tui.Wizard.BACK)
        if action is tui.Wizard.BACK:
            return tui.Wizard.BACK
        with tui.suspend(self.stdscr):
            if action == "start":
                servers.start(spec)
            else:
                servers.stop(spec.name)
        return tui.Wizard.BACK


def _download_models_action(stdscr) -> None:
    """Run the "Download Missing Models (audio.cpp)" action inside the TUI.

    Computes the missing models; when they map to install commands it
    suspends curses to stream the downloads, then flashes a result — instead
    of silently returning to the main menu. When the checkout/server.json is
    missing, nothing is missing, or the models do not map to an install
    command, it flashes an explanatory notice (the latter explaining how to
    install each model by hand).
    """
    checkout = audiocpp_backend.find_local_checkout()
    if checkout is None:
        tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp "
                  "first.", "err")
        return
    server_json = checkout / "server.json"
    if not server_json.exists():
        tui.flash(stdscr, "No audio.cpp server.json found — configure "
                  "audio.cpp first.", "err")
        return
    missing = audiocpp_backend.missing_model_entries(server_json)
    if not missing:
        tui.flash(stdscr, "Every configured audio.cpp model is already "
                  "downloaded.", "ok")
        return
    guidance = audiocpp_backend.missing_model_install_guidance(
        checkout, missing)
    if not guidance:
        tui.flash(stdscr, audiocpp_backend.hand_install_guidance(
            checkout, missing), "err")
        return
    with tui.suspend(stdscr):
        audiocpp_backend.install_models(checkout, guidance)
    tui.flash(stdscr, "Model download finished. See the output above for "
              "any warnings.", "ok")


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:
        if status.models_missing and not status.running:
            return ("installed (models missing)", "warn", "body")
        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_form(stdscr) -> Optional[tuple]:
    """Build the Convert-books form (fields + builders), or None to go back.

    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.

    Returns ``(fields, builders, statuses)``; None (after a flash) when
    there is nothing to convert with.
    """
    statuses = detect_all()
    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 "
                  "'Configure backends' 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()
    return fields, builders, statuses


def _preflight(stdscr, cmd: tuple) -> bool:
    """Run the overwrite checks in the TUI; stash the plan on the command.

    Asks every "output exists — overwrite?" question now (tui.confirm
    instead of the console input()) so the run view itself is unattended,
    and records the discovered books / accepted plan in the command's
    kwargs (``book_files``/``planned``) for ``audiobook.convert``. Returns
    False when nothing would be converted (a flash explains why), so the
    user stays in the menu instead of entering an empty run.
    """
    _kind, backend, kwargs = cmd
    voice_mode = voice_mode_for(backend, kwargs.get("voice"),
                                kwargs.get("clone"))

    def confirm(message: str, default: bool) -> bool:
        answer = tui.confirm(stdscr, message, default=default,
                             cancel_value=_CANCEL)
        if answer is _CANCEL:
            raise _BackToForm()
        return answer

    with contextlib.redirect_stdout(io.StringIO()):
        book_files, planned = AudiobookConverter.preflight_overwrites(
            backend=backend, voice=kwargs.get("voice"),
            voice_mode=voice_mode,
            voice_clone_ref_audio=kwargs.get("clone"),
            output_format=kwargs.get("output_format") or config.AUDIO_FORMAT,
            instructions=kwargs.get("instructions"),
            confirm=confirm)
    if not book_files:
        tui.flash(stdscr, "No books to convert. Add a .txt, .pdf or .epub "
                  "file to the input folder first.")
        return False
    if not planned:
        tui.flash(stdscr, "Nothing to convert — every existing output was "
                  "kept.")
        return False
    kwargs["book_files"] = book_files
    kwargs["planned"] = planned
    return True


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 _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 "
                      "'Configure backends' 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

    if local:
        # Only offer entries whose model files are actually on disk: a
        # server.json can reference a package that was never downloaded,
        # and picking it would fail the whole run at model-load time.
        missing = audiocpp_backend.missing_model_entries(server_json)
        if missing:
            missing_ids = {item["id"] for item in missing}
            models = [entry for entry in models
                      if entry.get("id") not in missing_ids]
            if not models:
                hints = audiocpp_backend.model_install_hints(checkout,
                                                             missing)
                message = hints[0] if hints \
                    else "Download the models first."
                tui.flash(stdscr, "No model files are downloaded for "
                          f"audio.cpp. {message}")
                return None

    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_fields() -> list:
    """The global output-settings field list (Save writes to config.py)."""
    return [
        {"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": "unload_models", "label": "Unload models", "kind": "bool",
         "value": config.AUDIOCPP_UNLOAD_MODELS,
         "note": "audio.cpp only. Ask the server to unload resident models "
                 "before converting."},
        {"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},
    ]


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,
        "AUDIOCPP_UNLOAD_MODELS": bool(values["unload_models"]),
        "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 _prepare_run_config(backend: str, kwargs: dict
                        ) -> Optional[runview.RunConfig]:
    """Build the run view's config from the accepted conversion kwargs.

    A remote conversion (``api_url``) 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 server was not running), the run view boots it first; a
    managed server whose port is already occupied by a server this tool
    did not start is left alone but flagged with a notice. Returns None
    when the backend disappeared between the menu and the dispatch.
    """
    label = backend
    info = get(backend)
    if info is not None:
        label = info.label
    log_path = str(LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log")
    autostart = kwargs.pop("autostart", None)
    api_url = kwargs.get("api_url")

    if api_url:
        identity = _remote_identity(backend, kwargs)
        return runview.RunConfig(
            backend=backend, backend_label=f"{label} [remote]",
            kwargs=kwargs, book_files=kwargs.get("book_files") or [],
            planned=kwargs.get("planned") or [],
            server_url=api_url, server_identity=identity,
            log_path=log_path)

    status = next((s for s in detect_all() if s.key == backend), None)
    notice = ""
    spec: Optional[ServerSpec] = None
    if autostart:
        spec = _find_spec(autostart)
    elif status is not None:
        spec = _select_spec(status, kwargs)
        if spec is not None and common.server_running(spec.url) \
                and not servers.alive(spec.name):
            notice = (f"a server this tool did not start is running at "
                      f"{spec.url} — the conversion will talk to it")
    if autostart and spec is None:
        # The recorded server vanished (backend reconfigured meanwhile):
        # converting without it is still meaningful, so continue.
        notice = (f"no server named '{autostart}' — starting it was skipped")
    return runview.RunConfig(
        backend=backend, backend_label=label, kwargs=kwargs,
        book_files=kwargs.get("book_files") or [],
        planned=kwargs.get("planned") or [],
        server_name=spec.name if spec is not None else None,
        server_url=spec.url if spec is not None else None,
        server_identity=spec.identity if spec is not None else None,
        autostart_spec=spec if autostart else None,
        log_path=log_path, notice=notice)


def _remote_identity(backend: str, kwargs: dict) -> Optional[str]:
    """The probe identity of the remote server a conversion targets."""
    if backend == BACKEND_AUDIOCPP:
        return backend_probe.IDENTITY_AUDIOCPP
    if backend == BACKEND_QWEN:
        return backend_probe.IDENTITY_QWEN_CLONE if kwargs.get("clone") \
            else backend_probe.IDENTITY_QWEN_CUSTOM
    if backend == BACKEND_FASTER:
        return backend_probe.IDENTITY_FASTER
    return None


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
    ``_prepare_run_config`` 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 (an ``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 _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