diff options
| author | historia <historiavg@proton.me> | 2026-08-18 23:27:42 -0400 |
|---|---|---|
| committer | historia <historiavg@proton.me> | 2026-08-18 23:27:42 -0400 |
| commit | f3e21980320c1708ff17cc6f699a9aa4758accdf (patch) | |
| tree | 596ff71ba68600bb24a5a9d88424951c77e02808 | |
| parent | a97c0506f8b20cdc5ed8a11892ef10a9fc1938ef (diff) | |
| download | tts-audiobook-generator-f3e21980320c1708ff17cc6f699a9aa4758accdf.tar.gz | |
feat: support for faster-qwen3-tts backend server
| -rw-r--r-- | README.md | 78 | ||||
| -rwxr-xr-x | audiobook.py | 63 | ||||
| -rw-r--r-- | converter/config.py | 22 | ||||
| -rw-r--r-- | converter/converter.py | 72 | ||||
| -rw-r--r-- | converter/tts.py | 314 | ||||
| -rw-r--r-- | tests/test_converter.py | 4 | ||||
| -rw-r--r-- | tests/test_make_voices.py | 169 | ||||
| -rw-r--r-- | tests/test_tts.py | 220 | ||||
| -rwxr-xr-x | tools/make_voices.py | 120 |
9 files changed, 933 insertions, 129 deletions
@@ -2,7 +2,7 @@ Convert TXT, PDF, and EPUB files into audiobooks using the Qwen3-TTS voice model. -This builds upon [WhiskeyCoder/Qwen3-Audiobook-Converter](https://github.com/WhiskeyCoder/Qwen3-Audiobook-Converter) adding more output files, metadata, generated cover art, transcription/speed/language options, better text cleanup, and clearer instructions. It also expects the qwen-tts server to be on different ports per model, so two server processes can run at once. +This builds upon [WhiskeyCoder/Qwen3-Audiobook-Converter](https://github.com/WhiskeyCoder/Qwen3-Audiobook-Converter) adding more output files, support for a faster backend, metadata, generated cover art, transcription/speed/language options, better text cleanup, and clearer instructions. It also expects the qwen-tts server to be on different ports per model, so two server processes can run at once. ## Overview @@ -47,7 +47,7 @@ cd qwen3-audiobook-converter pip install -r requirements.txt ``` -## Running the Qwen server and audiobook script +## Running the Qwen Gradio server and audiobook script The audiobook script talks to a Qwen3-TTS Gradio server that is run using `qwen-tts-demo`. Add `--no-flash-attn` if FlashAttention isn't installed (see below). The script expects the custom voice model and base model to be on different ports depending on which you're using. The Qwen model(s) will automatically download. @@ -57,7 +57,7 @@ Put your book files (epub, etc.) in the `input/` folder. Then run the script. Th ```bash conda activate qwen3-tts -qwen-tts-demo Qwen/Qwen3-TTS-12Hz-1.7B-Base --ip 127.0.0.1 --port 7861 +qwen-tts-demo Qwen/Qwen3-TTS-12Hz-1.7B-Base --ip 127.0.0.1 --port 7861 --no-flash-attn ``` Then in another terminal: @@ -75,14 +75,76 @@ Whisper (`faster_whisper` or `whisper`) is used automatically to transcribe the ```bash conda activate qwen3-tts -qwen-tts-demo Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice --ip 127.0.0.1 --port 7860 +qwen-tts-demo Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice --ip 127.0.0.1 --port 7860 --no-flash-attn ``` ```bash python audiobook.py ``` -Edit `converter/config.py` to change which built-in voice is used. +Change the voice settings in `converter/config.py`. + + +## Using the `--faster` backend + +Instead of the Qwen Gradio demos, `--faster` talks to the OpenAI-compatible server from [faster-qwen3-tts](https://github.com/andimarafioti/faster-qwen3-tts), which uses CUDA graph capture for roughly 5-10x faster inference with the same models. **It requires an NVIDIA GPU**. + +**The `--faster` backend always uses voice cloning**. The reference voice and language are configured on the **server**, not through the converter. The server does not transcribe reference audio itself, so transcripts must come from you — either by hand, or with the `tools/make_voices.py` helper (see below). + +### Install + +Install into the **same `qwen3-tts` conda environment** used for the Gradio server. + +```bash +conda activate qwen3-tts +pip install "faster-qwen3-tts[demo]" +``` + +### Run the server + +The pip package does not include the server script, so clone the repository: + +```bash +git clone https://github.com/andimarafioti/faster-qwen3-tts +cd faster-qwen3-tts +``` + +Single voice (the voice is named `default`): + +```bash +python examples/openai_server.py \ + --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \ + --ref-audio /absolute/path/to/reference.wav \ + --ref-text "Transcript of the reference audio." \ + --language English --port 8000 +``` + +Multiple voices — create a `voices.json` mapping names to reference configurations: + +```json +{ + "default": {"ref_audio": "voice1.wav", "ref_text": "Transcript of voice 1.", "language": "English"}, + "obama": {"ref_audio": "voice2.wav", "ref_text": "Transcript of voice 2.", "language": "English"} +} +``` + +```bash +python examples/openai_server.py --voices voices.json --port 8000 +``` + +### Generating voices.json (optional) + +The `tools/make_voices.py` helper builds a `voices.json` for the server: it transcribes every `.wav` in a directory with using whisper (which is in the qwen3-tts environment). By default it puts voices.json into the input directory. Check the help with `-h` for more options. + +```bash +python tools/make_voices.py path/to/wavs +``` + +### Run the converter + +```bash +python audiobook.py --faster [--faster-voice NAME] +``` ## Options @@ -95,10 +157,14 @@ Edit `converter/config.py` to change which built-in voice is used. | `--speed <n>` | Playback speed, pitch-preserving (`1.0` = normal). A normal-speed copy is also output. | | `--single-file` | Merge all chapters into a single file (default: one file per chapter). `m4b` is always one file. | | `--language <lang>` | Output language for the synthesized speech. Can add an accent even if the text is English. | +| `--faster` | Use a faster-qwen3-tts OpenAI-compatible server (up to 5x faster in certain cases). | +| `--faster-voice <name>` | Chooses a voice from voices.json when using `--faster` with multiple voices. | + +Other options and defaults are configured in `converter/config.py` ## FlashAttention for qwen-tts-demo server (optional) -The server tries to use FlashAttention 2 by default, but `--no-flash-attn` works without it. On supported GPUs FlashAttention can give a modest speedup. +This is **not** used with the `--faster` backend. The Gradio server tries to use FlashAttention 2 by default, but requires `--no-flash-attn` without it. On supported GPUs FlashAttention can give a modest speedup. 1. Build from source (takes absolutely forever). If you run out of memory, lower MAX_JOBS until you don't. diff --git a/audiobook.py b/audiobook.py index 34806f2..ed92e53 100755 --- a/audiobook.py +++ b/audiobook.py @@ -37,6 +37,9 @@ Examples: # Use voice cloning with reference audio python audiobook.py --clone path/to/reference.wav + + # Use the faster-qwen3-tts server (voice cloning, configured server-side) + python audiobook.py --faster """ ) @@ -96,27 +99,65 @@ Examples: "Ignored for m4b, which is always a single file.") ) + parser.add_argument( + "--faster", + action="store_true", + help=("Use a faster-qwen3-tts OpenAI-compatible server instead of the Qwen " + "Gradio demos (5-10x faster inference via CUDA graphs). Always voice " + "cloning: the reference audio and transcript are configured on the " + "server itself (see the 'Faster backend' section of the README).") + ) + + parser.add_argument( + "--faster-voice", + type=str, + default=None, + metavar="NAME", + help=("Voice entry to request from the faster server's voice config " + "(default: the FASTER_TTS_VOICE setting in converter/config.py, " + "typically 'default'). Must match a key in the server's voices.json, " + "or 'default' when the server was started with --ref-audio.") + ) + args = parser.parse_args() if args.speed <= 0: parser.error(f"--speed must be a positive number (got {args.speed:g})") - if args.language is not None: - try: - args.language = normalize_language(args.language) - except ValueError as exc: - parser.error(str(exc)) - - if not args.clone and (args.transcription or args.no_transcription): - print("[WARNING] --transcription/--no-transcription " - "are ignored without --clone") + if args.faster: + if args.clone: + print("[WARNING] --clone is ignored with --faster: the faster backend " + "always uses voice cloning, and the reference voice is configured " + "on the faster server (see README)") + args.clone = None + if args.transcription or args.no_transcription: + print("[WARNING] --transcription/--no-transcription are ignored with " + "--faster: the reference transcript is configured on the faster " + "server (--ref-text or voices.json, see README)") + args.transcription = None + args.no_transcription = False + if args.language is not None: + print("[WARNING] --language is ignored with --faster: language is " + "configured on the faster server (see README)") + args.language = None + else: + if args.language is not None: + try: + args.language = normalize_language(args.language) + except ValueError as exc: + parser.error(str(exc)) + + if not args.clone and (args.transcription or args.no_transcription): + print("[WARNING] --transcription/--no-transcription " + "are ignored without --clone") setup_logging() setup_directories() try: converter = AudiobookConverter( - voice_mode=config.VOICE_MODE_CLONE if args.clone else config.VOICE_MODE_CUSTOM, + voice_mode=config.VOICE_MODE_CLONE if (args.clone or args.faster) + else config.VOICE_MODE_CUSTOM, voice_clone_ref_audio=args.clone, voice_clone_ref_text=args.transcription, skip_transcription=args.no_transcription, @@ -124,6 +165,8 @@ Examples: single_file=args.single_file, output_format=args.format, language=args.language, + faster=args.faster, + faster_voice=args.faster_voice, ) ok = converter.run() except KeyboardInterrupt: diff --git a/converter/config.py b/converter/config.py index f795520..efec0dd 100644 --- a/converter/config.py +++ b/converter/config.py @@ -108,6 +108,28 @@ VOICE_CLONE_SEED = -1 VOICE_CLONE_API_URL = "http://127.0.0.1:7861" # ============================================================================= +# FASTER TTS SETTINGS (optional --faster backend) +# ============================================================================= +# --faster talks to the OpenAI-compatible server from faster-qwen3-tts +# (examples/openai_server.py) instead of the qwen-tts Gradio demos. The +# reference voice (ref audio, ref text) and language are configured on the +# SERVER side (--ref-audio/--ref-text or a --voices JSON file); the converter +# only sends text. See the "Faster backend" section of the README. + +FASTER_TTS_API_URL = "http://127.0.0.1:8000" +# Voice entry to request. MUST match a key in the server's voices.json (or +# "default" when the server was launched with --ref-audio). NOTE: the stock +# server silently falls back to its first configured voice when the requested +# name is unknown, so a mismatch here is easy to miss. +FASTER_TTS_VOICE = "default" +FASTER_TTS_SAMPLE_RATE = 24000 # Qwen3-TTS 12Hz codec output rate +# The Gradio demo sub-chunked text server-side (~200 chars); the faster server +# takes one generation per request, so long chunks are sub-chunked client-side. +FASTER_SUBCHUNK_WORDS = 40 # ~200 chars per request +FASTER_HTTP_TIMEOUT = 300 # Seconds before a speech request times out +FASTER_SUBCHUNK_RETRIES = 3 # Attempts per sub-chunk request + +# ============================================================================= # PROCESSING SETTINGS # ============================================================================= diff --git a/converter/converter.py b/converter/converter.py index 4d20083..ec06bbb 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -13,7 +13,7 @@ from typing import Dict, List, Optional, Tuple from . import audio, chunking, config, cover, extractors from .audio import TrackMeta -from .tts import QwenTTSClient, normalize_language, speaker_display_name +from .tts import FasterTTSClient, QwenTTSClient, normalize_language, speaker_display_name logger = logging.getLogger(__name__) @@ -89,7 +89,8 @@ class AudiobookConverter: def __init__(self, voice_mode: str = config.VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None, voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, speed: float = 1.0, single_file: bool = False, output_format: str = config.AUDIO_FORMAT, - language: Optional[str] = None): + language: Optional[str] = None, faster: bool = False, + faster_voice: Optional[str] = None): if speed <= 0: raise ValueError(f"Speed must be a positive number, got {speed}") if output_format not in config.AUDIO_FORMATS: @@ -103,14 +104,21 @@ class AudiobookConverter: self.speed = speed self.single_file = single_file self.output_format = output_format + self.faster = faster + self.faster_voice = faster_voice self._validate_configuration() - self.tts = QwenTTSClient( - voice_mode=voice_mode, - voice_clone_ref_audio=voice_clone_ref_audio, - voice_clone_ref_text=voice_clone_ref_text, - skip_transcription=skip_transcription, - language=self.language, - ) + if faster: + # The faster backend always voice-clones using a reference voice + # configured on the server, so no local reference audio is needed. + self.tts = FasterTTSClient(voice=faster_voice) + else: + self.tts = QwenTTSClient( + voice_mode=voice_mode, + voice_clone_ref_audio=voice_clone_ref_audio, + voice_clone_ref_text=voice_clone_ref_text, + skip_transcription=skip_transcription, + language=self.language, + ) def _validate_configuration(self) -> None: """Validate configuration settings.""" @@ -119,7 +127,7 @@ class AudiobookConverter: f"Unknown voice mode: {self.voice_mode!r} " f"(expected one of {config.VOICE_MODES})" ) - if self.voice_mode == config.VOICE_MODE_CLONE: + if self.voice_mode == config.VOICE_MODE_CLONE and not self.faster: if not self.voice_clone_ref_audio: raise ValueError( "Voice Clone mode requires a reference audio file. " @@ -142,10 +150,13 @@ class AudiobookConverter: """Narrator name used in output file names. Custom voice mode uses the built-in speaker's display name; voice - clone mode uses the reference audio file's stem. Spaces become - underscores (e.g. "Uncle Fu" -> "Uncle_Fu"). + clone mode uses the reference audio file's stem; the faster backend + uses the server-side voice name. Spaces become underscores + (e.g. "Uncle Fu" -> "Uncle_Fu"). """ - if self.voice_mode == config.VOICE_MODE_CLONE: + if self.faster: + narrator = self.faster_voice or config.FASTER_TTS_VOICE + elif self.voice_mode == config.VOICE_MODE_CLONE: narrator = Path(self.voice_clone_ref_audio).stem else: narrator = speaker_display_name() @@ -371,24 +382,29 @@ class AudiobookConverter: logger.error(traceback.format_exc()) return False - def run(self) -> bool: - """Main conversion process. Returns True if all books converted.""" - api_url = config.VOICE_CLONE_API_URL if self.voice_mode == config.VOICE_MODE_CLONE else config.QWEN_API_URL - + def _print_banner(self) -> None: + """Print the startup summary for the selected backend.""" print("=" * 70) print("QWEN-BASED AUDIOBOOK CONVERTER") print("=" * 70) print(f"Books folder: {config.BOOKS_FOLDER}") print(f"Output folder: {config.AUDIOBOOKS_FOLDER}") - print(f"Qwen API endpoint: {api_url}") - print(f"Voice mode: {self.voice_mode}") - print("Model size: 1.7B (always)") - if self.voice_mode == config.VOICE_MODE_CUSTOM: - print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}") - print(f"Language: {self.language}") - elif self.voice_mode == config.VOICE_MODE_CLONE: - print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") - print(f"Language: {self.language}") + if self.faster: + print(f"Faster TTS endpoint: {config.FASTER_TTS_API_URL}") + print("Backend: faster (voice cloning, reference configured on server)") + print(f"Voice: {self.faster_voice or config.FASTER_TTS_VOICE}") + else: + api_url = (config.VOICE_CLONE_API_URL if self.voice_mode == config.VOICE_MODE_CLONE + else config.QWEN_API_URL) + print(f"Qwen API endpoint: {api_url}") + print(f"Voice mode: {self.voice_mode}") + print("Model size: 1.7B (always)") + if self.voice_mode == config.VOICE_MODE_CUSTOM: + print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}") + print(f"Language: {self.language}") + elif self.voice_mode == config.VOICE_MODE_CLONE: + print(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") + print(f"Language: {self.language}") print(f"Output format: {self.output_format}") if self.single_file and self.output_format != "m4b": print("Chapter mode: single file (--single-file)") @@ -396,6 +412,10 @@ class AudiobookConverter: print(f"Playback speed: {self.speed:g}x") print("=" * 70) + def run(self) -> bool: + """Main conversion process. Returns True if all books converted.""" + self._print_banner() + # Check for books book_files = sorted( f for f in config.BOOKS_FOLDER.iterdir() diff --git a/converter/tts.py b/converter/tts.py index b6049a1..cb09936 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -1,16 +1,27 @@ -"""Client wrapper for the Qwen3-TTS Gradio demos (custom voice / voice clone).""" +"""Client wrappers for the TTS backends. + +QwenTTSClient talks to the Qwen3-TTS Gradio demos (custom voice / voice clone). +FasterTTSClient talks to the OpenAI-compatible server from the +faster-qwen3-tts repository (voice cloning only; the reference voice is +configured server-side — see the "Faster backend" section of the README). +""" import contextlib import io +import json import logging import shutil import sys import threading import time +import urllib.error +import urllib.request +import wave from pathlib import Path -from typing import Any, Dict, Optional, Tuple +from typing import Any, Dict, List, Optional, Tuple from . import config +from .chunking import split_into_chunks logger = logging.getLogger(__name__) @@ -47,7 +58,105 @@ def normalize_language(value: Optional[str]) -> str: ) -class QwenTTSClient: +def transcribe_reference_audio(audio_path: str, model_name: str = "base") -> Optional[str]: + """Transcribe reference audio locally using an optional Whisper backend. + + The current qwen-tts demo does not expose a transcription endpoint, so + transcription is done client-side when a Whisper package is available. + Returns None if no backend is installed. + """ + for backend in ("faster_whisper", "whisper"): + try: + if backend == "faster_whisper": + from faster_whisper import WhisperModel + model = WhisperModel(model_name, device="cpu", compute_type="int8") + segments, _ = model.transcribe(audio_path) + text = " ".join(seg.text.strip() for seg in segments).strip() + else: + import whisper + model = whisper.load_model(model_name) + result = model.transcribe(audio_path) + text = (result.get("text") or "").strip() + if text: + logger.info("Transcription complete via %s: %s", backend, text) + return text + except ImportError: + continue + except Exception as exc: + logger.warning("%s transcription failed: %s", backend, exc) + logger.warning("No Whisper backend available; transcription skipped.") + return None + + +class _BaseTTSClient: + """Shared chunk retry logic, heartbeat, and chunk file bookkeeping.""" + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder.""" + raise NotImplementedError + + def _chunk_path(self, chunk_num: int, suffix: str) -> Path: + """Resolve the target path for a chunk, removing stale files first. + + Any stale chunk file for this index is removed so a retry or extension + change can never leave two files matching chunk_NNNN.*. + """ + for stale in config.CHUNKS_FOLDER.glob(f"chunk_{chunk_num:04d}.*"): + try: + stale.unlink() + except OSError as exc: + logger.debug("Could not remove stale chunk file %s: %s", stale, exc) + return config.CHUNKS_FOLDER / f"chunk_{chunk_num:04d}{suffix}" + + def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]: + """Process a chunk with retry logic and rate limiting. + + Returns the generated chunk file's path, or None when all attempts + failed. + """ + # Small delay between chunks to avoid rate limiting (only if not first chunk) + if chunk_num > 1: + time.sleep(config.MIN_DELAY_BETWEEN_CHUNKS) + + for attempt in range(config.MAX_RETRIES): + try: + result = self.generate_chunk(text, chunk_num) + if result and Path(result).exists(): + return Path(result) + logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) + except Exception as exc: + logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) + + if attempt < config.MAX_RETRIES - 1: + sleep_time = 5 + (2 ** attempt) + logger.info("Waiting %ds before retry...", sleep_time) + time.sleep(sleep_time) + + logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) + return None + + @contextlib.contextmanager + def _chunk_heartbeat(self, chunk_num: int): + """Print a periodic "still working" message while a chunk generates.""" + stop = threading.Event() + + def _beat(): + start = time.time() + while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): + elapsed = time.time() - start + print(f"[...] Chunk {chunk_num} still generating — " + f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True) + + thread = threading.Thread(target=_beat, daemon=True) + thread.start() + try: + yield + finally: + stop.set() + thread.join() + + +class QwenTTSClient(_BaseTTSClient): """Generates audio chunks through a Qwen3-TTS Gradio server.""" def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, @@ -166,33 +275,8 @@ class QwenTTSClient: # ------------------------------------------------------------------ def transcribe_audio(self, audio_path: str) -> Optional[str]: - """Transcribe reference audio locally using an optional Whisper backend. - - The current qwen-tts demo does not expose a transcription endpoint, so - transcription is done client-side when a Whisper package is available. - Returns None if no backend is installed. - """ - for backend in ("faster_whisper", "whisper"): - try: - if backend == "faster_whisper": - from faster_whisper import WhisperModel - model = WhisperModel("base", device="cpu", compute_type="int8") - segments, _ = model.transcribe(audio_path) - text = " ".join(seg.text.strip() for seg in segments).strip() - else: - import whisper - model = whisper.load_model("base") - result = model.transcribe(audio_path) - text = (result.get("text") or "").strip() - if text: - logger.info("Transcription complete via %s: %s", backend, text) - return text - except ImportError: - continue - except Exception as exc: - logger.warning("%s transcription failed: %s", backend, exc) - logger.warning("No Whisper backend available; transcription skipped.") - return None + """Transcribe reference audio locally using an optional Whisper backend.""" + return transcribe_reference_audio(audio_path) # ------------------------------------------------------------------ # Chunk generation @@ -222,14 +306,7 @@ class QwenTTSClient: raise RuntimeError(f"Generated audio file not found: {audio_path}") suffix = source.suffix or ".wav" - # Remove any stale chunk file for this index first so a retry or - # extension change can never leave two files matching chunk_NNNN.* - for stale in config.CHUNKS_FOLDER.glob(f"chunk_{chunk_num:04d}.*"): - try: - stale.unlink() - except OSError as exc: - logger.debug("Could not remove stale chunk file %s: %s", stale, exc) - output_path = config.CHUNKS_FOLDER / f"chunk_{chunk_num:04d}{suffix}" + output_path = self._chunk_path(chunk_num, suffix) shutil.copy2(source, output_path) logger.debug("Chunk %d generated successfully", chunk_num) @@ -239,53 +316,6 @@ class QwenTTSClient: logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) return None - def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]: - """Process a chunk with retry logic and rate limiting. - - Returns the generated chunk file's path, or None when all attempts - failed. - """ - # Small delay between chunks to avoid rate limiting (only if not first chunk) - if chunk_num > 1: - time.sleep(config.MIN_DELAY_BETWEEN_CHUNKS) - - for attempt in range(config.MAX_RETRIES): - try: - result = self.generate_chunk(text, chunk_num) - if result and Path(result).exists(): - return Path(result) - logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) - except Exception as exc: - logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) - - if attempt < config.MAX_RETRIES - 1: - sleep_time = 5 + (2 ** attempt) - logger.info("Waiting %ds before retry...", sleep_time) - time.sleep(sleep_time) - - logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) - return None - - @contextlib.contextmanager - def _chunk_heartbeat(self, chunk_num: int): - """Print a periodic "still working" message while a chunk generates.""" - stop = threading.Event() - - def _beat(): - start = time.time() - while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): - elapsed = time.time() - start - print(f"[...] Chunk {chunk_num} still generating — " - f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True) - - thread = threading.Thread(target=_beat, daemon=True) - thread.start() - try: - yield - finally: - stop.set() - thread.join() - # ------------------------------------------------------------------ # API payloads # ------------------------------------------------------------------ @@ -363,3 +393,117 @@ class QwenTTSClient: payload[name] = value return self.clone_client.predict(**payload, api_name=clone_api) + + +class FasterTTSClient(_BaseTTSClient): + """Generates audio chunks through a faster-qwen3-tts server. + + Talks to the OpenAI-compatible server shipped in the faster-qwen3-tts + repository (examples/openai_server.py). The reference voice (ref audio, + ref text) and language are configured on the server itself via + --ref-audio/--ref-text or a --voices JSON file; this client only sends + text. Unlike the Gradio demo, the server performs one generation per + request, so long chunks are sub-chunked client-side. + """ + + def __init__(self, voice: Optional[str] = None, api_url: Optional[str] = None): + self.voice = voice or config.FASTER_TTS_VOICE + self.api_url = (api_url or config.FASTER_TTS_API_URL).rstrip("/") + self._check_health() + + def _check_health(self) -> None: + """Verify the server is reachable and its model is loaded.""" + url = f"{self.api_url}/health" + try: + with urllib.request.urlopen(url, timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + except Exception as exc: + raise RuntimeError( + f"Faster TTS server not reachable at {url}: {exc}. " + "Start the faster-qwen3-tts OpenAI-compatible server first " + "(see the 'Faster backend' section of the README)." + ) from exc + if not payload.get("model_loaded"): + raise RuntimeError( + "The faster TTS server is running but its model is not loaded yet; " + "wait for model download and startup to finish, then retry." + ) + print(f"[OK] Connected to faster TTS API at {self.api_url} (voice '{self.voice}')") + print(f"[INFO] The server silently falls back to its first configured voice if " + f"'{self.voice}' is not defined in its voice config (see README).") + + # ------------------------------------------------------------------ + # HTTP requests + # ------------------------------------------------------------------ + + def _request_pcm(self, text: str) -> bytes: + """POST one sub-chunk and return raw 16-bit mono PCM bytes.""" + url = f"{self.api_url}/v1/audio/speech" + payload = json.dumps({ + "model": "tts-1", + "input": text, + "voice": self.voice, + "response_format": "pcm", + }).encode("utf-8") + request = urllib.request.Request( + url, data=payload, headers={"Content-Type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(request, timeout=config.FASTER_HTTP_TIMEOUT) as response: + pcm = response.read() + except urllib.error.HTTPError as exc: + detail = "" + try: + detail = exc.read().decode("utf-8", errors="replace")[:200] + except Exception: + pass + raise RuntimeError(f"Faster TTS server returned HTTP {exc.code}: {detail}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"Faster TTS request failed: {exc.reason}") from exc + if not pcm: + raise RuntimeError("Faster TTS server returned empty audio") + return pcm + + def _request_pcm_with_retry(self, text: str, chunk_num: int, sub_num: int, + sub_total: int) -> bytes: + """Request one sub-chunk, retrying transient failures.""" + for attempt in range(config.FASTER_SUBCHUNK_RETRIES): + try: + return self._request_pcm(text) + except Exception as exc: + logger.warning("Chunk %d sub-chunk %d/%d attempt %d failed: %s", + chunk_num, sub_num, sub_total, attempt + 1, exc) + if attempt < config.FASTER_SUBCHUNK_RETRIES - 1: + time.sleep(2 + 2 * attempt) + raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after " + f"{config.FASTER_SUBCHUNK_RETRIES} attempts") + + # ------------------------------------------------------------------ + # Chunk generation + # ------------------------------------------------------------------ + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder.""" + try: + sub_chunks = split_into_chunks(text, max_words=config.FASTER_SUBCHUNK_WORDS) + if not sub_chunks: + raise RuntimeError("No text to synthesize") + + pcm_parts: List[bytes] = [] + with self._chunk_heartbeat(chunk_num): + for sub_num, sub_text in enumerate(sub_chunks, 1): + pcm_parts.append(self._request_pcm_with_retry( + sub_text, chunk_num, sub_num, len(sub_chunks))) + + output_path = self._chunk_path(chunk_num, ".wav") + with wave.open(str(output_path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(config.FASTER_TTS_SAMPLE_RATE) + wav_file.writeframes(b"".join(pcm_parts)) + + logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks)) + return str(output_path) + + except Exception as exc: + logger.error("Faster chunk processing failed for chunk %d: %s", chunk_num, exc) + return None diff --git a/tests/test_converter.py b/tests/test_converter.py index ae307a2..48dc162 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -114,6 +114,8 @@ class NarratorTagTests(unittest.TestCase): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio + converter.faster = False + converter.faster_voice = None return converter def test_custom_voice_uses_speaker_display_name(self): @@ -189,6 +191,8 @@ class RunOverwritePromptTests(unittest.TestCase): self.converter = AudiobookConverter.__new__(AudiobookConverter) self.converter.voice_mode = config.VOICE_MODE_CUSTOM self.converter.voice_clone_ref_audio = None + self.converter.faster = False + self.converter.faster_voice = None self.converter.speed = 1.0 self.converter.single_file = False self.converter.output_format = "mp3" diff --git a/tests/test_make_voices.py b/tests/test_make_voices.py new file mode 100644 index 0000000..cf76a57 --- /dev/null +++ b/tests/test_make_voices.py @@ -0,0 +1,169 @@ +"""Tests for the voices.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from tools import make_voices + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_voices.find_wav_files(self.folder), []) + + +class BuildVoicesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_voices_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_voices, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": transcripts[path]): + voices = make_voices.build_voices([self.narrator, self.other], + "English", "base") + self.assertEqual(list(voices), ["narrator", "other"]) + self.assertEqual(voices["narrator"]["ref_text"], "First transcript.") + self.assertEqual(voices["narrator"]["language"], "English") + self.assertTrue(Path(voices["narrator"]["ref_audio"]).is_absolute()) + self.assertEqual(Path(voices["narrator"]["ref_audio"]), self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value=None): + voices = make_voices.build_voices([self.narrator], "English", "base") + self.assertEqual(voices["narrator"]["ref_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_voices.build_voices([self.narrator], "English", "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + self.output = self.folder / "voices.json" + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv): + with patch.object(sys, "argv", ["make_voices.py"] + argv): + return make_voices.main() + + def test_writes_json_with_alphabetical_voice_order(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + self.assertEqual(data["alpha"]["ref_text"], "hello") + self.assertEqual(data["alpha"]["language"], "English") + + def test_custom_output_path(self): + custom = Path(self._tmp.name) / "custom.json" + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + self._run([str(self.folder), "--output", str(custom)]) + self.assertTrue(custom.exists()) + self.assertFalse(self.output.exists()) + + def test_invalid_language_errors_before_work(self): + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder), "--language", "klingon"]) + self.assertEqual(ctx.exception.code, 2) + mock_transcribe.assert_not_called() + + def test_missing_input_dir_errors(self): + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder / "nope")]) + self.assertEqual(ctx.exception.code, 2) + + def test_no_wav_files_errors(self): + empty = Path(tempfile.mkdtemp()) + try: + with self.assertRaises(SystemExit) as ctx: + self._run([str(empty)]) + self.assertEqual(ctx.exception.code, 2) + finally: + empty.rmdir() + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ + patch("builtins.input", return_value="n"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 1) + mock_transcribe.assert_not_called() + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_existing_output_accepted_overwrites(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", return_value="y"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + def test_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", side_effect=AssertionError("prompted")): + exit_code = self._run([str(self.folder), "--force"]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tts.py b/tests/test_tts.py index b605daa..dfeda6f 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -1,12 +1,15 @@ -"""Tests for the Qwen TTS client wrapper (language handling and payloads).""" +"""Tests for the TTS client wrappers (language handling and payloads).""" +import json import tempfile import unittest +import wave from pathlib import Path from unittest.mock import MagicMock, patch from converter import config -from converter.tts import QwenTTSClient, normalize_language +from converter.converter import AudiobookConverter +from converter.tts import FasterTTSClient, QwenTTSClient, normalize_language class NormalizeLanguageTests(unittest.TestCase): @@ -151,5 +154,218 @@ class PayloadLanguageTests(unittest.TestCase): self.assertNotIn("max_chunk_chars", kwargs) +class FasterTTSClientHealthTests(unittest.TestCase): + """Connection behavior of the faster-qwen3-tts client.""" + + def _health_response(self, model_loaded=True): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps( + {"status": "ok", "model_loaded": model_loaded}).encode("utf-8") + return response + + def test_unreachable_server_raises_with_readme_pointer(self): + import urllib.error + with patch("converter.tts.urllib.request.urlopen", + side_effect=urllib.error.URLError("Connection refused")): + with self.assertRaises(RuntimeError) as ctx: + FasterTTSClient() + message = str(ctx.exception) + self.assertIn("not reachable", message) + self.assertIn("README", message) + + def test_model_not_loaded_raises(self): + with patch("converter.tts.urllib.request.urlopen", + return_value=self._health_response(model_loaded=False)): + with self.assertRaises(RuntimeError) as ctx: + FasterTTSClient() + self.assertIn("not loaded", str(ctx.exception)) + + def test_healthy_server_defaults_from_config(self): + with patch("converter.tts.urllib.request.urlopen", + return_value=self._health_response()): + client = FasterTTSClient() + self.assertEqual(client.voice, config.FASTER_TTS_VOICE) + self.assertEqual(client.api_url, config.FASTER_TTS_API_URL.rstrip("/")) + + def test_explicit_voice_and_url_override_config(self): + with patch("converter.tts.urllib.request.urlopen", + return_value=self._health_response()): + client = FasterTTSClient(voice="narrator", api_url="http://10.0.0.5:9000/") + self.assertEqual(client.voice, "narrator") + self.assertEqual(client.api_url, "http://10.0.0.5:9000") + + +class FasterTTSClientGenerateTests(unittest.TestCase): + """Chunk generation: sub-chunking, WAV output, retries, bookkeeping.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(config, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + self._sleep = patch("converter.tts.time.sleep") + self._sleep.start() + + def tearDown(self): + self._sleep.stop() + self._chunks.stop() + self._tmp.cleanup() + + def _make_client(self): + client = FasterTTSClient.__new__(FasterTTSClient) + client.voice = "default" + client.api_url = "http://127.0.0.1:8000" + return client + + def _read_wav(self, path): + with wave.open(str(path), "rb") as wav_file: + return (wav_file.getnchannels(), wav_file.getsampwidth(), + wav_file.getframerate(), wav_file.readframes(wav_file.getnframes())) + + def test_generate_chunk_writes_valid_wav(self): + client = self._make_client() + pcm = b"\x01\x00" * 100 + with patch.object(client, "_request_pcm", return_value=pcm): + result = client.generate_chunk("Hello world.", 1) + self.assertIsNotNone(result) + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + channels, sampwidth, framerate, frames = self._read_wav(path) + self.assertEqual(channels, 1) + self.assertEqual(sampwidth, 2) + self.assertEqual(framerate, config.FASTER_TTS_SAMPLE_RATE) + self.assertEqual(frames, pcm) + + def test_long_text_is_subchunked_and_concatenated_in_order(self): + client = self._make_client() + sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)] + text = " ".join(sentences) + pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30] + with patch.object(config, "FASTER_SUBCHUNK_WORDS", 10), \ + patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm: + result = client.generate_chunk(text, 1) + self.assertEqual(mock_pcm.call_count, 3) + _, _, _, frames = self._read_wav(Path(result)) + self.assertEqual(frames, b"".join(pcm_parts)) + + def test_stale_chunk_files_are_removed(self): + stale = Path(self._tmp.name) / "chunk_0001.mp3" + stale.write_bytes(b"old") + client = self._make_client() + with patch.object(client, "_request_pcm", return_value=b"\x01\x00"): + client.generate_chunk("Hello.", 1) + remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*")) + self.assertEqual(remaining, ["chunk_0001.wav"]) + + def test_transient_failure_is_retried(self): + client = self._make_client() + pcm = b"\x01\x00" * 10 + with patch.object(client, "_request_pcm", + side_effect=[RuntimeError("boom"), pcm]) as mock_pcm: + result = client.generate_chunk("Hello.", 1) + self.assertIsNotNone(result) + self.assertEqual(mock_pcm.call_count, 2) + + def test_empty_pcm_response_is_treated_as_failure(self): + client = self._make_client() + pcm = b"\x01\x00" * 10 + + def _response(body): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = body + return response + + with patch("converter.tts.urllib.request.urlopen", + side_effect=[_response(b""), _response(pcm)]) as mock_urlopen: + result = client.generate_chunk("Hello.", 1) + self.assertIsNotNone(result) + self.assertEqual(mock_urlopen.call_count, 2) + _, _, _, frames = self._read_wav(Path(result)) + self.assertEqual(frames, pcm) + + def test_exhausted_subchunk_retries_fail_the_chunk(self): + client = self._make_client() + with patch.object(client, "_request_pcm", + side_effect=RuntimeError("down")) as mock_pcm: + result = client.generate_chunk("Hello.", 1) + self.assertIsNone(result) + self.assertEqual(mock_pcm.call_count, config.FASTER_SUBCHUNK_RETRIES) + + def test_empty_text_fails_the_chunk(self): + client = self._make_client() + with patch.object(client, "_request_pcm") as mock_pcm: + result = client.generate_chunk(" ", 1) + self.assertIsNone(result) + mock_pcm.assert_not_called() + + def test_request_payload_includes_voice_text_and_format(self): + client = self._make_client() + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = b"\x01\x00" * 10 + with patch("converter.tts.urllib.request.urlopen", + return_value=response) as mock_urlopen: + pcm = client._request_pcm("Hello world.") + self.assertEqual(pcm, b"\x01\x00" * 10) + request = mock_urlopen.call_args[0][0] + self.assertEqual(request.full_url, "http://127.0.0.1:8000/v1/audio/speech") + payload = json.loads(request.data.decode("utf-8")) + self.assertEqual(payload["input"], "Hello world.") + self.assertEqual(payload["voice"], "default") + self.assertEqual(payload["response_format"], "pcm") + + +class FasterModeWiringTests(unittest.TestCase): + """AudiobookConverter wiring for the --faster backend.""" + + def test_faster_mode_uses_faster_client_without_reference(self): + with patch("converter.converter.FasterTTSClient") as mock_faster, \ + patch("converter.converter.QwenTTSClient") as mock_qwen: + AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE, + faster=True, faster_voice="narrator") + mock_faster.assert_called_once_with(voice="narrator") + mock_qwen.assert_not_called() + + def test_non_faster_clone_mode_still_requires_reference(self): + with patch("converter.converter.QwenTTSClient"): + with self.assertRaises(ValueError): + AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE) + + def test_faster_mode_still_validates_other_settings(self): + with patch("converter.converter.FasterTTSClient"): + with self.assertRaises(ValueError): + AudiobookConverter(faster=True, speed=0) + with self.assertRaises(ValueError): + AudiobookConverter(faster=True, language="klingon") + + def _faster_converter(self, faster_voice=None): + with patch("converter.converter.FasterTTSClient"): + return AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE, + faster=True, faster_voice=faster_voice) + + def test_narrator_tag_uses_faster_voice_name(self): + converter = self._faster_converter(faster_voice="male_richard_poe") + self.assertEqual(converter._narrator_tag(), "male_richard_poe") + + def test_narrator_tag_falls_back_to_config_voice(self): + converter = self._faster_converter() + self.assertEqual(converter._narrator_tag(), config.FASTER_TTS_VOICE) + + def test_banner_and_narrator_work_without_reference_audio(self): + converter = self._faster_converter(faster_voice="male_richard_poe") + converter._print_banner() # must not raise (regression: Path(None)) + self.assertIsNone(converter.voice_clone_ref_audio) + + def test_non_faster_narrator_tag_unchanged(self): + with tempfile.TemporaryDirectory() as tmp: + ref = Path(tmp) / "ref.wav" + ref.write_bytes(b"x") + with patch("converter.converter.QwenTTSClient"): + converter = AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE, + voice_clone_ref_audio=str(ref)) + self.assertEqual(converter._narrator_tag(), "ref") + + if __name__ == "__main__": unittest.main() diff --git a/tools/make_voices.py b/tools/make_voices.py new file mode 100755 index 0000000..913c454 --- /dev/null +++ b/tools/make_voices.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""Generate a voices.json file for the faster-qwen3-tts server. + +Scans a directory for .wav files, transcribes each with a local Whisper +backend (faster_whisper or whisper), and writes a voices.json + +Usage: + python tools/make_voices.py INPUT_DIR [--output PATH] [--language LANG] + [--whisper-model NAME] [--force] + +The output can be passed to the faster server: + python examples/openai_server.py --voices voices.json --port 8000 +""" + +import argparse +import json +import sys +from pathlib import Path + +# Allow running from any working directory. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from converter.tts import normalize_language, transcribe_reference_audio + + +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 build_voices(wav_files: list, language: str, whisper_model: str) -> dict: + """Transcribe each wav file and build the voices mapping.""" + voices = {} + 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}'; the faster backend " + "strongly recommends an accurate transcript — consider editing " + "voices.json by hand before starting the server") + voices[name] = { + "ref_audio": str(wav_file.resolve()), + "ref_text": text or "", + "language": language, + } + return voices + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Generate a voices.json for the faster-qwen3-tts server " + "from a directory of .wav reference files.") + parser.add_argument("input_dir", type=Path, + help="Directory containing .wav reference audio files") + parser.add_argument("--output", type=Path, default=None, + help="Output path for voices.json " + "(default: INPUT_DIR/voices.json)") + parser.add_argument("--language", type=str, default="English", + help="Language for all voices, as passed to the TTS model " + "(default: English; names and short codes accepted)") + 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 without prompting") + args = parser.parse_args() + + try: + language = normalize_language(args.language) + except ValueError as exc: + parser.error(str(exc)) + + if not args.input_dir.is_dir(): + parser.error(f"Input directory not found: {args.input_dir}") + + wav_files = find_wav_files(args.input_dir) + if not wav_files: + parser.error(f"No .wav files found in {args.input_dir}") + + output_path = args.output if args.output is not None \ + else args.input_dir / "voices.json" + if output_path.exists() and not args.force and not prompt_overwrite(output_path): + print("[INFO] Aborted; existing voices.json kept") + return 1 + + voices = build_voices(wav_files, language, args.whisper_model) + + with output_path.open("w", encoding="utf-8") as handle: + json.dump(voices, handle, indent=4, ensure_ascii=False) + handle.write("\n") + + print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " + f"{', '.join(voices)}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) |
