From f00249db9d1ea051d29aa1bcca869fc4b88e83eb Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 02:59:26 -0400 Subject: refactor: add app directory, dir structure change --- app/backends/qwen.py | 272 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 app/backends/qwen.py (limited to 'app/backends/qwen.py') diff --git a/app/backends/qwen.py b/app/backends/qwen.py new file mode 100644 index 0000000..21280a0 --- /dev/null +++ b/app/backends/qwen.py @@ -0,0 +1,272 @@ +#!/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 as a +TUI: pip-install the package, configure the two ports and the built-in +speaker in ``app/converter/config.py``, and print the launch commands. It is +driven by ``audiobook.py``'s hub but can also be run directly with flags. + +Usage: + python app/backends/qwen.py [--port-custom PORT] [--port-clone PORT] + [--speaker NAME] [--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, + ConfigureAction, + ServerSpec, + common, + envs, + format_launch_hint, +) +from converter import config +from ui import tui + +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). +QWEN_SPEAKERS = ("Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric", "Ryan", + "Aiden", "Ono_Anna", "Sohee") + + +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) -> Optional[dict]: + """Linear TUI wizard collecting every qwen-setup decision.""" + _GO_BACK = object() + + def confirm(question: str, default: bool = True) -> Optional[bool]: + res = tui.confirm(stdscr, question, default=default, + cancel_value=_GO_BACK) + return None if res is _GO_BACK else res + + # Step 0: pip install (if not installed and not skipped). + do_install = False + if not _is_installed() and not args.skip_install: + choice = confirm("qwen-tts is not installed. pip install it now?", + default=True) + if choice is None: + return None + do_install = choice + + # Step 1: ports. + custom_port = args.port_custom + if custom_port is None: + port_text = tui.line_edit( + stdscr, "CustomVoice (built-in speaker) port", + str(_config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT)), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535", + help_lines=["The port for qwen-tts-demo CustomVoice (speaker mode)"]) + custom_port = int(port_text) + clone_port = args.port_clone + if clone_port is None: + port_text = tui.line_edit( + stdscr, "Base (voice clone) port", + str(_config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT)), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535", + help_lines=["The port for qwen-tts-demo Base (voice cloning)"]) + clone_port = int(port_text) + + # Step 2: built-in speaker. + speaker = args.speaker + if speaker is None: + speaker = tui.menu( + stdscr, "Built-in CustomVoice speaker", + [(s, s) for s in QWEN_SPEAKERS], + default_index=max(0, QWEN_SPEAKERS.index(config.SPEAKER) + if config.SPEAKER in QWEN_SPEAKERS else 0), + help_lines=["Used by audiobook.py --backend qwen without --clone"]) + + return { + "do_install": do_install, + "custom_port": custom_port, + "clone_port": clone_port, + "speaker": speaker, + } + + +def _execute(settings: dict) -> int: + """Console tail: install, sync config, advise.""" + if settings["do_install"]: + rc = common.pip_install([QWEN_PIP_PKG]) + 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") + + custom_url = common.url_with_port(config.QWEN_API_URL, settings["custom_port"]) + if custom_url != config.QWEN_API_URL: + if common.update_config_value("QWEN_API_URL", custom_url): + print(f"[OK] Updated QWEN_API_URL to {custom_url}") + else: + print("[WARNING] Could not update QWEN_API_URL; edit " + "app/converter/config.py by hand") + clone_url = common.url_with_port(config.CLONE_API_URL, settings["clone_port"]) + if clone_url != config.CLONE_API_URL: + if common.update_config_value("CLONE_API_URL", clone_url): + print(f"[OK] Updated CLONE_API_URL to {clone_url}") + else: + print("[WARNING] Could not update CLONE_API_URL; edit " + "app/converter/config.py by hand") + if settings["speaker"] != config.SPEAKER: + if common.update_config_value("SPEAKER", settings["speaker"]): + print(f"[OK] Updated SPEAKER to {settings['speaker']}") + else: + print("[WARNING] Could not update SPEAKER; edit " + "app/converter/config.py by hand") + + _print_launch_hint(settings["custom_port"], settings["clone_port"]) + return 0 + + +def _print_launch_hint(custom_port: int, clone_port: int) -> None: + demo = envs.env_script("qwen-tts-demo") + print() + print("Start the servers (in separate terminals), or use the hub's") + print("'Server' menu / let a conversion start one automatically:") + print(f" {demo} {QWEN_CUSTOMVOICE_MODEL} --ip 127.0.0.1 " + f"--port {custom_port}") + print(f" {demo} {QWEN_BASE_MODEL} --ip 127.0.0.1 " + f"--port {clone_port}") + print("Then run: python audiobook.py --backend qwen") + + +def run_tui(args: Optional[argparse.Namespace] = None) -> int: + """Run the qwen setup wizard end-to-end.""" + import curses + if args is None: + args = build_parser().parse_args([]) + try: + settings = curses.wrapper(_wizard, args) + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + try: + curses.curs_set(1) + except curses.error: + pass + if settings is None: + print("[INFO] Aborted") + return 1 + return _execute(settings) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> dict: + return { + "do_install": (not _is_installed()) and not args.skip_install, + "custom_port": args.port_custom if args.port_custom is not None + else _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT), + "clone_port": args.port_clone if args.port_clone is not None + else _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT), + "speaker": args.speaker or config.SPEAKER, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Set up the Qwen3-TTS demo backend: pip install, " + "configure ports/speaker, and print launch commands.") + parser.add_argument("--port-custom", type=int, default=None, + help="CustomVoice (speaker) port (default: " + f"{DEFAULT_CUSTOM_PORT})") + parser.add_argument("--port-clone", type=int, default=None, + help="Base (voice clone) port (default: " + f"{DEFAULT_CLONE_PORT})") + parser.add_argument("--speaker", type=str, default=None, + choices=QWEN_SPEAKERS, + help="Built-in CustomVoice speaker (default: " + f"{config.SPEAKER})") + 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) + # Running when either server is up — CustomVoice (speaker mode) or Base + # (voice clone) each suffice for a conversion on their own. + running = (common.server_running(config.QWEN_API_URL) + or common.server_running(config.CLONE_API_URL)) + 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")) + servers = [ + ServerSpec("qwen-custom", config.QWEN_API_URL, + [demo, QWEN_CUSTOMVOICE_MODEL, "--ip", "127.0.0.1", + "--port", str(custom_port)]), + ServerSpec("qwen-clone", config.CLONE_API_URL, + [demo, QWEN_BASE_MODEL, "--ip", "127.0.0.1", + "--port", str(clone_port)]), + ] + return BackendStatus("qwen", "qwen-tts", + installed=installed, configured=installed, + running=running, details=details, + launch_hint=format_launch_hint(servers), + servers=servers) + + +configure_actions: List[ConfigureAction] = [ + ConfigureAction("Reconfigure qwen-tts (ports/speaker)", run_tui), +] + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + if _interactive(): + return run_tui(args) + + settings = _collect_from_flags(args, parser) + return _execute(settings) + + +def _interactive() -> bool: + try: + import curses # noqa: F401 + except ImportError: + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +if __name__ == "__main__": + sys.exit(main()) -- cgit v1.2.3