path: root/tools/make_audiocpp_server_json.py

#!/usr/bin/env python3
"""Interactively generate a server.json for the audio.cpp audiocpp_server.

Reads the model catalog (``model_specs/*.json``) from a local audio.cpp
checkout and offers every TTS model family audio.cpp supports as a
multi-select checklist, so one server.json can host several lazily-loaded
model entries at once. The converter itself is family-agnostic (it detects
the family of the selected entry from ``GET /v1/models`` at startup), so any
TTS family listed in the catalog works without further changes.

Cloning reference .wav files (the required WAV_DIR argument) are transcribed
with a local Whisper backend (faster_whisper or whisper) and published as a
server-level ``voice_dir`` plus a ``prompt_text`` mapping file written into
WAV_DIR, so every hosted clone-capable family can use them with ``--voice``.

Every value can also be supplied as a command-line flag; anything missing is
asked interactively with the default shown in brackets. Pressing Enter accepts
the default (the Qwen3-TTS built-in-speakers + voice-cloning flow).

Usage:
    python tools/make_audiocpp_server_json.py WAV_DIR [--output PATH]
        [--audiocpp-dir PATH] [--families FAM1,FAM2]
        [--models {both,custom,clone}] [--host HOST] [--port PORT]
        [--backend {cuda,vulkan,hip,cpu}] [--lazy-load]
        [--whisper-model NAME] [--force]

WAV_DIR is required: a directory of .wav reference files used as voice
cloning presets. It is checked up front and reported with its resolved
absolute path if it does not exist.

--audiocpp-dir defaults to a detected audio.cpp checkout (the AUDIOCPP_DIR
environment variable, or an ``audio.cpp`` directory next to or above the
current working directory); if none is found it is asked interactively. The
checkout must contain a ``model_specs/`` directory.
"""

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

# Allow running from any working directory.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from converter import config
from converter.tts import transcribe_reference_audio, whisper_backend_available

DEFAULT_HOST = "127.0.0.1"
FALLBACK_PORT = 8080
DEFAULT_CUSTOM_VOICE_PATH = "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"
DEFAULT_BASE_PATH = "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF"
CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py"

MODEL_SELECTIONS = ("both", "custom", "clone")
BACKENDS = ("cuda", "vulkan", "hip", "cpu")

FAMILY_QWEN3_TTS = "qwen3_tts"
PROMPT_TEXT_FILENAME = "prompt_text"

# Families explicitly tested with this converter, in display order. These are
# listed first in the checklist and marked "[tested]"; every other TTS family
# in the catalog is offered too through the converter's generic profile.
TESTED_FAMILIES = (
    "qwen3_tts",
    "higgs_audio_tts",
    "voxcpm2",
    "index_tts2",
)

# Short, friendly default entry ids for tested families. Other families derive
# an id from their family name (see default_model_id).
PREFERRED_IDS = {
    "qwen3_tts": "qwen",
    "higgs_audio_tts": "higgs",
    "voxcpm2": "voxcpm2",
    "index_tts2": "indextts2",
}


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

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


def find_wav_files(input_dir: Path) -> list:
    """Return the .wav files in INPUT_DIR, sorted alphabetically by name."""
    return sorted(
        (path for path in input_dir.iterdir()
         if path.is_file() and path.suffix.lower() == ".wav"),
        key=lambda path: path.name.lower(),
    )


def prompt_overwrite(output_path: Path) -> bool:
    """Ask whether to overwrite an existing output file."""
    while True:
        try:
            answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower()
        except EOFError:
            print("\n[WARNING] No interactive input available; keeping existing file")
            return False
        if answer in ("y", "yes"):
            return True
        if answer in ("n", "no"):
            return False
        print("Please answer 'y' or 'n'.")


def ask(prompt: str, default: Optional[str] = None) -> Optional[str]:
    """Prompt for a free-text value with a default; EOF returns the default."""
    suffix = f" [{default}]" if default is not None else ""
    try:
        answer = input(f"{prompt}{suffix}: ").strip()
    except EOFError:
        return default
    return answer or default


