aboutsummaryrefslogtreecommitdiff
path: root/converter
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-19 03:55:13 -0400
committerhistoria <historiavg@proton.me>2026-08-19 03:55:13 -0400
commit87e5216cd287f411b2ffab04dbc435f48c1d4aae (patch)
treed24346d6793233a2af6305f8463ae17765e00d44 /converter
parentf1f8e899c46de80d7f9fbb8f0b53983e18167bd2 (diff)
downloadtts-audiobook-generator-87e5216cd287f411b2ffab04dbc435f48c1d4aae.tar.gz
refactor: config.py simplified
Diffstat (limited to 'converter')
-rw-r--r--converter/audio.py14
-rw-r--r--converter/config.py166
-rw-r--r--converter/converter.py87
-rw-r--r--converter/tts.py118
4 files changed, 175 insertions, 210 deletions
diff --git a/converter/audio.py b/converter/audio.py
index c9feeb8..6e5a910 100644
--- a/converter/audio.py
+++ b/converter/audio.py
@@ -12,6 +12,8 @@ from . import config
logger = logging.getLogger(__name__)
+CHUNKS_FOLDER = Path(__file__).resolve().parent.parent / "chunks"
+
def atempo_filters(speed: float) -> str:
"""Return a comma-joined ffmpeg ``atempo`` filter chain for ``speed``.
@@ -272,7 +274,7 @@ def _collect_chunk_files(total_chunks: int,
else:
missing.append(i)
continue
- matches = sorted(config.CHUNKS_FOLDER.glob(f"chunk_{i:04d}.*"))
+ matches = sorted(CHUNKS_FOLDER.glob(f"chunk_{i:04d}.*"))
if matches:
chunk_files.append(matches[0])
else:
@@ -311,7 +313,7 @@ def combine_chunks(total_chunks: int, output_path: Path,
if missing_chunks:
logger.warning("Missing chunks: %s", missing_chunks)
- concat_list = config.CHUNKS_FOLDER / "_concat_list.txt"
+ concat_list = CHUNKS_FOLDER / "_concat_list.txt"
try:
with open(concat_list, "w", encoding="utf-8") as list_file:
for chunk_file in chunk_files:
@@ -380,7 +382,7 @@ def cleanup_chunks() -> None:
try:
chunk_count = 0
for pattern in ("chunk_*", "chapter_*"):
- for chunk_file in config.CHUNKS_FOLDER.glob(pattern):
+ for chunk_file in CHUNKS_FOLDER.glob(pattern):
try:
if chunk_file.is_file():
chunk_file.unlink()
@@ -462,9 +464,9 @@ def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str],
logger.error("No chapter files provided")
return False
- concat_list = config.CHUNKS_FOLDER / "_concat_list.txt"
- metadata_file = config.CHUNKS_FOLDER / "_chapters.txt"
- speed_metadata_file = config.CHUNKS_FOLDER / "_chapters_speed.txt"
+ concat_list = CHUNKS_FOLDER / "_concat_list.txt"
+ metadata_file = CHUNKS_FOLDER / "_chapters.txt"
+ speed_metadata_file = CHUNKS_FOLDER / "_chapters_speed.txt"
try:
chapters = []
start_ms = 0
diff --git a/converter/config.py b/converter/config.py
index e5e5ff7..eb01462 100644
--- a/converter/config.py
+++ b/converter/config.py
@@ -1,161 +1,41 @@
"""Configuration for the audiobook converter.
-Edit these values to change the default voice and processing behavior.
-All paths are resolved relative to the project root, so the converter can
-be run from any working directory.
+Edit these values to change voices and processing behavior. Everything
+else (voice mode names, languages, speaker names, model ids, folders,
+file formats) is fixed in the code where it is used.
"""
-from pathlib import Path
+# Server endpoints. Voice clone needs the Base-model demo, which is a
+# separate server from the CustomVoice demo (that one only exposes
+# /run_instruct); the faster backend is the OpenAI-compatible server from
+# the faster-qwen3-tts repository.
+QWEN_API_URL = "http://127.0.0.1:7860" # CustomVoice demo
+VOICE_CLONE_API_URL = "http://127.0.0.1:7861" # Base-model demo
+FASTER_TTS_API_URL = "http://127.0.0.1:8000" # faster-qwen3-tts server
-# Project root (directory containing audiobook.py)
-BASE_DIR = Path(__file__).resolve().parent.parent
-
-# =============================================================================
-# VOICE MODES
-# =============================================================================
-
-VOICE_MODE_CUSTOM = "custom_voice"
-VOICE_MODE_CLONE = "voice_clone"
-VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE)
-
-# =============================================================================
-# TTS LANGUAGE SETTINGS
-# =============================================================================
-# Languages understood by the Qwen3-TTS API. The Gradio demo silently falls
-# back to "Auto" for unrecognized values, so languages are validated
-# client-side (see converter.tts.normalize_language) before reaching the API.
-# Display names must match the demo dropdown exactly.
-
-TTS_LANGUAGES = (
- "Auto",
- "Chinese",
- "English",
- "German",
- "Italian",
- "Portuguese",
- "Spanish",
- "Japanese",
- "Korean",
- "French",
- "Russian",
-)
-
-# Short aliases accepted on the command line (ISO 639-1 codes and common
-# shorthands), mapped to the display names above.
-TTS_LANGUAGE_ALIASES = {
- "zh": "Chinese",
- "en": "English",
- "de": "German",
- "it": "Italian",
- "pt": "Portuguese",
- "es": "Spanish",
- "ja": "Japanese",
- "ko": "Korean",
- "fr": "French",
- "ru": "Russian",
- "zh-cn": "Chinese",
- "zh-tw": "Chinese",
- "pt-br": "Portuguese",
- "en-us": "English",
- "en-gb": "English",
-}
-
-# =============================================================================
-# QWEN API CONFIGURATION
-# =============================================================================
+# Words per TTS generation request. Each API call is ONE model generation:
+# long generations lose prosody and can degrade into garbled audio.
+CHUNK_SIZE_WORDS = 40
+VOICE_CLONE_MAX_CHUNK_CHARS = 200 # Server-side re-chunking limit for clone requests
+VOICE_CLONE_CHUNK_GAP = 0 # Pause (seconds) between server-side clone chunks
+MIN_DELAY_BETWEEN_CHUNKS = 0 # Pause between API calls (rate-limit protection; local servers need none)
-QWEN_API_URL = "http://127.0.0.1:7860" # CustomVoice demo endpoint
API_TIMEOUT = 300 # Seconds before an API call times out
-MAX_RETRIES = 3 # Retry failed chunks
+MAX_RETRIES = 3 # Attempts per chunk request
+HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" this often during a chunk
-# =============================================================================
-# CUSTOM VOICE SETTINGS (pre-built speakers, always uses the 1.7B model)
-# =============================================================================
+LANGUAGE = "English"
+SEED = -1
CUSTOM_VOICE_SPEAKER = "Vivian"
-CUSTOM_VOICE_LANGUAGE = "English"
CUSTOM_VOICE_INSTRUCT = "Speak naturally and clearly, as if reading a dramatic book to an adult audience."
-CUSTOM_VOICE_MODEL_SIZE = "1.7B"
-CUSTOM_VOICE_SEED = -1
-CUSTOM_VOICE_MODEL_ID = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
-
-# Map canonical speaker names to the display names used by the qwen-tts Gradio demo.
-SPEAKER_DISPLAY_NAMES = {
- "ryan": "Ryan",
- "serena": "Serena",
- "vivian": "Vivian",
- "uncle_fu": "Uncle Fu",
- "aiden": "Aiden",
- "ono_anna": "Ono Anna",
- "sohee": "Sohee",
- "eric": "Eric",
- "dylan": "Dylan",
-}
-# =============================================================================
-# VOICE CLONE SETTINGS (clone a voice from a reference audio file)
-# =============================================================================
-# Voice clone requires the Base-model demo (Qwen3-TTS-12Hz-1.7B-Base), which
-# exposes /run_voice_clone. The CustomVoice demo only exposes /run_instruct, so
-# run the Base demo on a separate port and point this at it.
-
-VOICE_CLONE_LANGUAGE = "English"
VOICE_CLONE_USE_XVECTOR_ONLY = False
-VOICE_CLONE_MODEL_SIZE = "1.7B"
-VOICE_CLONE_MAX_CHUNK_CHARS = 200
-VOICE_CLONE_CHUNK_GAP = 0
-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.
+# Must match a key in the faster server's voices.json ("default" when the
+# server was launched with --ref-audio). The server silently falls back to
+# its first voice for unknown names.
FASTER_TTS_VOICE = "default"
-FASTER_TTS_SAMPLE_RATE = 24000 # Qwen3-TTS 12Hz codec output rate
-# Safety net: the faster server takes one generation per request, so any
-# chunk longer than this is sub-chunked client-side (a no-op at the default
-# CHUNK_SIZE_WORDS above; kept in case the chunk size is ever raised).
-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
-# =============================================================================
-
-BOOKS_FOLDER = BASE_DIR / "input" # Input folder
-AUDIOBOOKS_FOLDER = BASE_DIR / "output" # Output folder
-CHUNKS_FOLDER = BASE_DIR / "chunks" # Scratch space for per-chunk audio (cleaned per book)
-LOGS_FOLDER = BASE_DIR / "logs"
-DEBUG_FOLDER = BASE_DIR / "debug" # Per-chunk audio + text dumps for --debug (kept across runs)
-
-# Words per TTS generation request. Each API call is ONE model generation:
-# long generations lose prosody, can degrade into garbled audio, and text
-# past the model's token limit is never spoken. ~40 words (~200 chars) is
-# the per-request length the old qwen-tts demo enforced server-side.
-CHUNK_SIZE_WORDS = 40
-# Pause between API calls (rate-limit protection for hosted demos; a local
-# server needs no delay).
-MIN_DELAY_BETWEEN_CHUNKS = 0
-HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" this often during a chunk
-
-# =============================================================================
-# AUDIO OUTPUT SETTINGS
-# =============================================================================
AUDIO_FORMAT = "m4b" # Default output container
-AUDIO_FORMATS = ("mp3", "m4b", "ogg", "flac")
AUDIO_BITRATE = "128k"
-
-SUPPORTED_FORMATS = [".txt", ".pdf", ".epub"]
diff --git a/converter/converter.py b/converter/converter.py
index 30418e5..8371c02 100644
--- a/converter/converter.py
+++ b/converter/converter.py
@@ -14,10 +14,32 @@ from typing import Dict, List, Optional, Tuple
from . import audio, chunking, config, cover, extractors
from .audio import TrackMeta
-from .tts import FasterTTSClient, QwenTTSClient, normalize_language, speaker_display_name
+from .tts import (
+ VOICE_MODE_CLONE,
+ VOICE_MODE_CUSTOM,
+ VOICE_MODES,
+ FasterTTSClient,
+ QwenTTSClient,
+ normalize_language,
+ speaker_display_name,
+)
logger = logging.getLogger(__name__)
+# Folders, resolved from the project root so the converter runs from any
+# working directory.
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+BOOKS_FOLDER = BASE_DIR / "input"
+AUDIOBOOKS_FOLDER = BASE_DIR / "output"
+CHUNKS_FOLDER = BASE_DIR / "chunks" # Per-chunk scratch audio, cleaned per book
+LOGS_FOLDER = BASE_DIR / "logs"
+DEBUG_FOLDER = BASE_DIR / "debug" # --debug dumps, kept across runs
+
+# Output containers and supported input formats.
+AUDIO_FORMATS = ("mp3", "m4b", "ogg", "flac")
+SUPPORTED_FORMATS = [".txt", ".pdf", ".epub"]
+
def _console_log_filter(record: logging.LogRecord) -> bool:
"""Keep httpx/httpcore request logs out of the console (file only)."""
@@ -32,9 +54,9 @@ def setup_logging(debug: bool = False) -> None:
(DEBUG with --debug) so progress prints are never mirrored as
timestamped log lines; httpx/httpcore request logs stay file-only.
"""
- config.LOGS_FOLDER.mkdir(parents=True, exist_ok=True)
+ LOGS_FOLDER.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(
- config.LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log",
+ LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log",
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG if debug else logging.INFO)
@@ -52,8 +74,8 @@ def setup_logging(debug: bool = False) -> None:
def setup_directories() -> None:
"""Create necessary directories."""
- for directory in (config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER,
- config.CHUNKS_FOLDER, config.LOGS_FOLDER):
+ for directory in (BOOKS_FOLDER, AUDIOBOOKS_FOLDER,
+ CHUNKS_FOLDER, LOGS_FOLDER):
Path(directory).mkdir(parents=True, exist_ok=True)
@@ -64,7 +86,7 @@ def find_existing_outputs(output_name: str, output_format: str) -> List[Path]:
named ``{name}_suffix.{ext}``; exact chapter file names are only known
after text extraction, so any file matching that pattern counts.
"""
- folder = config.AUDIOBOOKS_FOLDER
+ folder = AUDIOBOOKS_FOLDER
existing: List[Path] = []
primary = folder / f"{output_name}.{output_format}"
if primary.exists():
@@ -102,18 +124,17 @@ def prompt_overwrite(existing: List[Path], output_name: str) -> bool:
class AudiobookConverter:
"""Audiobook converter using the Qwen TTS API."""
- def __init__(self, voice_mode: str = config.VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None,
+ def __init__(self, voice_mode: str = 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, faster: bool = False,
faster_voice: Optional[str] = None, debug: bool = False):
if speed <= 0:
raise ValueError(f"Speed must be a positive number, got {speed}")
- if output_format not in config.AUDIO_FORMATS:
+ if output_format not in AUDIO_FORMATS:
raise ValueError(f"Unsupported output format: {output_format}")
if language is None:
- language = (config.VOICE_CLONE_LANGUAGE if voice_mode == config.VOICE_MODE_CLONE
- else config.CUSTOM_VOICE_LANGUAGE)
+ language = config.LANGUAGE
self.language = normalize_language(language)
self.voice_mode = voice_mode
self.voice_clone_ref_audio = voice_clone_ref_audio
@@ -139,12 +160,12 @@ class AudiobookConverter:
def _validate_configuration(self) -> None:
"""Validate configuration settings."""
- if self.voice_mode not in config.VOICE_MODES:
+ if self.voice_mode not in VOICE_MODES:
raise ValueError(
f"Unknown voice mode: {self.voice_mode!r} "
- f"(expected one of {config.VOICE_MODES})"
+ f"(expected one of {VOICE_MODES})"
)
- if self.voice_mode == config.VOICE_MODE_CLONE and not self.faster:
+ if self.voice_mode == VOICE_MODE_CLONE and not self.faster:
if not self.voice_clone_ref_audio:
raise ValueError(
"Voice Clone mode requires a reference audio file. "
@@ -173,7 +194,7 @@ class AudiobookConverter:
"""
if self.faster:
narrator = self.faster_voice or config.FASTER_TTS_VOICE
- elif self.voice_mode == config.VOICE_MODE_CLONE:
+ elif self.voice_mode == VOICE_MODE_CLONE:
narrator = Path(self.voice_clone_ref_audio).stem
else:
narrator = speaker_display_name()
@@ -243,13 +264,13 @@ class AudiobookConverter:
stem = output_name or f"{file_path.stem}_{self._narrator_tag()}"
# --debug: chunk text/audio dumps land in a per-book folder
- debug_dir = config.DEBUG_FOLDER / stem if self.debug else None
+ debug_dir = DEBUG_FOLDER / stem if self.debug else None
# Cover art: generated once per book. Named with the chunk_
# prefix so cleanup_chunks() removes it with the other scratch
# files at the end of the book.
cover_path = cover.generate_cover(
- book.title, config.CHUNKS_FOLDER / "chunk_cover.png")
+ book.title, CHUNKS_FOLDER / "chunk_cover.png")
if cover_path:
print(f"[INFO] Generated cover art for '{book.title}'")
meta = TrackMeta(title=book.title, artist=book.author, album=book.title)
@@ -261,20 +282,20 @@ class AudiobookConverter:
return self._convert_m4b_with_chapters(sections, stem, start_time,
meta=meta, cover=cover_path,
debug_dir=debug_dir)
- output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
+ output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
return self._convert_text(sections[0].text, output_path, start_time,
meta=meta, cover=cover_path, debug_dir=debug_dir)
if self.single_file or len(sections) == 1:
text = "\n\n".join(section.text for section in sections)
- output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
+ output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
return self._convert_text(text, output_path, start_time,
meta=meta, cover=cover_path, debug_dir=debug_dir)
success = True
for index, section in enumerate(sections, 1):
chapter_name = f"{stem}_{index:02d}_{self._sanitize_filename(section.title)}"
- output_path = config.AUDIOBOOKS_FOLDER / f"{chapter_name}.{self.output_format}"
+ output_path = AUDIOBOOKS_FOLDER / f"{chapter_name}.{self.output_format}"
track_meta = meta._replace(
title=(section.title or "").strip() or f"Chapter {index}",
track=index, total_tracks=len(sections))
@@ -309,7 +330,7 @@ class AudiobookConverter:
titles = []
total_chapters = len(sections)
for index, section in enumerate(sections, 1):
- chapter_path = config.CHUNKS_FOLDER / f"chapter_{index:04d}.wav"
+ chapter_path = CHUNKS_FOLDER / f"chapter_{index:04d}.wav"
title = (section.title or "").strip() or f"Chapter {index}"
print(f"\n{'=' * 50}")
print(f"CHAPTER {index}/{total_chapters}: {title}")
@@ -329,7 +350,7 @@ class AudiobookConverter:
logger.error("No chapters were successfully converted")
return False
- output_path = config.AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
+ output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}"
if not audio.combine_chapters_to_m4b(chapter_files, titles, output_path, speed=self.speed,
meta=meta, cover=cover):
return False
@@ -472,22 +493,22 @@ class AudiobookConverter:
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"Books folder: {BOOKS_FOLDER}")
+ print(f"Output folder: {AUDIOBOOKS_FOLDER}")
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
+ api_url = (config.VOICE_CLONE_API_URL if self.voice_mode == 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:
+ if self.voice_mode == VOICE_MODE_CUSTOM:
print(f"Speaker: {config.CUSTOM_VOICE_SPEAKER}")
print(f"Language: {self.language}")
- elif self.voice_mode == config.VOICE_MODE_CLONE:
+ elif self.voice_mode == 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}")
@@ -496,7 +517,7 @@ class AudiobookConverter:
if abs(self.speed - 1.0) >= 1e-6:
print(f"Playback speed: {self.speed:g}x")
if self.debug:
- print(f"Debug dumps (per-chunk text + raw audio): {config.DEBUG_FOLDER}")
+ print(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}")
print("=" * 70)
def run(self) -> bool:
@@ -506,16 +527,16 @@ class AudiobookConverter:
# Check for books
book_files = sorted(
- f for f in config.BOOKS_FOLDER.iterdir()
- if f.is_file() and f.suffix.lower() in config.SUPPORTED_FORMATS
+ f for f in BOOKS_FOLDER.iterdir()
+ if f.is_file() and f.suffix.lower() in SUPPORTED_FORMATS
)
if not book_files:
- print(f"[INFO] No supported files found in {config.BOOKS_FOLDER}")
- print(f"Supported formats: {', '.join(config.SUPPORTED_FORMATS)}")
+ print(f"[INFO] No supported files found in {BOOKS_FOLDER}")
+ print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}")
# Create sample file
- sample_file = config.BOOKS_FOLDER / "sample.txt"
+ sample_file = BOOKS_FOLDER / "sample.txt"
sample_file.write_text(
"This is a sample audiobook for testing the Qwen-based converter. "
"The system will send this text to the Qwen API for voice generation. "
@@ -579,7 +600,7 @@ class AudiobookConverter:
print(f"{status} {filename}")
if successful > 0:
- print(f"\n[INFO] Audiobooks saved to: {config.AUDIOBOOKS_FOLDER}/")
+ print(f"\n[INFO] Audiobooks saved to: {AUDIOBOOKS_FOLDER}/")
elapsed = int(time.time() - run_start)
hours, remainder = divmod(elapsed, 3600)
diff --git a/converter/tts.py b/converter/tts.py
index 07cfa65..9930558 100644
--- a/converter/tts.py
+++ b/converter/tts.py
@@ -25,18 +25,81 @@ from .chunking import split_into_chunks
logger = logging.getLogger(__name__)
+# Voice modes (re-exported for the CLI and the converter orchestrator).
+VOICE_MODE_CUSTOM = "custom_voice"
+VOICE_MODE_CLONE = "voice_clone"
+VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE)
+
+# Languages understood by the Qwen3-TTS API. Display names must match the
+# demo dropdown exactly (the demo silently falls back to "Auto" for
+# unrecognized values, so languages are validated client-side first).
+TTS_LANGUAGES = (
+ "Auto",
+ "Chinese",
+ "English",
+ "German",
+ "Italian",
+ "Portuguese",
+ "Spanish",
+ "Japanese",
+ "Korean",
+ "French",
+ "Russian",
+)
+
+# Short aliases accepted on the command line (ISO 639-1 codes and common
+# shorthands), mapped to the display names above.
+TTS_LANGUAGE_ALIASES = {
+ "zh": "Chinese",
+ "en": "English",
+ "de": "German",
+ "it": "Italian",
+ "pt": "Portuguese",
+ "es": "Spanish",
+ "ja": "Japanese",
+ "ko": "Korean",
+ "fr": "French",
+ "ru": "Russian",
+ "zh-cn": "Chinese",
+ "zh-tw": "Chinese",
+ "pt-br": "Portuguese",
+ "en-us": "English",
+ "en-gb": "English",
+}
+
+# Canonical speaker names -> display names used by the qwen-tts demo.
+SPEAKER_DISPLAY_NAMES = {
+ "ryan": "Ryan",
+ "serena": "Serena",
+ "vivian": "Vivian",
+ "uncle_fu": "Uncle Fu",
+ "aiden": "Aiden",
+ "ono_anna": "Ono Anna",
+ "sohee": "Sohee",
+ "eric": "Eric",
+ "dylan": "Dylan",
+}
+
+# Fixed model facts: both demos run the 1.7B model (the CustomVoice demo
+# takes its full HuggingFace id), and the 12Hz codec outputs 24 kHz audio.
+MODEL_SIZE = "1.7B"
+CUSTOM_VOICE_MODEL_ID = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
+SAMPLE_RATE = 24000
+
+CHUNKS_FOLDER = Path(__file__).resolve().parent.parent / "chunks"
+
def speaker_display_name() -> str:
"""Return the Gradio display name for the configured custom speaker."""
- return config.SPEAKER_DISPLAY_NAMES.get(
+ return SPEAKER_DISPLAY_NAMES.get(
config.CUSTOM_VOICE_SPEAKER.lower(), config.CUSTOM_VOICE_SPEAKER)
def normalize_language(value: Optional[str]) -> str:
"""Normalize a user-provided language name to a Qwen3-TTS display name.
- Accepts the display names in config.TTS_LANGUAGES case-insensitively as
- well as the short aliases in config.TTS_LANGUAGE_ALIASES (ISO 639-1 codes
+ Accepts the display names in TTS_LANGUAGES case-insensitively as
+ well as the short aliases in TTS_LANGUAGE_ALIASES (ISO 639-1 codes
and common shorthands). Raises ValueError for anything else, since the
Qwen3-TTS demo silently falls back to "Auto" for unrecognized languages.
"""
@@ -45,16 +108,16 @@ def normalize_language(value: Optional[str]) -> str:
candidate = value.strip()
if not candidate:
raise ValueError("Language must not be empty")
- for name in config.TTS_LANGUAGES:
+ for name in TTS_LANGUAGES:
if candidate.lower() == name.lower():
return name
- alias = config.TTS_LANGUAGE_ALIASES.get(candidate.lower())
+ alias = TTS_LANGUAGE_ALIASES.get(candidate.lower())
if alias:
return alias
raise ValueError(
f"Unknown language: {value!r}. Expected one of "
- f"{', '.join(config.TTS_LANGUAGES)} (or an alias: "
- f"{', '.join(sorted(config.TTS_LANGUAGE_ALIASES))})."
+ f"{', '.join(TTS_LANGUAGES)} (or an alias: "
+ f"{', '.join(sorted(TTS_LANGUAGE_ALIASES))})."
)
@@ -101,12 +164,12 @@ class _BaseTTSClient:
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}.*"):
+ for stale in 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}"
+ return 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.
@@ -162,17 +225,16 @@ class QwenTTSClient(_BaseTTSClient):
def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None,
voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False,
language: Optional[str] = None):
- if voice_mode not in config.VOICE_MODES:
+ if voice_mode not in VOICE_MODES:
raise ValueError(
- f"Unknown voice mode: {voice_mode!r} (expected one of {config.VOICE_MODES})"
+ f"Unknown voice mode: {voice_mode!r} (expected one of {VOICE_MODES})"
)
self.voice_mode = voice_mode
self.voice_clone_ref_audio = voice_clone_ref_audio
self.voice_clone_ref_text = (voice_clone_ref_text or "").strip()
self.skip_transcription = skip_transcription
if language is None:
- language = (config.VOICE_CLONE_LANGUAGE if voice_mode == config.VOICE_MODE_CLONE
- else config.CUSTOM_VOICE_LANGUAGE)
+ language = config.LANGUAGE
# Validate before connecting so bad values fail fast without a server.
self.language = normalize_language(language)
self.client = None
@@ -187,9 +249,9 @@ class QwenTTSClient(_BaseTTSClient):
# ------------------------------------------------------------------
def _connect(self) -> None:
- api_url = config.VOICE_CLONE_API_URL if self.voice_mode == config.VOICE_MODE_CLONE else config.QWEN_API_URL
+ api_url = config.VOICE_CLONE_API_URL if self.voice_mode == VOICE_MODE_CLONE else config.QWEN_API_URL
try:
- if self.voice_mode == config.VOICE_MODE_CLONE:
+ if self.voice_mode == VOICE_MODE_CLONE:
# Voice clone uses the Base-model demo, which is a separate server
# from the CustomVoice demo (that one only exposes /run_instruct).
self._init_client(config.VOICE_CLONE_API_URL, clone=True)
@@ -295,10 +357,10 @@ class QwenTTSClient(_BaseTTSClient):
def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]:
"""Generate one audio chunk; returns its path in the chunks folder."""
try:
- if self.voice_mode == config.VOICE_MODE_CUSTOM:
+ if self.voice_mode == VOICE_MODE_CUSTOM:
with self._chunk_heartbeat(chunk_num):
result = self._generate_custom_voice(text)
- elif self.voice_mode == config.VOICE_MODE_CLONE:
+ elif self.voice_mode == VOICE_MODE_CLONE:
with self._chunk_heartbeat(chunk_num):
result = self._generate_voice_clone(text)
else:
@@ -348,12 +410,12 @@ class QwenTTSClient(_BaseTTSClient):
instruct=config.CUSTOM_VOICE_INSTRUCT,
)
if self._endpoint_accepts_param(custom_api, "model_id_cv"):
- payload["model_id_cv"] = config.CUSTOM_VOICE_MODEL_ID
+ payload["model_id_cv"] = CUSTOM_VOICE_MODEL_ID
elif self._endpoint_accepts_param(custom_api, "model_size"):
- payload["model_size"] = config.CUSTOM_VOICE_MODEL_SIZE
+ payload["model_size"] = MODEL_SIZE
if self._endpoint_accepts_param(custom_api, "seed"):
- payload["seed"] = config.CUSTOM_VOICE_SEED
+ payload["seed"] = config.SEED
return self.client.predict(**payload, api_name=custom_api)
@@ -393,10 +455,10 @@ class QwenTTSClient(_BaseTTSClient):
use_xvector_only=use_xvector,
)
optional_params = {
- "model_size": config.VOICE_CLONE_MODEL_SIZE,
+ "model_size": MODEL_SIZE,
"max_chunk_chars": config.VOICE_CLONE_MAX_CHUNK_CHARS,
"chunk_gap": config.VOICE_CLONE_CHUNK_GAP,
- "seed": config.VOICE_CLONE_SEED,
+ "seed": config.SEED,
}
for name, value in optional_params.items():
if self._endpoint_accepts_param(clone_api, name, api_info=self.clone_api_info):
@@ -458,7 +520,7 @@ class FasterTTSClient(_BaseTTSClient):
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:
+ with urllib.request.urlopen(request, timeout=config.API_TIMEOUT) as response:
pcm = response.read()
except urllib.error.HTTPError as exc:
detail = ""
@@ -476,16 +538,16 @@ class FasterTTSClient(_BaseTTSClient):
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):
+ for attempt in range(config.MAX_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:
+ if attempt < config.MAX_RETRIES - 1:
time.sleep(2 + 2 * attempt)
raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after "
- f"{config.FASTER_SUBCHUNK_RETRIES} attempts")
+ f"{config.MAX_RETRIES} attempts")
# ------------------------------------------------------------------
# Chunk generation
@@ -494,7 +556,7 @@ class FasterTTSClient(_BaseTTSClient):
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)
+ sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE_WORDS)
if not sub_chunks:
raise RuntimeError("No text to synthesize")
@@ -508,7 +570,7 @@ class FasterTTSClient(_BaseTTSClient):
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.setframerate(SAMPLE_RATE)
wav_file.writeframes(b"".join(pcm_parts))
logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks))