path: root/app/backends/qwen.py

#!/usr/bin/env python3
"""Set up the Qwen3-TTS demo backend for the audiobook generator.

qwen-tts is a pip package providing the ``qwen-tts-demo`` server, which
hosts the Qwen3-TTS CustomVoice (built-in speakers) and Base (voice
cloning) models on separate ports. This module sets it up end-to-end:
pip-install the package into the managed venv. There are no questions to
ask — the ports live in ``app/converter/config.py`` (edit them in the
hub's Settings screen) and the speaker is chosen per run on the
Generate-audiobooks screen. It is driven by ``audiobook.py``'s hub but
can also be run directly:

Usage:
    python app/backends/qwen.py [--skip-install]
"""

import argparse
import sys
from pathlib import Path
from typing import List, Optional

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from backends import (
    BackendStatus,
    ServerSpec,
    common,
    envs,
    format_launch_hint,
    probe,
    servers,
    setup,
)
from converter import config
from converter.clients import QWEN3_TTS_SPEAKERS
from ui import taskview

QWEN_PIP_PKG = "qwen-tts"
QWEN_CUSTOMVOICE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
QWEN_BASE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-Base"
DEFAULT_CUSTOM_PORT = 7860
DEFAULT_CLONE_PORT = 7861

# Built-in CustomVoice speakers (see app/converter/config.py SPEAKER). The
# canonical list lives in converter.clients.speakers (shared with the
# audio.cpp backend's Convert-form Speaker picker).
QWEN_SPEAKERS = QWEN3_TTS_SPEAKERS


def _is_installed() -> bool:
    if envs.env_script("qwen-tts-demo").is_file():
        return True
    return envs.module_available("qwen_tts")


def _config_port(url: str, fallback: int) -> int:
    import urllib.parse
    try:
        return urllib.parse.urlsplit(url).port or fallback
    except ValueError:
        return fallback


def _wizard(stdscr, args: argparse.Namespace) -> dict:
    """Collect the setup settings without asking anything.

    The qwen backend has no per-install choices: install happens when the
    package is missing (and not skipped by flag), and every other value —
    ports, speaker — lives in app/converter/config.py, managed from the
    hub's Settings and Generate-audiobooks screens.
    """
    return {
        "do_install": (not _is_installed()) and not args.skip_install,
    }


def _execute_steps(settings: dict) -> List[taskview.TaskStep]:
    """Build the ordered setup steps for the in-TUI task view.

    The same work ``_execute`` runs on the console. The pip install streams
    through EMIT and aborts on CANCEL; the list is empty (no-op) when
    there is nothing to install.
    """
    steps: List[taskview.TaskStep] = []
    if not settings["do_install"]:
        return steps

    def install(emit, cancel):
        rc = common.pip_install([QWEN_PIP_PKG], emit=emit, cancel=cancel)
        if rc != 0:
            print(f"[WARNING] pip install failed (exit {rc}); install "
                  f"{QWEN_PIP_PKG} manually")
        else:
            print(f"[OK] {QWEN_PIP_PKG} installed")
        return rc

    steps.append(taskview.TaskStep(f"Install {QWEN_PIP_PKG}", install))
    return steps


def _execute(settings: dict) -> int:
    """Console tail: pip install (no-op when already installed)."""
    return taskview.run_steps_inline(_execute_steps(settings))


def setup_screen(stdscr) -> int:
    """Run the setup on an existing curses screen (the hub's).

    There are no questions: settings are computed up front and the install
    runs inside the TUI task view on this same screen — skipped entirely
    (a silent no-op) when nothing needs installing. Returns 0 always —
    the flow cannot be aborted, so Esc/Ctrl-C never short-circuits it.
    """
    args = build_parser().parse_args([])
    settings = _wizard(stdscr, args)
    if not settings["do_install"]:
        return 0
    return taskview.run_steps(stdscr, "Setting up qwen-tts",
                              _execute_steps(settings))


def run_tui(args: Optional[argparse.Namespace] = None) -> int:
    """Run the qwen setup end-to-end."""
    if args is None:
        args = build_parser().parse_args([])
    return _execute(_wizard(None, args))


def _collect_from_flags(args: argparse.Namespace,
                        parser: argparse.ArgumentParser) -> dict:
    return {
        "do_install": (not _is_installed()) and not args.skip_install,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Set up the Qwen3-TTS demo backend: pip install "
                    "qwen-tts into the managed venv.")
    parser.add_argument("--skip-install", action="store_true",
                        help="Do not pip install qwen-tts")
    return parser