def ask_bool(prompt: str, default: bool = False) -> bool:
    """Prompt for a yes/no answer; Enter or EOF accepts the default."""
    suffix = " [Y/n]" if default else " [y/N]"
    while True:
        try:
            answer = input(f"{prompt}{suffix}: ").strip().lower()
        except EOFError:
            return default
        if not answer:
            return default
        if answer in ("y", "yes"):
            return True
        if answer in ("n", "no"):
            return False
        print("Please answer 'y' or 'n'.")


def ask_port(default: int) -> int:
    """Prompt for a port number; Enter or EOF accepts the default."""
    while True:
        try:
            answer = input(f"Port [{default}]: ").strip()
        except EOFError:
            return default
        if not answer:
            return default
        try:
            value = int(answer)
        except ValueError:
            value = None
        if value is not None and 1 <= value <= 65535:
            return value
        print("Please enter a port number between 1 and 65535.")


def ask_menu(title: str, options: list, default_index: int = 1) -> str:
    """Show a numbered menu and return the chosen option's value."""
    print(title)
    for number, (label, _) in enumerate(options, 1):
        print(f"  {number}) {label}")
    while True:
        try:
            answer = input(f"Choice [{default_index}]: ").strip()
        except EOFError:
            return options[default_index - 1][1]
        if not answer:
            return options[default_index - 1][1]
        if answer.isdigit() and 1 <= int(answer) <= len(options):
            return options[int(answer) - 1][1]
        print(f"Please enter a number between 1 and {len(options)}.")


def ask_models() -> str:
    return ask_menu(
        "Which Qwen3-TTS models should the server host?",
        [
            ("Both (recommended) - built-in speakers + voice cloning", "both"),
            ("CustomVoice only - built-in speakers", "custom"),
            ("Base only - voice cloning (converting then requires --voice)", "clone"),
        ])


def ask_backend() -> str:
    return ask_menu(
        "Which inference backend was audiocpp_server built for?",
        [
            ("cuda - NVIDIA GPUs (fastest)", "cuda"),
            ("vulkan - cross-vendor GPU", "vulkan"),
            ("hip - AMD GPUs", "hip"),
            ("cpu - no GPU required", "cpu"),
        ])


def ask_distinct_clone_id(primary_id: str) -> str:
    """Prompt until a non-empty id different from PRIMARY_ID is entered."""
    prompt = (f"Enter a new id for the cloning (Base) model "
              f"(must differ from '{primary_id}'): ")
    while True:
        try:
            answer = input(prompt).strip()
        except EOFError:
            print()
            sys.exit("[FATAL] No interactive input available to resolve the "
                     "duplicate model id; give the two models distinct "
                     "AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID values in "
                     "converter/config.py first")
        if answer and answer != primary_id:
            return answer
        print(f"[WARNING] The id must be unique; it cannot be empty or "
              f"equal to '{primary_id}'.")


def config_port() -> int:
    """Return the port of AUDIOCPP_API_URL in converter/config.py."""
    try:
        return urllib.parse.urlsplit(config.AUDIOCPP_API_URL).port or FALLBACK_PORT
    except ValueError:
        return FALLBACK_PORT


def _url_with_port(url: str, port: int) -> str:
    parts = urllib.parse.urlsplit(url)
    host = parts.hostname or "127.0.0.1"
    return urllib.parse.urlunsplit(
        (parts.scheme or "http", f"{host}:{port}", parts.path, "", ""))


