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; every long step — setup tails,
model downloads, server start/stop, and uninstall — runs inside the
``ui.taskview`` task view, so the user is never dropped to the console.
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 sys
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_VOICE_CLONE,
    AUDIOCPP_VOICE_DESIGN,
    AUDIOCPP_VOICE_SPEAKER,
    BACKEND_AUDIOCPP,
    BACKEND_FASTER,
    BACKEND_QWEN,
    QWEN3_TTS_SPEAKERS,
    audiocpp_entry_voice_capability,
    normalize_language,
)
from ui import runview, taskview, 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
    finally:
        # The curses session is over and the terminal is restored: surface
        # anything setup steps queued for the console (e.g. a failed audio.cpp
        # build's copy-pastable command and build log path).
        for notice in common.drain_post_tui_notices():
            print(notice)
    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.

        The audio.cpp "next step" — build its server (when a checkout has
        no binary) or download its missing models (only once built, so
        build > configure > download — Build and Download never appear
        together) — heads the menu with a yellow ``[recommended]`` tag,
        separated from the rest by a blank line. The remaining options are
        populated from the detected statuses: configure each installed
        backend, install (backends with nothing on disk), and uninstall.
        Selecting one pushes the next screen; Esc pops back to the main
        menu. The Build action downloads any missing models alongside the
        build (a split view), so it heals a configured-but-unbuilt backend
        in one step; Download Missing Models stays as the fallback for when
        a download fails or is interrupted.
        """
        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]

            audiocpp_status = by_key.get("audiocpp")
            missing = []
            needs_build = False
            if audiocpp_status is not None:
                checkout = audiocpp_backend.find_local_checkout()
                if checkout is not None:
                    built = audiocpp_backend.find_audiocpp_server_bin(
                        checkout) is not None
                    if not built:
                        needs_build = True
                    server_json = checkout / "server.json"
                    # Models can only be downloaded once the server binary
                    # exists (build > configure > download), so Build and
                    # Download never appear together.
                    if built and audiocpp_status.configured \
                            and server_json.exists():
                        missing = audiocpp_backend.missing_model_entries(
                            server_json)

            options = []
            if needs_build:
                options.append(("Build audio.cpp server", "build_audiocpp",
                                ("[recommended]", "warn")))
            elif missing:
                options.append(("Download Missing Models (audio.cpp)",
                                "download_models",
                                ("[recommended]", "warn")))
            if needs_build or missing:
                options.append(tui.MENU_SEPARATOR)

            options += [(f"Configure {info.label}", ("configure", info.key))
                        for info in installed]
            if any(_installable(info, by_key) for info in REGISTRY):
                options.append(("Install Backend", "install"))
            if any(_uninstallable(info, by_key) for info in REGISTRY):
                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
            if choice == "build_audiocpp":
                audiocpp_backend.build_screen(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():
            self._run_setup(info)
            return tui.Wizard.BACK
        return screen

    def _run_setup(self, info) -> None:
        """Run one backend's setup wizard on this session (no stack frame)."""
        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")

    def screen_install(self):
        """Pick a backend to install and run its setup inline.

        The setup is not pushed as a stack frame: when it finishes this
        screen returns BACK, popping straight past the picker to the
        Configure menu, whose status table re-detects the new install.
        Esc on the picker still pops back one screen normally.
        """
        info = self._pick_backend(installed_only=False)
        if info is None:
            return tui.Wizard.BACK
        self._run_setup(info)
        return tui.Wizard.BACK

    def screen_uninstall(self):
        """Pick a backend, confirm, then uninstall it inside the task view.

        Esc on the picker or the confirm (or answering No) backs out
        untouched. A confirmed uninstall runs as one task-view step on this
        session — no console drop: the backend's ``uninstall`` stops its
        servers, pip-uninstalls, and deletes its files, honoring cancel
        between phases only. A flash summarizes the result and the stack
        lands back on the Configure menu, whose status table re-detects the
        removal.
        """
        info = self._pick_backend(installed_only=True)
        if info is None:
            return tui.Wizard.BACK
        answer = tui.confirm(
            self.stdscr, f"Uninstall {info.label}?",
            body=[f"This permanently removes {info.label} from this "
                  "machine: managed servers are stopped and every installed "
                  "file — including downloaded models — is deleted."],
            default=False, cancel_value=tui.Wizard.BACK)
        if answer is not True:
            return tui.Wizard.BACK
        title = f"Uninstall {info.label}"
        step = taskview.TaskStep(
            title,
            lambda emit, cancel: info.uninstall(emit=emit, cancel=cancel))
        rc = taskview.run_steps(self.stdscr, title, [step],
                                wait_on_finish=False)
        # The uninstallers warn-and-continue (a failed pip step still
        # returns 0), so rc == 0 means "finished"; anything else covers a
        # failure or an Esc-cancelled run between phases.
        if rc == 0:
            tui.flash(self.stdscr, f"{info.label} uninstalled.", "ok")
        else:
            tui.flash(self.stdscr,
                      f"Could not fully uninstall {info.label}.", "err")
        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 with nothing on disk yet is
        listed (the install list — audio.cpp only without a checkout, since
        a downloaded-but-unbuilt checkout is past install); with it True the
        ones with something on disk to remove are (the uninstall list —
        including a downloaded-but-unbuilt audio.cpp checkout, which
        ``uninstall`` deletes whole). 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 _uninstallable(info, by_key)]
        else:
            candidates = [info for info in REGISTRY
                          if _installable(info, by_key)]
        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
            if self._run_conversion(cmd[1], cmd[2]):
                # "Stop server and exit after generating" was on: returning
                # None ends the wizard stack (the whole TUI).
                return None
            return tui.Wizard.BACK

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

        Returns True when the run view's stop-and-exit toggle was on — the
        caller then quits the whole TUI (results are printed after curses
        closes) instead of landing back on the menu. 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 False
        view = runview.RunView(self.stdscr, run_config)
        try:
            return bool(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
        return False

    # -- 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):
        """One flat menu of every installed backend's servers.

        A status table above the menu shows each server's live state —
        "running" (green) or "stopped" (red) — and selecting an entry
        toggles it directly (starts a stopped server, stops a running one)
        without an extra action menu. The state lives in the table, not on
        the entries, because the menu's selection bar would cover inline
        colors. A backend with a single server is labelled by its name; a
        multi-server backend (qwen: CustomVoice + Base) gets one entry per
        server, suffixed with the server name.
        """
        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 = []
        rows = []
        for st in candidates:
            specs = st.servers
            for spec in specs:
                running = common.server_running(spec.url)
                label = st.label if len(specs) == 1 \
                    else f"{st.label} — {spec.name}"
                options.append((label, spec))
                rows.append((label,
                             "running" if running else "stopped",
                             "ok" if running else "err", "body"))
        if not options:
            tui.flash(self.stdscr, "No backend server is configured yet — "
                      "use 'Configure backends' first.")
            return tui.Wizard.BACK
        spec = tui.menu(self.stdscr, "Start / Stop a server", options,
                        back_value=tui.Wizard.BACK,
                        help_lines=["Select a server to start or stop it."],
                        table_title="Server status",
                        table_rows=rows,
                        notice_lines=_notice_lines())
        if spec is tui.Wizard.BACK:
            return tui.Wizard.BACK
        return functools.partial(self._server_toggle, spec)

    def _server_toggle(self, spec):
        """Start SPEC's server when stopped, stop it when running.

        Runs inside the task view (no console drop); the server module's
        plain-console output is tee'd to a log file under ``app/logs`` so
        nothing is lost, and on failure a flash points the user at that
        file. Returns BACK so the stack lands back on the server list,
        which re-reads each server's live state.
        """
        running = common.server_running(spec.url)
        action = "stop" if running else "start"
        step, log_path = _server_action_step(spec, action)
        taskview.run_steps(self.stdscr, f"{action.capitalize()} "
                                       f"{spec.name} server", [step],
                           wait_on_finish=False)
        # Re-check the server instead of trusting the step's exit code
        # (cancel and failure both come back non-zero): did the toggle take?
        now_running = common.server_running(spec.url)
        if action == "start" and not now_running:
            tui.flash(self.stdscr, f"Could not start the {spec.name} "
                      f"server. See the log: {log_path}", "err")
        elif action == "stop" and now_running:
            tui.flash(self.stdscr, f"Could not stop the {spec.name} "
                      f"server. See the log: {log_path}", "err")
        return tui.Wizard.BACK


class _TeeWriter:
    """A file-like that mirrors writes to a log file and an inner stream.

    Used to capture the server module's plain-console output (the task view
    already redirects stdout to its line-writer) into a persistent log file
    under ``servers.LOG_DIR`` without losing the on-screen log tail.
    """

    def __init__(self, logf, inner):
        self._logf = logf
        self._inner = inner

    def write(self, text):
        if not text:
            return 0
        try:
            self._logf.write(text)
        except OSError:
            pass
        try:
            self._inner.write(text)
        except OSError:
            pass
        return len(text)

    def flush(self):
        try:
            self._logf.flush()
        except OSError:
            pass
        try:
            self._inner.flush()
        except OSError:
            pass


def _server_action_step(spec, action: str):
    """Build a task step that starts/stops SPEC's server, logged to a file.

    ACTION is "start" or "stop". The step runs inside the task view (no
    console drop): the server module's output is tee'd to
    ``<servers.LOG_DIR>/<name>-<action>.log`` and to the view's log tail.
    Returns ``(TaskStep, log_path)`` so the caller can point the user at the
    file on failure.
    """
    log_path = servers.LOG_DIR / f"{spec.name}-{action}.log"
    title = (f"Start {spec.name} server" if action == "start"
             else f"Stop {spec.name} server")

    def work(emit, cancel):
        servers.LOG_DIR.mkdir(parents=True, exist_ok=True)
        inner = sys.stdout  # the task view's line-writer, when run in TUI
        with log_path.open("w", encoding="utf-8") as logf, \
                contextlib.redirect_stdout(_TeeWriter(logf, inner)):
            if action == "start":
                ok = servers.start(spec, cancel=cancel)
            else:
                ok = servers.stop(spec.name)
        return 0 if ok else 1

    return taskview.TaskStep(title, work), log_path


def _installable(info, by_key: dict) -> bool:
    """True when INFO has nothing on disk yet — an install-entry candidate.

    audio.cpp is installable only without a checkout: a downloaded-but-unbuilt
    checkout is already past the install step (its next action is the hub's
    Build entry), so listing it under "Install Backend" would duplicate that
    and suggest re-running setup from scratch. The other backends are
    installable while not installed.
    """
    if info.key == "audiocpp":
        return audiocpp_backend.find_local_checkout() is None
    status = by_key.get(info.key)
    return status is None or not status.installed


def _uninstallable(info, by_key: dict) -> bool:
    """True when INFO has something on disk that uninstall removes.

    audio.cpp's ``installed`` flag means *built*, so a downloaded-but-unbuilt
    checkout would otherwise miss the Uninstall menu — but its checkout
    (binary, models, server.json) lives on disk and ``uninstall()`` removes
    it, so it counts too. The other backends' ``installed`` already covers
    everything their uninstaller touches.
    """
    if info.key == "audiocpp":
        return audiocpp_backend.find_local_checkout() is not None
    status = by_key.get(info.key)
    return status is not None and status.installed


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 runs
    the downloads in the task view (with real byte progress and cancellation)
    and flashes a result — instead of dropping to the console. 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

    def run(emit, cancel):
        audiocpp_backend.install_models(checkout, guidance,
                                         emit=emit, cancel=cancel)
        return 0

    taskview.run_steps(stdscr, "Download models",
                       [taskview.TaskStep("Download missing models", run)])
    tui.flash(stdscr, "Model download finished. Any warnings were shown in "
              "the log.", "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 a backend set up only part-way
    (``status.partial``) shows that label verbatim (amber), e.g. audio.cpp's
    "downloaded (not built)" or "built (not configured)"; '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.partial:
        # Part-way set up (audio.cpp: "downloaded (not built)" /
        # "built (not configured)"): amber text, name dimmed while the
        # backend is still unusable.
        name_kind = "dim" if not status.installed else "body"
        return (status.partial, "warn", name_kind)
    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


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},
        {"key": "stop_and_exit", "label": "Stop server and exit after "
         "generating", "kind": "bool", "value": True},
    ]


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"]),
        "stop_and_exit": bool(values["stop_and_exit"]),
    }


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.
    # A missing family is left empty (an unknown family resolves to the
    # clone capability, requiring a --voice) rather than guessing a specific
    # one — audiocpp_server always reports family for entries it hosts.
    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 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_capability(fields) -> str:
        entry = model_entry(fields)
        return audiocpp_entry_voice_capability(
            entry.get("family") or "", entry.get("task") or "tts",
            entry.get("id") 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")
        capability = model_capability(fields)
        if capability == AUDIOCPP_VOICE_DESIGN:
            voice_field["value"] = None
        elif capability == AUDIOCPP_VOICE_SPEAKER:
            voice_field["value"] = (config.SPEAKER
                                    if config.SPEAKER in QWEN3_TTS_SPEAKERS
                                    else QWEN3_TTS_SPEAKERS[0])
        else:  # clone
            voices = voices_for(_field_value(fields, "model_id"))
            voice_field["value"] = voices[0] if voices else ""

    def voice_choices(fields) -> list:
        capability = model_capability(fields)
        if capability == AUDIOCPP_VOICE_SPEAKER:
            # Built-in Qwen3-TTS CustomVoice speakers; no server query needed.
            return [(s, s) for s in QWEN3_TTS_SPEAKERS]
        if capability == AUDIOCPP_VOICE_CLONE:
            return [(v, v) for v in voices_for(_field_value(fields,
                                                           "model_id"))]
        return []  # design: the field is hidden

    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])
    default_capability = audiocpp_entry_voice_capability(
        default_entry.get("family") or "", default_entry.get("task") or "tts",
        default_entry.get("id") or "")
    initial_voice = None
    if default_capability == AUDIOCPP_VOICE_SPEAKER:
        initial_voice = (config.SPEAKER if config.SPEAKER in QWEN3_TTS_SPEAKERS
                         else QWEN3_TTS_SPEAKERS[0])
    elif default_capability == AUDIOCPP_VOICE_CLONE:
        initial = voices_for(default_model)
        initial_voice = initial[0] if initial else ""

    def _label(entry: dict) -> str:
        capability = audiocpp_entry_voice_capability(
            entry.get("family") or "", entry.get("task") or "tts",
            entry.get("id") or "")
        return (f"{entry.get('id')}  ({entry.get('family') or '?'}, "
                f"{capability})")

    fields = [
        {"key": "model_id", "label": "Model", "kind": "choice",
         "value": default_model,
         "choices": [(_label(m), 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_capability(fs) != AUDIOCPP_VOICE_DESIGN,
         "validate": lambda value: None
         if (model_capability(fields) != AUDIOCPP_VOICE_CLONE or value)
         else "This model needs a voice — pick one or switch models"},
        {"key": "instructions", "label": "Instructions", "kind": "text",
         "value": config.AUDIOCPP_INSTRUCTIONS,
         "visible": lambda fs: model_capability(fs) in (AUDIOCPP_VOICE_DESIGN,
                                                        AUDIOCPP_VOICE_SPEAKER),
         "validate": lambda value: None
         if (model_capability(fields) != AUDIOCPP_VOICE_DESIGN or str(value).strip())
         else "Describe the voice, e.g. 'A warm female narrator'"},
    ]

    def mapper(result) -> Optional[tuple]:
        model_id = result["model_id"]
        entry = next((m for m in models if m.get("id") == model_id), {})
        capability = audiocpp_entry_voice_capability(
            entry.get("family") or "", entry.get("task") or "tts",
            entry.get("id") or "")
        # The picked voice (a built-in speaker name on a CustomVoice entry,
        # a server-side preset otherwise); the client resolves which it is.
        voice = result["audiocpp_voice"] or None
        # design: the voice comes from --instructions
        instructions = None
        if capability in (AUDIOCPP_VOICE_DESIGN, AUDIOCPP_VOICE_SPEAKER):
            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: Unload previously-loaded models before converting to prevent VRAM exhaustion."},
        {"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. 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")
    # The run view points failures at this file, so make sure it exists
    # from the moment a run starts — even when the run dies before the
    # converter's setup_logging creates it.
    LOGS_FOLDER.mkdir(parents=True, exist_ok=True)
    Path(log_path).touch()
    autostart = kwargs.pop("autostart", None)
    # The run-view behavior toggle (not a converter kwarg): stop the server
    # and quit the TUI once the generation ends.
    stop_and_exit = bool(kwargs.pop("stop_and_exit", False))
    # book_files/planned travel on the dedicated RunConfig fields; keeping
    # them in kwargs too would collide with convert()'s named parameters.
    book_files = kwargs.pop("book_files", None) or []
    planned = kwargs.pop("planned", None) or []
    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=book_files,
            planned=planned,
            server_url=api_url, server_identity=identity,
            log_path=log_path, stop_and_exit=stop_and_exit)

    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=book_files, planned=planned,
        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, stop_and_exit=stop_and_exit)


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