def detect() -> BackendStatus:
    """Detect whether qwen-tts is installed, plus the launch commands."""
    installed = _is_installed()
    custom_port = _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT)
    clone_port = _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT)
    details: List[str] = []
    details.append("pip: installed" if installed else
                   "not installed — run setup to pip install qwen-tts")
    details.append(f"CustomVoice port: {custom_port}")
    details.append(f"Base (clone) port: {clone_port}")
    details.append(f"speaker: {config.SPEAKER}")
    demo = str(envs.env_script("qwen-tts-demo"))
    specs = [
        ServerSpec("qwen-custom", config.QWEN_API_URL,
                   [demo, QWEN_CUSTOMVOICE_MODEL, "--ip", "127.0.0.1",
                    "--port", str(custom_port)],
                   identity=probe.IDENTITY_QWEN_CUSTOM),
        ServerSpec("qwen-clone", config.CLONE_API_URL,
                   [demo, QWEN_BASE_MODEL, "--ip", "127.0.0.1",
                    "--port", str(clone_port)],
                   identity=probe.IDENTITY_QWEN_CLONE),
    ]
    managed = servers.manages(specs)
    # Which local servers this tool started (pid alive) name the running
    # models; a remotely-run demo names them via the probe instead.
    local_models = [name for name, spec in
                    (("Base", specs[1]), ("CustomVoice", specs[0]))
                    if servers.alive(spec.name)]
    remote_models, remote_urls = _detect_remote(managed)
    running_models = list(dict.fromkeys(local_models + remote_models))
    return BackendStatus("qwen", "qwen-tts",
                         installed=installed, configured=installed,
                         running=managed or bool(remote_urls),
                         details=details,
                         launch_hint=format_launch_hint(specs),
                         servers=specs,
                         managed=managed,
                         remote=bool(remote_urls),
                         remote_urls=remote_urls,
                         remote_models=remote_models,
                         running_models=running_models)


def _detect_remote(managed: bool = False):
    """Detect externally-run qwen demo servers at the remote URLs.

    Returns ``([model, ...], {spec_name: url})``. Each remote URL (CustomVoice
    and Base) is probed independently and must answer as the matching demo
    (see probe.identify_server); a remote URL equal to the local URL for a
    server this tool started is ignored (already reported "[local]").
    """
    remote_models = []
    remote_urls = {}
    for spec_name, url, local_url, identity in (
            ("qwen-clone", config.CLONE_REMOTE_URL, config.CLONE_API_URL,
             probe.IDENTITY_QWEN_CLONE),
            ("qwen-custom", config.QWEN_REMOTE_URL, config.QWEN_API_URL,
             probe.IDENTITY_QWEN_CUSTOM)):
        url = (url or "").strip()
        if not url:
            continue
        if managed and probe.same_endpoint(url, local_url):
            continue
        if probe.identify_server(url) == identity:
            remote_urls[spec_name] = url
            remote_models.append(
                "Base" if spec_name == "qwen-clone" else "CustomVoice")
    return remote_models, remote_urls


def uninstall(*, emit=None, cancel=None) -> int:
    """Remove the qwen-tts backend entirely: stop its servers, pip uninstall.

    qwen-tts is a pip package (``qwen_tts`` + the ``qwen-tts-demo`` script)
    installed into the managed venv, so uninstalling it removes the backend.
    Any server this tool started is stopped first (best-effort).

    With EMIT given (the in-TUI task view) pip runs piped, streaming into
    EMIT, so its output never touches the terminal behind curses. CANCEL is
    a ``threading.Event`` honored between phases only (after the servers
    have been stopped, before pip starts) — a started phase always completes,
    so pip is never killed mid-run. Returns the exit code (130 when
    cancelled before pip ran).
    """
    for name in ("qwen-custom", "qwen-clone"):
        # Only stop when a pid file exists: without one this tool never
        # started the server, so the "not started by this tool" notice
        # would be uninstall-time noise.
        if servers.pid_for(name) is not None:
            servers.stop(name)
    if common.cancel_requested(cancel):
        return 130
    rc = common.pip_uninstall([QWEN_PIP_PKG], emit=emit)
    if rc != 0:
        print(f"[WARNING] pip uninstall failed (exit {rc}); remove "
              f"{QWEN_PIP_PKG} from the managed venv manually")
    else:
        print(f"[OK] {QWEN_PIP_PKG} removed.")
    return rc


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if setup.interactive():
        return run_tui(args)

    settings = _collect_from_flags(args, parser)
    return _execute(settings)


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