def update_config_api_url_port(port: int, config_path: Optional[Path] = None) -> bool:
    """Rewrite the port inside AUDIOCPP_API_URL in converter/config.py.

    Only the quoted URL literal is replaced; surrounding lines and the
    trailing comment are preserved. Returns True when the file was changed.
    """
    path = Path(config_path) if config_path is not None else CONFIG_PATH
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return False
    match = re.search(r'(?m)^(\s*AUDIOCPP_API_URL\s*=\s*")([^"]*)(")', text)
    if not match:
        return False
    new_url = _url_with_port(match.group(2), port)
    if new_url == match.group(2):
        return False
    text = text[:match.start(2)] + new_url + text[match.end(2):]
    try:
        path.write_text(text, encoding="utf-8")
    except OSError:
        return False
    return True


def update_config_model_ids(model_id: str,
                            clone_model_id: Optional[str] = None,
                            config_path: Optional[Path] = None) -> bool:
    """Rewrite AUDIOCPP_MODEL_ID (and AUDIOCPP_CLONE_MODEL_ID when given).

    Only the quoted id literals are replaced; surrounding lines and
    comments are preserved. Returns True when the file was changed.
    """
    path = Path(config_path) if config_path is not None else CONFIG_PATH
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return False
    updates: List[Tuple[str, str]] = [("AUDIOCPP_MODEL_ID", model_id)]
    if clone_model_id is not None:
        updates.append(("AUDIOCPP_CLONE_MODEL_ID", clone_model_id))
    changed = False
    for name, value in updates:
        match = re.search(r'(?m)^(\s*' + name + r'\s*=\s*")([^"]*)(")', text)
        if match and match.group(2) != value:
            text = text[:match.start(2)] + value + text[match.end(2):]
            changed = True
    if not changed:
        return False
    try:
        path.write_text(text, encoding="utf-8")
    except OSError:
        return False
    return True


def default_model_id(family: str) -> str:
    """Derive a default server entry id from a family name."""
    if family in PREFERRED_IDS:
        return PREFERRED_IDS[family]
    name = family
    if name.endswith("_tts"):
        name = name[:-4]
    return name.replace("_", "") or family


def detect_audiocpp_dir() -> Optional[Path]:
    """Best-effort location of a local audio.cpp checkout with model_specs.

    Checks the AUDIOCPP_DIR environment variable, then an ``audio.cpp``
    directory in or above the current working directory. Returns the path
    only when it contains a ``model_specs`` directory.
    """
    candidates: List[Path] = []
    env_dir = os.environ.get("AUDIOCPP_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    cwd = Path.cwd()
    candidates.append(cwd / "audio.cpp")
    candidates.append(cwd.parent / "audio.cpp")
    candidates.append(cwd.parent.parent / "audio.cpp")
    for candidate in candidates:
        try:
            resolved = candidate.resolve()
        except OSError:
            continue
        if (resolved / "model_specs").is_dir():
            return resolved
    return None


def _default_package(spec: dict) -> Optional[dict]:
    """Pick the default installable package from a model spec.

    Prefers the package flagged ``default: true``, then the first GGUF
    package, then the first package overall. Returns None if the spec
    declares no packages.
    """
    packages = spec.get("packages") or []
    if not packages:
        return None
    for package in packages:
        if package.get("default"):
            return package
    for package in packages:
        if package.get("format") == "gguf":
            return package
    return packages[0]


def load_model_catalog(audiocpp_dir: Path) -> List[dict]:
    """Read model_specs/*.json and return the TTS-capable families.

    Each returned entry has: family, display_name, description, languages,
    clone_capable, install_id (default package id), default_path
    (``models/<target_directory>``), tested, and preferred_id. Tested
    families come first (in TESTED_FAMILIES order), the rest follow
    alphabetically by display name.
    """
    specs_dir = audiocpp_dir / "model_specs"
    if not specs_dir.is_dir():
        raise NotADirectoryError(
            f"{audiocpp_dir} has no model_specs/ directory; point "
            "--audiocpp-dir at an audio.cpp checkout")
    entries: List[dict] = []
    for spec_path in sorted(specs_dir.glob("*.json")):
        try:
            spec = json.loads(spec_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        tasks = spec.get("tasks") or []
        if "tts" not in tasks and spec.get("category") != "tts":
            continue
        family = spec.get("family") or spec_path.stem
        package = _default_package(spec)
        if package is None:
            # No installable package: skip (cannot be hosted from a path).
            continue
        target_directory = package.get("target_directory") or family
        languages = spec.get("languages") or []
        display_name = spec.get("display_name") or family
        description = spec.get("description") or ""
        entries.append({
            "family": family,
            "display_name": display_name,
            "description": description,
            "languages": languages,
            "clone_capable": "clone" in tasks,
            "install_id": package.get("id") or family,
            "default_path": f"models/{target_directory}",
            "tested": family in TESTED_FAMILIES,
            "preferred_id": default_model_id(family),
        })

    def sort_key(entry: dict) -> tuple:
        family = entry["family"]
        if family in TESTED_FAMILIES:
            return (0, TESTED_FAMILIES.index(family), "")
        return (1, 0, entry["display_name"].lower())

    entries.sort(key=sort_key)
    return entries


def ask_families(catalog: List[dict]) -> List[str]:
    """Show a numbered checklist and return the chosen family keys.

    Input is comma/space-separated numbers; Enter alone selects the first
    entry (the default Qwen3-TTS flow). At least one family is required.
    """
    print("Select TTS model families to host (comma-separated numbers,")
    print("or press Enter for the default Qwen3-TTS flow):")
    for number, entry in enumerate(catalog, 1):
        marker = " [tested with this converter]" if entry["tested"] else ""
        langs = entry["languages"]
        lang_text = ", ".join(langs[:6]) + ("..." if len(langs) > 6 else "")
        if entry["family"] == FAMILY_QWEN3_TTS:
            caps = "built-in speakers + voice cloning"
        elif entry["clone_capable"]:
            caps = "voice cloning"
        else:
            caps = "TTS (no cloning)"
        detail = f"({lang_text}; {caps})" if lang_text else f"({caps})"
        print(f"  {number}) {entry['display_name']}{marker} {detail}")
    while True:
        try:
            answer = input("Choice [1]: ").strip()
        except EOFError:
            return [catalog[0]["family"]]
        if not answer:
            return [catalog[0]["family"]]
        parts = [p for p in re.split(r"[,\s]+", answer) if p]
        indices: List[int] = []
        valid = True
        for part in parts:
            if part.isdigit() and 1 <= int(part) <= len(catalog):
                indices.append(int(part))
            else:
                valid = False
                break
        if valid and indices:
            chosen: List[str] = []
            seen = set()
            for index in indices:
                family = catalog[index - 1]["family"]
                if family not in seen:
                    seen.add(family)
                    chosen.append(family)
            return chosen
        print(f"Please enter comma-separated numbers between 1 and {len(catalog)}.")


def build_model_entry(family: str, model_id: str, model_path: str) -> dict:
    """Assemble one server.json model entry."""
    return {
        "id": model_id,
        "family": family,
        "path": model_path,
        "task": "tts",
        "mode": "offline",
    }


def build_server_config(host: str, port: int, backend: str, lazy_load: bool,
                        model_entries: List[dict],
                        voice_dir: Optional[str] = None) -> dict:
    """Assemble the server.json document.

    ``voice_dir`` is a server-level cloning voice library; when set, every
    hosted clone-capable family can use its voices with ``--voice``.
    """
    config_doc = {
        "host": host,
        "port": port,
        "backend": backend,
        "lazy_load": lazy_load,
        "models": model_entries,
    }
    if voice_dir:
        config_doc["voice_dir"] = voice_dir
    return config_doc


def transcribe_wav_dir(wav_files: list, whisper_model: str) -> Dict[str, str]:
    """Transcribe each wav file and return a mapping of stem -> transcript."""
    transcripts: Dict[str, str] = {}
    for wav_file in wav_files:
        name = wav_file.stem
        print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...")
        text = transcribe_reference_audio(str(wav_file), model_name=whisper_model)
        if text:
            print(f"[OK] {name}: {text}")
        else:
            print(f"[WARNING] No transcript for '{name}'; cloning works best "
                  "with an accurate transcript — consider editing prompt_text "
                  "by hand before starting the server")
        transcripts[name] = text or ""
    return transcripts


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

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


def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None:
    """Print a loud, final warning for voices whose transcript is empty."""
    empty = sorted(name for name, text in transcripts.items() if not text)
    if not empty:
        return
    bar = "=" * 70
    print()
    print(bar)
    print("[WARNING] MANUAL TRANSCRIPTION REQUIRED")
    print(bar)
    listing = "  - " + "\n  - ".join(empty) if len(empty) > 1 else f"  - {empty[0]}"
    print(f"The following voice(s) have an EMPTY transcript in prompt_text:\n"
          f"{listing}")
    print("Those voices will NOT work until you add an accurate transcript.")
    print(f"Edit {PROMPT_TEXT_FILENAME} in your voice directory and fill in the "
          "text after '|' for each voice above.")
    print(bar)


def _ask_host_port_backend_lazy(args: argparse.Namespace,
                                default_lazy: bool
                                ) -> Tuple[str, int, str, bool]:
    """Ask for (or take from flags) the shared server settings."""
    host = args.host if args.host else ask("Bind host", DEFAULT_HOST)
    port = args.port if args.port is not None else ask_port(config_port())
    if port != config_port():
        if ask_bool(f"Update AUDIOCPP_API_URL in converter/config.py to port "
                    f"{port} so audiobook.py talks to this server", True):
            if update_config_api_url_port(port):
                print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}")
            else:
                print(f"[WARNING] Could not update {CONFIG_PATH}; edit "
                      "AUDIOCPP_API_URL by hand so audiobook.py uses the "
                      "new port")
        else:
            print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py "
                  f"will still use port {config_port()}")
    backend = args.backend if args.backend else ask_backend()
    lazy_load = args.lazy_load or ask_bool(
        "Load models lazily (on first use instead of at startup)", default_lazy)
    return host, port, backend, lazy_load


def _collect_transcripts(args: argparse.Namespace,
                         include_clone: bool) -> Dict[str, str]:
    """Transcribe the wav directory into a stem -> transcript mapping.

    Returns the mapping (empty when no wavs were found or cloning is not
    used by any selected family). Runs only when a cloning voice library is
    needed; a run without any clone-capable family ignores the wav directory
    entirely.
    """
    if not include_clone:
        print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family "
              "selected, so voice presets are not used")
        return {}

    wav_files = find_wav_files(args.input_dir)
    if not wav_files:
        print(f"[WARNING] No .wav files found in {args.input_dir}; writing the "
              "config without a voice_dir")
        return {}
    if whisper_backend_available() is None:
        print("[WARNING] Neither faster_whisper nor whisper was found, so "
              "reference .wav files cannot be transcribed automatically and "
              "every transcript will be empty.")
        print('          Did you remember to "conda activate qwen3-tts"? '
              "Transcripts must be added by hand (see the warning at the end).")
    return transcribe_wav_dir(wav_files, args.whisper_model)


def _offer_config_model_id_sync(model_id: str) -> None:
    """Offer to point converter/config.py at a single non-Qwen model entry.

    The converter requests the model id configured in AUDIOCPP_MODEL_ID,
    and single-model servers use the same id for the clone entry, so both
    ids are rewritten together.
    """
    if config.AUDIOCPP_MODEL_ID == model_id \
            and config.AUDIOCPP_CLONE_MODEL_ID == model_id:
        return
    if ask_bool("Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID in "
                f"converter/config.py to '{model_id}' so audiobook.py uses "
                "this model", True):
        if update_config_model_ids(model_id, model_id):
            print(f"[OK] Updated the model ids in {CONFIG_PATH}")
        else:
            print(f"[WARNING] Could not update {CONFIG_PATH}; edit "
                  "AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID by hand so "
                  "audiobook.py uses this model")
    else:
        print("[WARNING] Left the model ids unchanged; audiobook.py will "
              f"still request model '{config.AUDIOCPP_MODEL_ID}'")


def _print_multi_model_model_id_note(entry_ids: List[str]) -> None:
    """Tell the user how to select one entry per run for a multi-model server."""
    print("[INFO] Several model entries were configured. audiobook.py uses one "
          "entry per run: pass --model <id> when converting, or set "
          "AUDIOCPP_MODEL_ID in converter/config.py to one of: "
          f"{', '.join(entry_ids)}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Generate a server.json for the audio.cpp audiocpp_server "
                    "hosting one or more TTS model families used by this converter.")
    parser.add_argument("input_dir", type=resolve_wav_dir_arg, metavar="WAV_DIR",
                        help="Directory with .wav reference files to publish as "
                             "a server-level voice_dir cloning library (required)")
    parser.add_argument("--output", type=Path, default=Path("server.json"),
                        help="Output path for server.json (default: "
                             "server.json in the current directory)")
    parser.add_argument("--audiocpp-dir", type=Path, default=None,
                        help="Path to a local audio.cpp checkout containing a "
                             "model_specs/ directory (default: detected from "
                             "AUDIOCPP_DIR or an audio.cpp directory next to/above "
                             "the current working directory; prompted otherwise)")
    parser.add_argument("--families", type=str, default=None,
                        help="Comma-separated model families to host, as named "
                             "in the audio.cpp catalog (e.g. "
                             "qwen3_tts,higgs_audio_tts). Skips the family checklist")
    parser.add_argument("--models", choices=MODEL_SELECTIONS, default=None,
                        help="Which Qwen3-TTS models to host: both (default), "
                             "custom (CustomVoice speakers only), or clone "
                             "(Base voice cloning only). Only valid when the "
                             "qwen3_tts family is selected")
    parser.add_argument("--host", type=str, default=None,
                        help="Bind host for the server (default: 127.0.0.1)")
    parser.add_argument("--port", type=int, default=None,
                        help="Port for the server (default: the port in "
                             "AUDIOCPP_API_URL from converter/config.py)")
    parser.add_argument("--backend", choices=BACKENDS, default=None,
                        help="Inference backend audiocpp_server was built "
                             "for (default: cuda)")
    parser.add_argument("--lazy-load", action="store_true",
                        help="Load models on first use instead of at startup "
                             "(default: on when more than one model is hosted)")
    parser.add_argument("--whisper-model", type=str, default="base",
                        help="Whisper model size for transcription "
                             "(default: base)")
    parser.add_argument("--force", action="store_true",
                        help="Overwrite the output file (and prompt_text) "
                             "without prompting")
    args = parser.parse_args()

    if not args.input_dir.is_dir():
        parser.error(
            f"WAV directory not found: {args.input_dir}\n"
            f"      (resolved from the current working directory: "
            f"{Path.cwd()})\n"
            "      WAV_DIR must be a directory containing the .wav "
            "reference files to use as voice cloning presets")

    # Resolve the audio.cpp checkout and load its model catalog.
    audiocpp_dir = args.audiocpp_dir
    if audiocpp_dir is None:
        audiocpp_dir = detect_audiocpp_dir()
    if audiocpp_dir is None:
        audiocpp_dir = Path(ask("Path to your audio.cpp checkout", "") or "")
    if not audiocpp_dir:
        parser.error(
            "An audio.cpp checkout is required to read the model catalog. "
            "Clone one with `git clone https://github.com/0xShug0/audio.cpp` "
            "and pass --audiocpp-dir PATH (or set the AUDIOCPP_DIR environment "
            "variable)")
    audiocpp_dir = audiocpp_dir.resolve()
    if not audiocpp_dir.is_dir():
        parser.error(f"audio.cpp checkout not found: {audiocpp_dir}")
    try:
        catalog = load_model_catalog(audiocpp_dir)
    except NotADirectoryError as exc:
        parser.error(str(exc))
    if not catalog:
        parser.error(
            f"No TTS model families found in {audiocpp_dir}/model_specs; "
            "check the checkout is up to date")

    if args.output.exists() and not args.force \
            and not prompt_overwrite(args.output):
        print("[INFO] Aborted; existing server.json kept")
        return 1

    # Select families.
    if args.families is not None:
        requested = [f.strip() for f in args.families.split(",") if f.strip()]
        catalog_families = {entry["family"] for entry in catalog}
        unknown = [f for f in requested if f not in catalog_families]
        if unknown:
            parser.error(
                f"Unknown family in --families: {', '.join(unknown)}. "
                f"Available: {', '.join(entry['family'] for entry in catalog)}")
        family_keys: List[str] = []
        for fam in requested:
            if fam not in family_keys:
                family_keys.append(fam)
    else:
        family_keys = ask_families(catalog)

    catalog_by_family = {entry["family"]: entry for entry in catalog}
    is_qwen = FAMILY_QWEN3_TTS in family_keys
    if not is_qwen and args.models is not None:
        parser.error("--models only applies to the qwen3_tts family")
    if is_qwen and args.models is not None and len(family_keys) > 1 \
            and args.models != "both":
        parser.error(
            "--models custom/clone selects Qwen3-TTS sub-entries and is only "
            "valid when qwen3_tts is the sole selected family")

    print("[INFO] Model ids from converter/config.py:")
    print(f"       built-in speakers (CustomVoice): '{config.AUDIOCPP_MODEL_ID}'")
    print(f"       voice cloning (Base):            '{config.AUDIOCPP_CLONE_MODEL_ID}'")

    model_entries: List[dict] = []
    entry_ids: List[str] = []
    non_qwen_single_id: Optional[str] = None

    if is_qwen:
        selection = args.models if args.models is not None else ask_models()
        # When qwen3_tts is selected with other families, keep both entries so
        # speaker mode and cloning are both available; custom/clone sub-choice
        # is only honored when qwen3_tts is the sole family.
        if len(family_keys) > 1 and args.models is None:
            selection = "both"
        include_custom = selection in ("both", "custom")
        include_clone = selection in ("both", "clone")

        custom_voice_id = config.AUDIOCPP_MODEL_ID
        clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID
        if include_custom and include_clone and custom_voice_id == clone_model_id:
            print(f"[WARNING] AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID are "
                  f"both '{custom_voice_id}' in converter/config.py, but server "
                  "model ids must be unique.")
            clone_model_id = ask_distinct_clone_id(custom_voice_id)

        custom_voice_path = base_path = None
        if include_custom:
            custom_voice_path = ask("Path to the Qwen3-TTS CustomVoice GGUF package",
                                    DEFAULT_CUSTOM_VOICE_PATH)
            model_entries.append(build_model_entry(
                FAMILY_QWEN3_TTS, custom_voice_id, custom_voice_path))
            entry_ids.append(custom_voice_id)
        if include_clone:
            base_path = ask("Path to the Qwen3-TTS Base GGUF package",
                            DEFAULT_BASE_PATH)
            model_entries.append(build_model_entry(
                FAMILY_QWEN3_TTS, clone_model_id, base_path))
            entry_ids.append(clone_model_id)
        qwen_include_clone = include_clone
    else:
        qwen_include_clone = False

    # Non-Qwen families: one entry each.
    for family in family_keys:
        if family == FAMILY_QWEN3_TTS:
            continue
        entry = catalog_by_family[family]
        model_id = entry["preferred_id"]
        # Ensure uniqueness against already-chosen ids.
        if model_id in entry_ids:
            model_id = ask(f"Server model id for {entry['display_name']}",
                           f"{model_id}-2")
        model_path = entry["default_path"]
        # For a single non-Qwen family, ask the path (matching the old flow);
        # for several, use the catalog default to keep the prompt count sane.
        if len(family_keys) == 1:
            model_path = ask(f"Path to the {entry['display_name']} package",
                             model_path)
        model_entries.append(build_model_entry(family, model_id, model_path))
        entry_ids.append(model_id)
        if len(family_keys) == 1:
            non_qwen_single_id = model_id

    # Whether any selected family can clone (drives voice_dir / wav transcription).
    include_clone = qwen_include_clone or any(
        catalog_by_family[f]["clone_capable"]
        for f in family_keys if f != FAMILY_QWEN3_TTS)

    # Default to lazy loading only when hosting more than one family: a
    # single-family server (including the Qwen3-TTS CustomVoice+Base pair)
    # loads at startup as before, while a multi-family server avoids loading
    # every model until it is actually used.
    default_lazy = len(family_keys) > 1
    host, port, backend, lazy_load = _ask_host_port_backend_lazy(args, default_lazy)

    transcripts = _collect_transcripts(args, include_clone)

    voice_dir: Optional[str] = None
    if transcripts:
        prompt_path = args.input_dir / PROMPT_TEXT_FILENAME
        if prompt_path.exists() and not args.force:
            if not ask_bool(f"Overwrite existing {prompt_path}", True):
                print(f"[INFO] Kept existing {prompt_path}; new transcripts "
                      "were not written")
            else:
                write_prompt_text(args.input_dir, transcripts)
                print(f"[OK] Wrote {prompt_path}")
        else:
            write_prompt_text(args.input_dir, transcripts)
            print(f"[OK] Wrote {prompt_path}")
        voice_dir = str(args.input_dir.resolve())

    server_config = build_server_config(
        host=host, port=port, backend=backend, lazy_load=lazy_load,
        model_entries=model_entries, voice_dir=voice_dir)

    print("\nGenerated server.json:")
    print(json.dumps(server_config, indent=2, ensure_ascii=False))
    if not ask_bool(f"\nWrite this to {args.output}", True):
        print("[INFO] Aborted; nothing written")
        return 1

    with args.output.open("w", encoding="utf-8") as handle:
        json.dump(server_config, handle, indent=2, ensure_ascii=False)
        handle.write("\n")

    # Post-generation guidance.
    print(f"\n[OK] Wrote {args.output} with {len(model_entries)} model entry/entries"
          + (f" and voice_dir '{voice_dir}'" if voice_dir else ""))
    for family in family_keys:
        entry = catalog_by_family[family]
        if family == FAMILY_QWEN3_TTS:
            print("[INFO] Install the Qwen3-TTS packages from the audio.cpp "
                  "checkout:")
            print("       python3 tools/model_manager_v2.py install "
                  "qwen3_tts_1_7b_customvoice_q8_0")
            print("       python3 tools/model_manager_v2.py install "
                  "qwen3_tts_1_7b_base_q8_0")
        else:
            print(f"[INFO] Install {entry['display_name']} from the audio.cpp "
                  f"checkout: python3 tools/model_manager_v2.py install "
                  f"{entry['install_id']}")
    if len(model_entries) > 1:
        print("[INFO] Models load lazily and stay in memory until the server "
              "exits; restart the server (or POST /v1/tasks/unload_models) "
              "before switching to a large model to free VRAM.")
    if family_keys != [FAMILY_QWEN3_TTS]:
        for family in family_keys:
            if family == FAMILY_QWEN3_TTS:
                continue
            entry = catalog_by_family[family]
            print(f"[INFO] Clone-only family {entry['display_name']}: run "
                  "audiobook.py with --backend audiocpp --voice <preset name>")
    if len(entry_ids) == 1 and non_qwen_single_id is not None:
        _offer_config_model_id_sync(non_qwen_single_id)
    elif len(entry_ids) > 1:
        _print_multi_model_model_id_note(entry_ids)
    print_empty_transcript_warning(transcripts)
    return 0


if __name__ == "__main__":
    sys.exit(main())