diff options
Diffstat (limited to 'converter')
| -rw-r--r-- | converter/__init__.py | 1 | ||||
| -rw-r--r-- | converter/audio.py | 616 | ||||
| -rw-r--r-- | converter/chunking.py | 91 | ||||
| -rw-r--r-- | converter/config.py | 77 | ||||
| -rw-r--r-- | converter/converter.py | 780 | ||||
| -rw-r--r-- | converter/cover.py | 279 | ||||
| -rw-r--r-- | converter/extractors.py | 328 | ||||
| -rw-r--r-- | converter/tts.py | 1305 |
8 files changed, 0 insertions, 3477 deletions
diff --git a/converter/__init__.py b/converter/__init__.py deleted file mode 100644 index 86a827f..0000000 --- a/converter/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""TTS audiobook generator package.""" diff --git a/converter/audio.py b/converter/audio.py deleted file mode 100644 index d1dc28b..0000000 --- a/converter/audio.py +++ /dev/null @@ -1,616 +0,0 @@ -"""Audio assembly: combining chunks, speed adjustment, cleanup.""" - -import logging -import re -import shutil -import subprocess -import traceback -import wave -from pathlib import Path -from typing import Dict, List, NamedTuple, Optional, Tuple - -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``. - - ``atempo`` accepts 0.5..2.0 per filter; values outside that range are - handled by chaining multiple filters. Returns "" when ``speed`` is 1.0. - """ - if speed <= 0: - raise ValueError(f"Speed must be a positive number, got {speed}") - if abs(speed - 1.0) < 1e-6: - return "" - remaining = float(speed) - chain = [] - while remaining > 2.0: - chain.append("atempo=2.0") - remaining /= 2.0 - while remaining < 0.5: - chain.append("atempo=0.5") - remaining /= 0.5 - chain.append(f"atempo={remaining:g}") - return ",".join(chain) - - -def speed_export_params(speed: float) -> List[str]: - """Return ffmpeg filter args for pitch-preserving speed adjustment.""" - filters = atempo_filters(speed) - if not filters: - return [] - return ["-filter:a", filters] - - -def _concat_escape(path: str) -> str: - """Escape a path for use inside single quotes in an ffmpeg concat list.""" - return path.replace("'", "'\\''") - - -def _concat_wav_files(sources: List[Path], destination: Path) -> bool: - """Concatenate WAV files with matching parameters using the wave module. - - Returns False (touching nothing) when any input is not a readable WAV - or the parameters differ, so the caller can fall back to ffmpeg. - """ - opened = [] - try: - parameters = None - for source in sources: - wav_file = wave.open(str(source), "rb") - opened.append(wav_file) - current = (wav_file.getnchannels(), wav_file.getsampwidth(), - wav_file.getframerate()) - if parameters is None: - parameters = current - elif current != parameters: - return False - if parameters is None or min(parameters) < 1: - return False - with wave.open(str(destination), "wb") as output: - output.setnchannels(parameters[0]) - output.setsampwidth(parameters[1]) - output.setframerate(parameters[2]) - for wav_file in opened: - output.writeframes(wav_file.readframes(wav_file.getnframes())) - return True - except (wave.Error, EOFError, OSError): - return False - finally: - for wav_file in opened: - try: - wav_file.close() - except Exception: - pass - - -def _concat_with_ffmpeg(sources: List[Path], destination: Path) -> None: - """Concatenate audio files with ffmpeg's concat demuxer, re-encoding to - 16-bit PCM WAV (handles inputs the wave module cannot).""" - if shutil.which("ffmpeg") is None: - raise RuntimeError( - "ffmpeg is required to concatenate audio parts in non-WAV formats " - "(install ffmpeg and try again)" - ) - list_path = destination.with_name(destination.stem + "_parts.txt") - try: - with open(list_path, "w", encoding="utf-8") as list_file: - for source in sources: - list_file.write(f"file '{_concat_escape(str(source))}'\n") - command = [ - "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", - "-f", "concat", "-safe", "0", "-i", str(list_path), - "-c:a", "pcm_s16le", str(destination), - ] - proc = subprocess.run(command, capture_output=True, text=True) - if proc.returncode != 0: - raise RuntimeError( - f"ffmpeg failed to concatenate audio parts: {proc.stderr[-500:]}") - finally: - try: - list_path.unlink(missing_ok=True) - except OSError: - pass - - -def concat_audio_files(sources: List[Path], destination: Path) -> None: - """Concatenate audio files into one file at ``destination``. - - Joins the audio returned by several TTS sub-requests for a single - chunk. Uses the stdlib wave module when every input is a WAV with - matching parameters (lossless, no external tools); otherwise falls - back to ffmpeg's concat demuxer with re-encoding. - """ - if not sources: - raise ValueError("No audio files to concatenate") - if _concat_wav_files(sources, destination): - return - _concat_with_ffmpeg(sources, destination) - - -def _encode_args(output_format: str) -> List[str]: - """Return ffmpeg output codec/bitrate args for the requested container.""" - if output_format == "m4b": - return ["-c:a", "aac", "-b:a", config.AUDIO_BITRATE] - if output_format == "ogg": - return ["-c:a", "libvorbis", "-b:a", config.AUDIO_BITRATE] - if output_format == "flac": - return ["-c:a", "flac"] - if output_format == "wav": - # Lossless intermediate for per-chapter scratch audio; avoids - # generational loss when the final m4b re-encodes to AAC. - return ["-c:a", "pcm_s16le"] - return ["-b:a", config.AUDIO_BITRATE] - - -class TrackMeta(NamedTuple): - """Tags embedded into a finished audiobook file.""" - - title: str - artist: str = "" - album: str = "" - track: Optional[int] = None - total_tracks: Optional[int] = None - - -def _tag_args(meta: TrackMeta, output_format: str) -> List[str]: - """Return -metadata args plus format-specific tagging flags.""" - args = ["-metadata", f"title={meta.title}"] - if meta.artist: - args += ["-metadata", f"artist={meta.artist}"] - if meta.album: - args += ["-metadata", f"album={meta.album}"] - if meta.track and meta.total_tracks: - args += ["-metadata", f"track={meta.track}/{meta.total_tracks}"] - if output_format == "mp3": - # ID3v2.3 is what essentially every player reads; ffmpeg's default - # (v2.4) still confuses some of them. - args += ["-id3v2_version", "3"] - return args - - -# Formats whose container has no reliable embedded-picture support. -_NO_COVER_FORMATS = ("ogg", "wav") - - -def _cover_args(output_format: str, cover_input_index: int) -> List[str]: - """Return per-output args attaching a cover image as an embedded picture. - - The cover must already be added as an ffmpeg input; ``cover_input_index`` - is that input's position on the command line. mp3/flac keep the PNG - stream as-is; m4b re-encodes to JPEG, which audiobook players expect. - """ - if output_format in _NO_COVER_FORMATS: - return [] - codec = "mjpeg" if output_format == "m4b" else "copy" - args = ["-map", f"{cover_input_index}:v", "-c:v", codec, - "-disposition:v", "attached_pic", - "-metadata:s:v", "title=Album cover"] - if output_format == "m4b": - args += ["-q:v", "3"] - return args - - -_brand_supported: Optional[bool] = None - - -def _detect_brand_support() -> bool: - """Check whether the local ffmpeg muxer accepts the ``-brand`` option.""" - if shutil.which("ffmpeg") is None: - return False - try: - proc = subprocess.run( - ["ffmpeg", "-hide_banner", "-h", "muxer=ipod"], - capture_output=True, text=True, timeout=15, - ) - except (OSError, subprocess.TimeoutExpired): - return False - return "-brand" in proc.stdout - - -def _m4b_container_args() -> List[str]: - """Return per-output container flags for m4b files. - - ``+faststart`` moves the ``moov`` index to the front of the file so - streaming players (and naive linear readers like web players) can index - it; without it they may misreport the duration or refuse the file. - The ``M4B `` major brand identifies the file as an audiobook to - players that sniff brands instead of trusting the extension (the ffmpeg - default brand for .m4b is ``M4A ``). - """ - global _brand_supported - if _brand_supported is None: - _brand_supported = _detect_brand_support() - args = ["-movflags", "+faststart"] - if _brand_supported: - args += ["-brand", "M4B "] - return args - - -def build_concat_command(concat_list: Path, output_path: Path, output_format: str, - speed: float = 1.0, speed_path: Optional[Path] = None, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None) -> List[str]: - """Build the ffmpeg command that concatenates chunk audio into a book file.""" - encode = _encode_args(output_format) - container = _m4b_container_args() if output_format == "m4b" else [] - inputs = ["-f", "concat", "-safe", "0", "-i", str(concat_list)] - cover_index = None - if cover is not None and output_format not in _NO_COVER_FORMATS: - inputs += ["-i", str(cover)] - cover_index = 1 - cover_block = (_cover_args(output_format, cover_index) - if cover_index is not None else []) - tags = _tag_args(meta, output_format) if meta else [] - filters = atempo_filters(speed) - if filters: - if speed_path is None: - raise ValueError("speed_path is required when speed is not 1.0") - return [ - "ffmpeg", "-y", *inputs, "-filter_complex", - f"[0:a]split=2[base][spd];[spd]{filters}[spdout]", - "-map", "[base]", *encode, *tags, *cover_block, *container, str(output_path), - "-map", "[spdout]", *encode, *tags, *cover_block, *container, str(speed_path), - ] - output_maps = ["-map", "0:a"] if cover_block else [] - return [ - "ffmpeg", "-y", *inputs, *output_maps, - *encode, *tags, *cover_block, *container, str(output_path), - ] - - -def build_m4b_chapters_command(concat_list: Path, metadata_file: Path, output_path: Path, - speed: float = 1.0, speed_path: Optional[Path] = None, - speed_metadata_file: Optional[Path] = None, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None) -> List[str]: - """Build the ffmpeg command that assembles chapter audio into one m4b. - - Chapter metadata inputs are bound to their outputs with explicit - ``-map_chapters`` so the normal-speed and speed-adjusted copies each get - their own (correctly scaled) chapter markers. - """ - encode = _encode_args("m4b") - container = _m4b_container_args() - tags = _tag_args(meta, "m4b") if meta else [] - inputs = ["-f", "concat", "-safe", "0", "-i", str(concat_list), - "-i", str(metadata_file)] - input_count = 2 # concat audio + ffmetadata - if speed_path is not None: - inputs += ["-i", str(speed_metadata_file)] - input_count += 1 - cover_block: List[str] = [] - if cover is not None: - inputs += ["-i", str(cover)] - cover_block = _cover_args("m4b", input_count) - if speed_path is not None: - return [ - "ffmpeg", "-y", *inputs, "-filter_complex", - f"[0:a]split=2[base][spd];[spd]{atempo_filters(speed)}[spdout]", - "-map", "[base]", *cover_block, "-map_metadata", "1", "-map_chapters", "1", - *encode, *tags, *container, str(output_path), - "-map", "[spdout]", *cover_block, "-map_metadata", "2", "-map_chapters", "2", - *encode, *tags, *container, str(speed_path), - ] - return [ - "ffmpeg", "-y", *inputs, - "-map", "0:a", *cover_block, "-map_metadata", "1", "-map_chapters", "1", - *encode, *tags, *container, str(output_path), - ] - - -_DURATION_WARN_TOLERANCE = 0.05 # warn when output duration drifts >5% -_DURATION_FAIL_TOLERANCE = 0.25 # fail when output duration drifts >25% - - -def verify_output_duration(path: Path, expected_ms: int) -> bool: - """Sanity-check an assembled file's duration against the expected total. - - Catches corrupt assembly (truncated concat, bogus container metadata) - before the file reaches audiobook players. Duration drift beyond the - warn tolerance is logged; drift beyond the fail tolerance is an error - and the output is treated as broken. Returns True when unverifiable. - """ - if expected_ms <= 0: - return True - actual_ms = probe_duration_ms(path) - if actual_ms <= 0: - logger.warning("Could not verify duration of %s (ffprobe failed)", path) - return True - drift = abs(actual_ms - expected_ms) / expected_ms - if drift > _DURATION_FAIL_TOLERANCE: - logger.error( - "Duration mismatch for %s: expected ~%.1fs, got %.1fs (%.0f%% off); output is likely corrupt", - path.name, expected_ms / 1000.0, actual_ms / 1000.0, drift * 100.0, - ) - return False - if drift > _DURATION_WARN_TOLERANCE: - logger.warning( - "Duration drift for %s: expected ~%.1fs, got %.1fs (%.0f%% off)", - path.name, expected_ms / 1000.0, actual_ms / 1000.0, drift * 100.0, - ) - return True - - -def _collect_chunk_files(total_chunks: int, - chunk_results: Dict[int, Optional[Path]] - ) -> Tuple[List[Path], List[int]]: - """Resolve chunk audio files in book order. - - ``chunk_results`` maps chunk number -> path written (or None for a failed - chunk); recorded paths are used exactly as-is so stale files from a - previous chapter can never leak in. - """ - chunk_files: List[Path] = [] - missing: List[int] = [] - for i in range(1, total_chunks + 1): - recorded = chunk_results.get(i) - if recorded is not None and Path(recorded).exists(): - chunk_files.append(Path(recorded)) - else: - missing.append(i) - return chunk_files, missing - - -def combine_chunks(total_chunks: int, output_path: Path, - chunk_results: Dict[int, Optional[Path]], - speed: float = 1.0, output_format: str = config.AUDIO_FORMAT, - intermediate: bool = False, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None) -> bool: - """Combine audio chunks into the final audiobook using ffmpeg's concat demuxer. - - ``chunk_results`` maps chunk numbers to the audio file each chunk produced - (None for failed chunks); failed and missing chunks are skipped. When - ``speed`` differs from 1.0, an additional speed-adjusted copy is written - next to the normal-speed file. ``meta``/``cover`` embed tags and cover - art into the output (skipped for intermediate chapter scratch audio). - Chunks are streamed by ffmpeg, so the whole book is never held in - memory. Set ``intermediate`` for scratch chapter audio on the way to a - larger output (e.g. a chaptered m4b) so save messages don't present it - as the final audiobook. - """ - if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None: - logger.error("ffmpeg and ffprobe are required to combine audio chunks (install ffmpeg)") - return False - - chunk_files, missing_chunks = _collect_chunk_files(total_chunks, chunk_results) - - if not chunk_files: - logger.error("No valid chunks found") - return False - - if missing_chunks: - logger.warning("Missing chunks: %s", missing_chunks) - - 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: - list_file.write(f"file '{_concat_escape(str(chunk_file))}'\n") - - speed_path = None - if atempo_filters(speed): - speed_path = output_path.with_name(f"{output_path.stem}_{speed:g}{output_path.suffix}") - cmd = build_concat_command(concat_list, output_path, output_format, - speed=speed, speed_path=speed_path, - meta=None if intermediate else meta, - cover=None if intermediate else cover) - - proc = subprocess.run(cmd, capture_output=True, text=True) - if proc.returncode != 0: - logger.error("ffmpeg failed: %s", proc.stderr[-2000:]) - return False - - # Verify the assembled duration against the sum of chunk durations - # so corrupt output is caught before it reaches audiobook players. - # A failed probe returns 0, which would deflate the expected total - # and falsely fail the check, so unverifiable sums skip it. - durations = [probe_duration_ms(chunk_file) for chunk_file in chunk_files] - if any(duration <= 0 for duration in durations): - logger.warning("Could not probe every chunk duration; skipping duration verification") - duration_ok = True - else: - expected_ms = sum(durations) - duration_ok = verify_output_duration(output_path, expected_ms) - if speed_path is not None: - duration_ok = verify_output_duration(speed_path, int(expected_ms / speed)) and duration_ok - if not duration_ok: - return False - - if intermediate: - logger.info("Chapter audio saved (intermediate): %s (%d/%d chunks)", - output_path, len(chunk_files), total_chunks) - if len(chunk_files) == 1 and total_chunks == 1: - print(f"[INFO] Saved chapter audio (intermediate): " - f"{output_path.name}") - else: - print(f"[INFO] Saved chapter audio (intermediate): {output_path.name} " - f"({len(chunk_files)}/{total_chunks} chunks)") - else: - logger.info("Audiobook saved: %s (%d/%d chunks)", output_path, len(chunk_files), total_chunks) - if len(chunk_files) == 1 and total_chunks == 1: - print(f"[INFO] Saved audiobook: {output_path.name}") - else: - print(f"[INFO] Saved audiobook: {output_path.name} " - f"({len(chunk_files)}/{total_chunks} chunks)") - - if speed_path is not None: - logger.info("Saved speed-adjusted audiobook (%gx): %s", speed, speed_path) - print(f"[INFO] Saved speed-adjusted audiobook: {speed_path.name} ({speed:g}x)") - - return True - - except FileNotFoundError: - logger.error("ffmpeg not found on PATH (install ffmpeg and try again)") - return False - except Exception as exc: - logger.error("Failed to combine chunks: %s", exc) - logger.error(traceback.format_exc()) - return False - finally: - try: - concat_list.unlink(missing_ok=True) - except OSError: - pass - - -def cleanup_chunks() -> None: - """Remove temporary chunk and chapter files from the scratch folder.""" - try: - chunk_count = 0 - for pattern in ("chunk_*", "chapter_*"): - for chunk_file in CHUNKS_FOLDER.glob(pattern): - try: - if chunk_file.is_file(): - chunk_file.unlink() - chunk_count += 1 - except Exception as exc: - logger.warning("Failed to delete %s: %s", chunk_file, exc) - - if chunk_count > 0: - logger.info("Cleaned up %d chunk files", chunk_count) - print(f"[INFO] Cleaned up {chunk_count} chunk files") - except Exception as exc: - logger.warning("Cleanup failed: %s", exc) - - -def probe_duration_ms(path: Path) -> int: - """Return audio duration in milliseconds using ffprobe.""" - try: - result = subprocess.run( - ["ffprobe", "-v", "error", "-show_entries", "format=duration", - "-of", "default=noprint_wrappers=1:nokey=1", str(path)], - capture_output=True, text=True, timeout=30, - ) - except subprocess.TimeoutExpired: - logger.warning("ffprobe timed out for %s", path) - return 0 - if result.returncode != 0: - logger.warning("ffprobe failed for %s: %s", path, result.stderr[-200:]) - return 0 - try: - return max(0, int(round(float(result.stdout.strip()) * 1000.0))) - except ValueError: - logger.warning("Could not parse ffprobe duration for %s", path) - return 0 - - -def _escape_ffmetadata_value(value: str) -> str: - """Escape a metadata value for ffmpeg's FFMETADATA format. - - Backslash and the structural characters ``=``, ``;`` and ``#`` must be - backslash-escaped; line breaks would corrupt the file and are collapsed - to spaces. - """ - value = value.replace("\\", "\\\\") - value = re.sub(r"[\r\n]+", " ", value) - return re.sub(r"[=;#]", r"\\\g<0>", value) - - -def build_ffmetadata(chapters: List[tuple], path: Path) -> None: - """Write an ffmpeg FFMETADATA file with ``[CHAPTER]`` entries. - - ``chapters`` is a list of ``(start_ms, end_ms, title)`` tuples. - """ - with open(path, "w", encoding="utf-8") as meta_file: - meta_file.write(";FFMETADATA1\n") - for start_ms, end_ms, title in chapters: - meta_file.write("[CHAPTER]\n") - meta_file.write("TIMEBASE=1/1000\n") - meta_file.write(f"START={int(start_ms)}\n") - meta_file.write(f"END={int(end_ms)}\n") - meta_file.write(f"title={_escape_ffmetadata_value(title)}\n") - - -def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str], - output_path: Path, speed: float = 1.0, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None) -> bool: - """Concatenate per-chapter audio into a single m4b with embedded chapter markers. - - Chapter start/end times are derived from each chapter file's duration and - written as ffmpeg chapter metadata. ``meta``/``cover`` embed tags and - cover art. When ``speed`` differs from 1.0, a speed-adjusted copy (with - rescaled chapter markers) is written alongside the normal-speed file. - """ - if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None: - logger.error("ffmpeg and ffprobe are required to build an m4b with chapters") - return False - - if not chapter_files: - logger.error("No chapter files provided") - return False - - 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 - with open(concat_list, "w", encoding="utf-8") as list_file: - for chapter_file, title in zip(chapter_files, titles): - list_file.write(f"file '{_concat_escape(str(chapter_file))}'\n") - duration_ms = probe_duration_ms(chapter_file) - end_ms = start_ms + duration_ms - chapters.append((start_ms, end_ms, title or "Chapter")) - start_ms = end_ms - - build_ffmetadata(chapters, metadata_file) - - speed_path = None - if atempo_filters(speed): - speed_path = output_path.with_name(f"{output_path.stem}_{speed:g}{output_path.suffix}") - scaled = [(int(s / speed), int(e / speed), t) for s, e, t in chapters] - build_ffmetadata(scaled, speed_metadata_file) - cmd = build_m4b_chapters_command(concat_list, metadata_file, output_path, - speed=speed, speed_path=speed_path, - speed_metadata_file=speed_metadata_file - if speed_path is not None else None, - meta=meta, cover=cover) - - proc = subprocess.run(cmd, capture_output=True, text=True) - if proc.returncode != 0: - logger.error("ffmpeg failed: %s", proc.stderr[-2000:]) - return False - - # The last chapter's end time is the expected total duration; skip - # verification when any chapter duration probe failed (returned 0) - # so an unprobed chapter can't falsely fail the whole output. - expected_ms = chapters[-1][1] - if any(end_ms - start_ms <= 0 for start_ms, end_ms, _ in chapters): - logger.warning("Could not probe every chapter duration; skipping duration verification") - duration_ok = True - else: - duration_ok = verify_output_duration(output_path, expected_ms) - if speed_path is not None: - duration_ok = verify_output_duration(speed_path, int(expected_ms / speed)) and duration_ok - if not duration_ok: - return False - - logger.info("Audiobook saved: %s (%d chapters)", output_path, len(chapter_files)) - print(f"[INFO] Saved audiobook: {output_path.name} ({len(chapter_files)} chapters)") - - if speed_path is not None: - logger.info("Saved speed-adjusted audiobook (%gx): %s", speed, speed_path) - print(f"[INFO] Saved speed-adjusted audiobook: {speed_path.name} ({speed:g}x)") - - return True - - except FileNotFoundError: - logger.error("ffmpeg/ffprobe not found on PATH (install ffmpeg and try again)") - return False - except Exception as exc: - logger.error("Failed to combine chapters: %s", exc) - logger.error(traceback.format_exc()) - return False - finally: - for scratch in (concat_list, metadata_file, speed_metadata_file): - try: - scratch.unlink(missing_ok=True) - except OSError: - pass diff --git a/converter/chunking.py b/converter/chunking.py deleted file mode 100644 index 800b76a..0000000 --- a/converter/chunking.py +++ /dev/null @@ -1,91 +0,0 @@ -"""Split extracted book text into TTS-sized chunks.""" - -import logging -import re -from typing import List, Optional - -from . import config - -logger = logging.getLogger(__name__) - - -def split_into_chunks(text: str, max_words: Optional[int] = None) -> List[str]: - """Split text into chunks of at most ``max_words`` words. - - ``max_words`` defaults to ``config.CHUNK_SIZE`` (read at call time). - There is no ceiling beyond that setting, but note that the TTS - servers silently truncate audio when a single generation runs too - long without reporting an error, so very large values are at your - own risk (see CHUNK_SIZE in converter/config.py). - - Splits on sentence boundaries. Sentences longer than the limit are - split further at clause punctuation (which is kept attached for TTS - prosody). Clause splits only happen at whitespace after punctuation, - so tokens like "1,000,000" or "12:30" are never broken apart. A piece - with no usable punctuation split point longer than the limit is split - at word boundaries as a last resort: individual tokens stay intact, - but whitespace between them is normalized. - """ - if max_words is None: - max_words = config.CHUNK_SIZE - if max_words < 1: - max_words = 1 - - if not text.strip(): - return [] - - sentences = re.split(r"(?<=[.!?])\s+", text) - chunks = [] - current_chunk = "" - current_words = 0 - - for sentence in sentences: - sentence_words = len(sentence.split()) - - if sentence_words > max_words: - if current_chunk: - chunks.append(current_chunk.strip()) - current_chunk = "" - current_words = 0 - - # Split long sentences at clause boundaries, keeping punctuation. - # Only split where whitespace already follows the punctuation so - # tokens are never broken apart or re-joined with added spaces - # (no spaces are injected into "1,000,000" or "12:30"). - parts = re.split(r"(?<=[,;:])\s+", sentence) - for part in parts: - part_words = len(part.split()) - if part_words > max_words: - # Last resort: no punctuation split point is available, - # so split at word boundaries. Tokens themselves (and - # therefore numbers like "1,000,000") stay intact. - if current_chunk: - chunks.append(current_chunk.strip()) - current_chunk = "" - current_words = 0 - words = part.split() - for start in range(0, len(words), max_words): - chunks.append(" ".join(words[start:start + max_words])) - continue - if current_words + part_words <= max_words: - current_chunk += part + " " - current_words += part_words - else: - if current_chunk: - chunks.append(current_chunk.strip()) - current_chunk = part + " " - current_words = part_words - else: - if current_words + sentence_words <= max_words: - current_chunk += sentence + " " - current_words += sentence_words - else: - if current_chunk: - chunks.append(current_chunk.strip()) - current_chunk = sentence + " " - current_words = sentence_words - - if current_chunk.strip(): - chunks.append(current_chunk.strip()) - - return [chunk for chunk in chunks if chunk.strip()] diff --git a/converter/config.py b/converter/config.py deleted file mode 100644 index 6a98136..0000000 --- a/converter/config.py +++ /dev/null @@ -1,77 +0,0 @@ -# Default output options -AUDIO_FORMAT = "m4b" -AUDIO_BITRATE = "128k" -LANGUAGE = "English" - -API_TIMEOUT = 600 # Timeout per chunk request in seconds -MAX_RETRIES = 3 # Attempts per chunk request -HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" in console logs every N seconds - -# Words per TTS generation request (client-side chunking). -# The qwen and faster backends always chunk with this size -# The audio.cpp backend chunks long text itself, so this is ignored -# by default with that backend. Force chunking with --chunk -CHUNK_SIZE = 250 - -# Default TTS backend. -# audiocpp: audiocpp_server -# qwen: qwen-tts-demo -# faster: faster-qwen-tts -# The --backend CLI flag overrides this -BACKEND = "audiocpp" - -############################################################################### -# BACKEND 1: qwen-tts-demo (qwen) options # -############################################################################### - -# There are different API URLs for CustomVoice and Base models so you can run both at once -QWEN_API_URL = "http://127.0.0.1:7860" # CustomVoice model -CLONE_API_URL = "http://127.0.0.1:7861" # Base model - -# Custom voice options -SPEAKER = "Vivian" #Vivian, Serena, Uncle_Fu, Dylan, Eric, Ryan, Aiden, Ono_Anna, Sohee -INSTRUCT = "Speak naturally and clearly, as if reading a dramatic book to an adult audience." - -# Don't clone with transcription, only use x-vector-only cloning. Generally "worse" -XVECTOR_ONLY = False - -# Randomization seed. -1 means randomize with every generation -# With SEED = -1 and CONSTANT_SEED = True, one random seed will be used for the entire audiobook. -# This may keep the voice slightly more consistent across chunk boundaries -SEED = -1 -CONSTANT_SEED = False - -############################################################################### -# BACKEND 2: faster-qwen-tts options # -############################################################################### -FASTER_API_URL = "http://127.0.0.1:8000" # faster-qwen3-tts server (Base model only) - -# Default voice if no --voice is passed -FASTER_VOICE = "default" - -############################################################################### -# BACKEND 3: audio.cpp options # -############################################################################### -AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server - -# Model ids in the audio.cpp server.json config. AUDIOCPP_MODEL_ID may point -# at any TTS model entry the server hosts (qwen3_tts, higgs_audio_tts, -# voxcpm2, index_tts2, ...); the family is detected from the server at -# startup and adapts the request automatically. Only qwen3_tts has built-in -# speakers (speaker mode); every other family needs --voice with a -# server-side voice preset. For single-model servers, set -# AUDIOCPP_CLONE_MODEL_ID to the same id as AUDIOCPP_MODEL_ID (or leave it -# empty); for Qwen3-TTS it typically names a second entry with the Base -# (cloning) model. A multi-model server (one server.json hosting several -# lazily-loaded entries) does not need editing here: leave AUDIOCPP_MODEL_ID -# unset to auto-select when only one entry is hosted, or pick the entry per -# run with the --model CLI flag. -AUDIOCPP_MODEL_ID = "qwen" -AUDIOCPP_CLONE_MODEL_ID = "qwen" - -# Voice design / style instruction sent with every audio.cpp request when -# the --instructions CLI flag is not given. Required for server entries -# hosted with task "vdes" (voice design models such as Qwen3-TTS -# VoiceDesign); on other families it acts as a style/delivery instruction -# when the model supports one and is ignored otherwise. Empty by default. -AUDIOCPP_INSTRUCTIONS = "" diff --git a/converter/converter.py b/converter/converter.py deleted file mode 100644 index b24ac09..0000000 --- a/converter/converter.py +++ /dev/null @@ -1,780 +0,0 @@ -"""Orchestrates book-to-audiobook conversion.""" - -import glob -import logging -import re -import shutil -import sys -import time -import traceback -from collections import Counter -from datetime import datetime -from pathlib import Path -from typing import Dict, List, Optional, Tuple - -from . import audio, chunking, config, cover, extractors -from .audio import TrackMeta -from .tts import ( - BACKENDS, - BACKEND_AUDIOCPP, - BACKEND_FASTER, - BACKEND_QWEN, - MODEL_SIZE, - VOICE_MODE_CLONE, - VOICE_MODE_CUSTOM, - VOICE_MODES, - AudioCppTTSClient, - 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).""" - return not record.name.startswith(("httpx", "httpcore")) - - -def setup_logging(debug: bool = False) -> None: - """Configure logging to a dated file and the console. - - The file keeps the full record (DEBUG with --debug), including httpx - request logs. The console handler only surfaces warnings and errors - (DEBUG with --debug) so progress prints are never mirrored as - timestamped log lines; httpx/httpcore request logs stay file-only. - """ - LOGS_FOLDER.mkdir(parents=True, exist_ok=True) - file_handler = logging.FileHandler( - LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log", - encoding="utf-8", - ) - file_handler.setLevel(logging.DEBUG if debug else logging.INFO) - console_handler = logging.StreamHandler(sys.stdout) - console_handler.setLevel(logging.DEBUG if debug else logging.WARNING) - console_handler.addFilter(_console_log_filter) - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(levelname)s - %(message)s", - handlers=[file_handler, console_handler], - ) - if debug: - logging.getLogger("converter").setLevel(logging.DEBUG) - - -def setup_directories() -> None: - """Create necessary directories.""" - for directory in (BOOKS_FOLDER, AUDIOBOOKS_FOLDER, - CHUNKS_FOLDER, LOGS_FOLDER): - Path(directory).mkdir(parents=True, exist_ok=True) - - -def find_existing_outputs(output_name: str, output_format: str) -> List[Path]: - """Return existing output files that a conversion would overwrite. - - Multi-section books (e.g. EPUB chapters) and speed-adjusted copies are - named ``{name}_suffix.{ext}``; exact chapter file names are only known - after text extraction, so any file matching that pattern counts. - """ - folder = AUDIOBOOKS_FOLDER - existing: List[Path] = [] - primary = folder / f"{output_name}.{output_format}" - if primary.exists(): - existing.append(primary) - existing.extend(sorted( - folder.glob(f"{glob.escape(output_name)}_*.{output_format}"))) - return existing - - -def prompt_overwrite(existing: List[Path], output_name: str) -> bool: - """Ask whether to reconvert a book whose output files already exist. - - All overwrite questions are asked before any conversion starts so the - rest of the run is unattended. Pressing Enter defaults to yes (so a - user can just hit Enter through the prompts), but a closed stdin - (non-interactive run) declines and keeps existing files safe. - """ - if len(existing) == 1: - message = f"{existing[0].name} already exists. Convert anyway and overwrite it?" - else: - message = (f"{len(existing)} output files for '{output_name}' already exist " - f"(e.g. {existing[0].name}). Convert anyway and overwrite them?") - while True: - try: - answer = input(f"{message} [Y/n]: ").strip().lower() - except EOFError: - print("\n[WARNING] No interactive input available; keeping existing output") - return False - if not answer: - return True - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n' (or press Enter for yes).") - - -class AudiobookConverter: - """Audiobook converter using a local TTS API.""" - - 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, backend: str = config.BACKEND, - voice: Optional[str] = None, debug: bool = False, - chunk: bool = False, model_id: Optional[str] = None, - instructions: Optional[str] = None, - request_options: Optional[Dict[str, str]] = None): - if speed <= 0: - raise ValueError(f"Speed must be a positive number, got {speed}") - if output_format not in AUDIO_FORMATS: - raise ValueError(f"Unsupported output format: {output_format}") - if backend not in BACKENDS: - raise ValueError( - f"Unknown backend: {backend!r} (expected one of {BACKENDS})" - ) - if language is None: - language = config.LANGUAGE - self.language = normalize_language(language) - self.voice_mode = voice_mode - self.voice_clone_ref_audio = voice_clone_ref_audio - self.speed = speed - self.single_file = single_file - self.output_format = output_format - self.backend = backend - self.voice = voice - self.debug = bool(debug) - # Client-side chunking: the qwen and faster backends always chunk - # (their servers do one generation per request and silently truncate - # long text). The audio.cpp server chunks long text itself, so it - # defaults to one request per chapter; --chunk forces client-side - # chunking on top (possible needless double-chunking). - self.client_chunks = bool(chunk) or backend != BACKEND_AUDIOCPP - # Voice design / style instruction and free-form request options - # (audio.cpp only): forwarded to AudioCppTTSClient, which validates - # them against the server-hosted model at connect time. - self.instructions = instructions - self.request_options = dict(request_options or {}) - self._validate_configuration() - if backend == BACKEND_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=voice) - elif backend == BACKEND_AUDIOCPP: - # Speaker mode (no voice) uses a built-in CustomVoice speaker; - # an explicit voice selects a server-side preset (cloning). - # model_id overrides AUDIOCPP_MODEL_ID for multi-model servers; - # instructions describe or style the voice, request_options pass - # per-model controls through to the server. - self.tts = AudioCppTTSClient(voice=voice, language=self.language, - chunk_text=self.client_chunks, - model_id=model_id, - instructions=instructions, - request_options=self.request_options) - 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.""" - if self.voice_mode not in VOICE_MODES: - raise ValueError( - f"Unknown voice mode: {self.voice_mode!r} " - f"(expected one of {VOICE_MODES})" - ) - if self.voice_mode == VOICE_MODE_CLONE and self.backend == BACKEND_QWEN: - if not self.voice_clone_ref_audio: - raise ValueError( - "Voice Clone mode requires a reference audio file. " - "Use --clone <path> to specify it." - ) - - if not Path(self.voice_clone_ref_audio).exists(): - raise ValueError( - f"Reference audio file not found: {self.voice_clone_ref_audio}" - ) - - @staticmethod - def _sanitize_filename(name: str, fallback: str = "chapter") -> str: - """Make a chapter title safe to use as part of a file name.""" - cleaned = re.sub(r'[\\/:*?"<>|]', " ", name) - cleaned = re.sub(r"\s+", " ", cleaned).strip().strip(".") - return cleaned[:80] or fallback - - def _narrator_tag(self) -> str: - """Narrator name used in output file names (see compute_narrator_tag).""" - return self.compute_narrator_tag( - self.backend, self.voice, self.voice_mode, - self.voice_clone_ref_audio, self.instructions) - - @staticmethod - def compute_narrator_tag(backend: str, voice: Optional[str], - voice_mode: str, - voice_clone_ref_audio: Optional[str], - instructions: Optional[str] = None) -> str: - """Narrator name used in output file names, without a server connection. - - Custom voice mode uses the built-in speaker's display name; voice - clone mode uses the reference audio file's stem; the faster and - audiocpp backends use the server-side voice name (falling back to - the built-in speaker for the audiocpp backend's speaker mode). An - instruction without a voice (voice design, or instruction-defined - voices on families without built-in speakers) uses "designed". - Spaces become underscores (e.g. "Uncle Fu" -> "Uncle_Fu"). - - Pure (no I/O, no server) so the pre-flight overwrite check can - compute the exact output names a run would produce before spending - time connecting to a TTS server. - """ - if backend == BACKEND_FASTER: - narrator = voice or config.FASTER_VOICE - elif backend == BACKEND_AUDIOCPP: - if voice: - narrator = voice - elif instructions: - # The voice comes from the instruction, not a speaker name. - narrator = "designed" - else: - narrator = speaker_display_name() - elif voice_mode == VOICE_MODE_CLONE: - narrator = Path(voice_clone_ref_audio).stem - else: - narrator = speaker_display_name() - return AudiobookConverter._sanitize_filename( - narrator, fallback="narrator").replace(" ", "_") - - # ------------------------------------------------------------------ - # Debug dumps (--debug) - # ------------------------------------------------------------------ - - @staticmethod - def _write_debug_text(debug_dir: Path, chunk_num: int, text: str) -> None: - """Write the exact text sent for a chunk to the debug folder. - - Called before the request so the text survives a crash mid-generation. - A failed debug write must never abort a conversion. - """ - try: - debug_dir.mkdir(parents=True, exist_ok=True) - (debug_dir / f"chunk_{chunk_num:04d}.txt").write_text(text, encoding="utf-8") - except OSError as exc: - logger.warning("Could not write debug text for chunk %d: %s", chunk_num, exc) - - @staticmethod - def _copy_debug_audio(debug_dir: Path, chunk_num: int, source: Path) -> Optional[Path]: - """Copy a generated chunk's audio file into the debug folder. - - Returns the copy's path, or None when the copy failed (which never - affects the conversion itself). - """ - try: - debug_dir.mkdir(parents=True, exist_ok=True) - target = debug_dir / f"chunk_{chunk_num:04d}{source.suffix or '.wav'}" - shutil.copy2(source, target) - return target - except OSError as exc: - logger.warning("Could not write debug audio for chunk %d: %s", chunk_num, exc) - return None - - @staticmethod - def _chapter_debug_dir(book_debug_dir: Optional[Path], index: int, title: str) -> Optional[Path]: - """Per-chapter subfolder of a book's debug folder (None when not debugging). - - Chunk numbering restarts for each chapter, so chapters get their own - subfolder (e.g. debug/dune_Vivian/03_The_Trial/). - """ - if book_debug_dir is None: - return None - return book_debug_dir / f"{index:02d}_{AudiobookConverter._sanitize_filename(title)}" - - def convert_book(self, file_path: Path, output_name: Optional[str] = None) -> bool: - """Convert a single book to one or more audiobook files.""" - logger.info("Converting: %s", file_path.name) - start_time = time.time() - - try: - # Start from a clean scratch folder so a previous crash can never - # affect this run - audio.cleanup_chunks() - - logger.info("Extracting text...") - book = extractors.extract_book(file_path) - sections = book.sections - if not sections or all(not s.text.strip() for s in sections): - logger.error("No text extracted") - return False - - stem = output_name or f"{file_path.stem}_{self._narrator_tag()}" - - # --debug: chunk text/audio dumps land in a per-book folder - 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, 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) - - # m4b is always a single file; multi-chapter books get embedded - # chapter markers so listeners can skip between chapters. - if self.output_format == "m4b": - if len(sections) > 1: - return self._convert_m4b_with_chapters(sections, stem, start_time, - meta=meta, cover=cover_path, - debug_dir=debug_dir) - 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 = 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 = 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)) - success = self._convert_text( - section.text, output_path, time.time(), - meta=track_meta, cover=cover_path, - debug_dir=self._chapter_debug_dir(debug_dir, index, section.title) - ) and success - return success - - except Exception as exc: - logger.error("Conversion failed: %s", exc) - logger.error(traceback.format_exc()) - return False - finally: - # Always cleanup, even on failure or interrupt - audio.cleanup_chunks() - - def _convert_m4b_with_chapters(self, sections, stem: str, start_time: float, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None, - debug_dir: Optional[Path] = None) -> bool: - """Convert each chapter to audio, then assemble a single m4b with - embedded chapter markers. - - Chapters are synthesized to lossless WAV scratch files (~170 MB per - hour of audio) so the final AAC pass is the only lossy encode. When - ``debug_dir`` is given, each chapter's debug dumps land in its own - subfolder (chunk numbering restarts per chapter). - """ - chapter_files = [] - titles = [] - total_chapters = len(sections) - for index, section in enumerate(sections, 1): - 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}") - print(f"{'=' * 50}") - logger.info("Converting chapter %d/%d: %s", index, total_chapters, title) - if not self._convert_text(section.text, chapter_path, time.time(), - speed=1.0, output_format="wav", - chapter=(index, total_chapters), - debug_dir=self._chapter_debug_dir(debug_dir, index, title)): - logger.error("Chapter %d (%s) failed; aborting the conversion", - index, title) - return False - chapter_files.append(chapter_path) - titles.append(title) - - if not chapter_files: - logger.error("No chapters were successfully converted") - return False - - 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 - duration = time.time() - start_time - logger.info("Conversion completed in %dm %ds: %s", - int(duration // 60), int(duration % 60), output_path) - return True - - def _synthesize_chunks(self, chunks: List[str], - debug_dir: Optional[Path] = None) -> Dict[int, Optional[Path]]: - """Synthesize chunks sequentially, preserving order and naming. - - Returns a mapping of chunk number to the generated audio path, with - None for chunks that failed. Generation stops at the first failed - chunk: a partial audiobook is never assembled, so the remaining - chunks are not requested. When ``debug_dir`` is given (--debug), - each chunk's request text and returned audio are also dumped there, - and every request/response is logged. - """ - total_chunks = len(chunks) - if self.client_chunks: - print(f"\n{'=' * 50}") - print(f"PROCESSING {total_chunks} CHUNKS") - print(f"{'=' * 50}") - - results: Dict[int, Optional[Path]] = {} - for chunk_num, chunk_text in enumerate(chunks, 1): - if debug_dir is not None: - # Written before the request so the exact text survives a - # crash mid-generation; failed chunks keep their dumps. - self._write_debug_text(debug_dir, chunk_num, chunk_text) - logger.debug("Chunk %d/%d request text: %s", chunk_num, total_chunks, chunk_text) - request_start = time.time() - try: - result = self.tts.process_chunk_with_retry(chunk_num, chunk_text) - results[chunk_num] = result - - if result: - if debug_dir is not None: - copied = self._copy_debug_audio(debug_dir, chunk_num, Path(result)) - elapsed = time.time() - request_start - destination = f" -> {copied.name}" if copied else "" - logger.debug("Chunk %d/%d response in %.1fs%s", - chunk_num, total_chunks, elapsed, destination) - if self.client_chunks: - print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") - logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks) - else: - logger.error("Chunk %d/%d failed; aborting the remaining chunks", - chunk_num, total_chunks) - break - - except Exception as exc: - results[chunk_num] = None - logger.error("Chunk %d/%d error: %s; aborting the remaining chunks", - chunk_num, total_chunks, exc) - break - - successful_chunks = sum(1 for path in results.values() if path) - if self.client_chunks: - print(f"\n{'=' * 50}") - print("CHUNK PROCESSING COMPLETE") - print(f"Successful: {successful_chunks}/{total_chunks}") - print(f"{'=' * 50}") - logger.info("Chunk processing completed: %d/%d chunks", successful_chunks, total_chunks) - return results - - def _chapter_chunks(self, text: str) -> List[str]: - """Split chapter text into TTS requests. - - Client-side chunking splits into CHUNK_SIZE-word chunks (qwen and - faster always; audio.cpp only with --chunk). Otherwise (audio.cpp - default) the whole text is one request and the server does its own - long-form chunking. - """ - if self.client_chunks: - return chunking.split_into_chunks(text) - return [text] if text.strip() else [] - - def _convert_text(self, text: str, output_path: Path, start_time: float, - speed: Optional[float] = None, - output_format: Optional[str] = None, - chapter: Optional[Tuple[int, int]] = None, - meta: Optional[TrackMeta] = None, - cover: Optional[Path] = None, - debug_dir: Optional[Path] = None) -> bool: - """Chunk, synthesize, and assemble ``text`` into ``output_path``. - - When ``chapter`` (a ``(number, total)`` pair) is given, the output is - an intermediate per-chapter file and progress messages are phrased - accordingly instead of implying the whole book is done. ``debug_dir`` - (from --debug) receives the chunks' text and audio dumps. - """ - if speed is None: - speed = self.speed - if output_format is None: - output_format = self.output_format - - try: - if not text.strip(): - logger.error("No text to convert for %s", output_path.name) - return False - - logger.info("Extracted %d characters (%d words)", len(text), len(text.split())) - - chunks = self._chapter_chunks(text) - total_chunks = len(chunks) - if total_chunks == 0: - logger.error("No chunks created") - return False - - chunk_sizes = [len(chunk.split()) for chunk in chunks] - avg_chunk_size = sum(chunk_sizes) / len(chunk_sizes) - if len(chunks) == 1: - logger.info("Sending the whole text as one request (%d words; " - "the server chunks long text itself)", - chunk_sizes[0]) - else: - logger.info("Split into %d chunks (avg %.0f words per chunk)", - total_chunks, avg_chunk_size) - backend_labels = { - BACKEND_FASTER: "faster TTS API", - BACKEND_AUDIOCPP: "audio.cpp server", - } - backend = backend_labels.get(self.backend, "Qwen API") - if self.client_chunks: - print(f"[INFO] Processing {total_chunks} chunks via {backend}...") - else: - # The whole request is sent at once and the server does its - # own long-form chunking, so the chunk vocabulary does not - # apply; warn that this one request can take a very long time. - subject = (f"chapter {chapter[0]}/{chapter[1]}" - if chapter is not None else "text") - print(f"[INFO] Sending the {subject} to the {backend} as a " - "single request...") - print("[NOTE] It is expected for this to take a very long " - "time: the server synthesizes the entire request before " - "returning any audio.") - - results = self._synthesize_chunks(chunks, debug_dir=debug_dir) - successful_chunks = sum(1 for path in results.values() if path) - - if successful_chunks < total_chunks: - logger.error("Chunk processing incomplete (%d/%d chunks); " - "aborting without producing an audiobook", - successful_chunks, total_chunks) - return False - - success = audio.combine_chunks(total_chunks, output_path, chunk_results=results, - speed=speed, output_format=output_format, - intermediate=chapter is not None, - meta=meta, cover=cover) - - if success: - duration = time.time() - start_time - minutes = int(duration // 60) - seconds = int(duration % 60) - if chapter is not None: - logger.info("Chapter %d/%d converted in %dm %ds (%d/%d chunks)", - chapter[0], chapter[1], minutes, seconds, - successful_chunks, total_chunks) - if self.client_chunks: - print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted " - f"({successful_chunks}/{total_chunks} chunks)") - else: - print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted") - else: - logger.info("Conversion completed in %dm %ds: %s", minutes, seconds, output_path) - else: - logger.error("Failed to combine chunks into final audiobook") - - return success - - except Exception as exc: - logger.error("Conversion failed: %s", exc) - logger.error(traceback.format_exc()) - return False - - def _print_banner(self) -> None: - """Print the startup summary for the selected backend.""" - print("=" * 70) - print("TTS AUDIOBOOK GENERATOR") - print("=" * 70) - print(f"Books folder: {BOOKS_FOLDER}") - print(f"Output folder: {AUDIOBOOKS_FOLDER}") - if self.backend == BACKEND_FASTER: - print(f"Faster TTS endpoint: {config.FASTER_API_URL}") - print("Backend: faster (voice cloning, reference configured on server)") - print(f"Voice: {self.voice or config.FASTER_VOICE}") - elif self.backend == BACKEND_AUDIOCPP: - print(f"audio.cpp endpoint: {config.AUDIOCPP_API_URL}") - print(f"Model id: {self.tts.model_id}") - print(f"Model family: {getattr(self.tts, 'family', 'unknown')}") - if self.voice: - print("Backend: audio.cpp (voice cloning, reference configured on server)") - print(f"Voice: {self.voice}") - elif self.instructions: - print("Backend: audio.cpp (voice from --instructions description)") - print(f"Instruction: {self.instructions}") - else: - print("Backend: audio.cpp (custom voice, built-in speaker)") - print(f"Speaker: {config.SPEAKER}") - if self.request_options: - print(f"Request options: {self.request_options}") - if self.client_chunks: - print("Chunking: client-side (--chunk; the server also chunks " - "long text itself, so this may double-chunk)") - else: - print("Chunking: server-side (one request per chapter; " - "--chunk forces client-side chunking)") - print(f"Language: {self.language}") - else: - api_url = (config.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(f"Model size: {MODEL_SIZE} (always)") - if self.voice_mode == VOICE_MODE_CUSTOM: - print(f"Speaker: {config.SPEAKER}") - print(f"Language: {self.language}") - 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}") - if self.single_file and self.output_format != "m4b": - print("Chapter mode: single file (--single-file)") - 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): {DEBUG_FOLDER}") - print("=" * 70) - - # ------------------------------------------------------------------ - # Pre-flight: overwrite checks before connecting to a TTS server - # ------------------------------------------------------------------ - - @staticmethod - def preflight_overwrites(backend: str, voice: Optional[str], - voice_mode: str, - voice_clone_ref_audio: Optional[str], - output_format: str, - instructions: Optional[str] = None - ) -> Tuple[List[Path], List[Tuple[Path, str]]]: - """Discover books and ask every overwrite question up front. - - Pure of the TTS server: it scans the books folder, computes the - output name each book would produce (including the narrator tag - and stem-collision suffix), and asks whether to overwrite any - existing output files. Returns ``(book_files, planned)`` where - ``planned`` is the subset the user agreed to (re)convert. - - Asking before connecting means a user who declines a prompt (or has - nothing to convert) never waits on a slow server handshake. - """ - book_files = sorted( - f for f in BOOKS_FOLDER.iterdir() - if f.is_file() and f.suffix.lower() in SUPPORTED_FORMATS - ) - if not book_files: - return [], [] - - print(f"[INFO] Found {len(book_files)} books to convert") - - # Avoid output collisions when two books share a stem (e.g. dune.txt + dune.epub). - stem_counts: Dict[str, int] = Counter(book_file.stem for book_file in book_files) - - # Ask every overwrite question up front, before any conversion - # starts, so the rest of the run is unattended. - planned: List[Tuple[Path, str]] = [] - narrator_tag = AudiobookConverter.compute_narrator_tag( - backend, voice, voice_mode, voice_clone_ref_audio, instructions) - for book_file in book_files: - output_name = book_file.stem - if stem_counts[book_file.stem] > 1: - output_name = f"{book_file.stem}_{book_file.suffix.lstrip('.')}" - output_name = f"{output_name}_{narrator_tag}" - existing = find_existing_outputs(output_name, output_format) - if existing and not prompt_overwrite(existing, output_name): - print(f"[INFO] Skipping {book_file.name} (existing output kept)") - continue - planned.append((book_file, output_name)) - return book_files, planned - - # ------------------------------------------------------------------ - # Main conversion loop - # ------------------------------------------------------------------ - - def run(self) -> bool: - """Main conversion process. Returns True if all books converted.""" - run_start = time.time() - self._print_banner() - - # When main() has already done the pre-flight overwrite check, use - # its results so the prompts are not asked a second time; otherwise - # (e.g. a converter constructed directly) discover and ask here. - if getattr(self, "_planned", None) is not None: - book_files = self._book_files - planned = self._planned - else: - book_files, planned = AudiobookConverter.preflight_overwrites( - self.backend, self.voice, self.voice_mode, - self.voice_clone_ref_audio, self.output_format, - self.instructions) - - if not book_files: - print(f"[INFO] No supported files found in {BOOKS_FOLDER}") - print(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}") - print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " - f"to {BOOKS_FOLDER} and run again.") - return True - - if not planned: - print("[INFO] Nothing to convert (all books skipped)") - return True - - print(f"[INFO] Converting {len(planned)} of {len(book_files)} book(s)") - - results = {} - for book_file, output_name in planned: - try: - success = self.convert_book(book_file, output_name=output_name) - results[book_file.name] = success - except KeyboardInterrupt: - print("\n[WARNING] Conversion interrupted by user") - results[book_file.name] = False - break - except Exception as exc: - logger.error("Unexpected error: %s", exc) - results[book_file.name] = False - if not results[book_file.name]: - logger.error("Conversion of %s failed; aborting the remaining books", - book_file.name) - break - - successful = sum(results.values()) - total = len(results) - - print("\n" + "=" * 70) - print("CONVERSION SUMMARY") - print("=" * 70) - print(f"Total: {total} | Success: {successful} | Failed: {total - successful}") - print("=" * 70) - - for filename, success in results.items(): - status = "[OK]" if success else "[FAIL]" - print(f"{status} {filename}") - - if successful > 0: - print(f"\n[INFO] Audiobooks saved to: {AUDIOBOOKS_FOLDER}/") - - elapsed = int(time.time() - run_start) - hours, remainder = divmod(elapsed, 3600) - minutes, seconds = divmod(remainder, 60) - if hours: - duration = f"{hours}h {minutes}m {seconds}s" - elif minutes: - duration = f"{minutes}m {seconds}s" - else: - duration = f"{seconds}s" - print(f"\n[INFO] Generation completed in {duration}") - logger.info("Generation completed in %s", duration) - - return total > 0 and successful == total diff --git a/converter/cover.py b/converter/cover.py deleted file mode 100644 index b2d3cb5..0000000 --- a/converter/cover.py +++ /dev/null @@ -1,279 +0,0 @@ -"""Book cover generation: stdlib-only PNG with gradient + title text. - -Covers are a vertical gradient between two random light colors with the -book title rendered on top in white with a black outline and a drop -shadow, using an embedded 5x7 bitmap font. No third-party image libraries -are required: PNG scanlines are packed and compressed with zlib/struct -directly. -""" - -import logging -import random -import struct -import zlib -from colorsys import hsv_to_rgb -from pathlib import Path -from typing import List, Optional, Tuple - -logger = logging.getLogger(__name__) - -COVER_WIDTH = 600 -COVER_HEIGHT = 900 - -# 5x7 bitmap font for ASCII 32..126. Each glyph is 7 rows of 5 bits -# (MSB left), encoded as ints for compactness. -_FONT = { - " ": [0, 0, 0, 0, 0, 0, 0], - "!": [0x04, 0x04, 0x04, 0x04, 0x04, 0x00, 0x04], - '"': [0x0A, 0x0A, 0x00, 0x00, 0x00, 0x00, 0x00], - "#": [0x0A, 0x0A, 0x1F, 0x0A, 0x1F, 0x0A, 0x0A], - "$": [0x04, 0x0F, 0x14, 0x0E, 0x05, 0x1E, 0x04], - "%": [0x18, 0x19, 0x02, 0x04, 0x08, 0x13, 0x03], - "&": [0x08, 0x14, 0x14, 0x08, 0x15, 0x12, 0x0D], - "'": [0x04, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00], - "(": [0x02, 0x04, 0x08, 0x08, 0x08, 0x04, 0x02], - ")": [0x08, 0x04, 0x02, 0x02, 0x02, 0x04, 0x08], - "*": [0x00, 0x04, 0x15, 0x0E, 0x15, 0x04, 0x00], - "+": [0x00, 0x04, 0x04, 0x1F, 0x04, 0x04, 0x00], - ",": [0x00, 0x00, 0x00, 0x00, 0x04, 0x04, 0x08], - "-": [0x00, 0x00, 0x00, 0x1F, 0x00, 0x00, 0x00], - ".": [0x00, 0x00, 0x00, 0x00, 0x00, 0x0C, 0x0C], - "/": [0x00, 0x01, 0x02, 0x04, 0x08, 0x10, 0x00], - "0": [0x0E, 0x11, 0x13, 0x15, 0x19, 0x11, 0x0E], - "1": [0x04, 0x0C, 0x04, 0x04, 0x04, 0x04, 0x0E], - "2": [0x0E, 0x11, 0x01, 0x02, 0x04, 0x08, 0x1F], - "3": [0x1F, 0x02, 0x04, 0x02, 0x01, 0x11, 0x0E], - "4": [0x02, 0x06, 0x0A, 0x12, 0x1F, 0x02, 0x02], - "5": [0x1F, 0x10, 0x1E, 0x01, 0x01, 0x11, 0x0E], - "6": [0x06, 0x08, 0x10, 0x1E, 0x11, 0x11, 0x0E], - "7": [0x1F, 0x01, 0x02, 0x04, 0x08, 0x08, 0x08], - "8": [0x0E, 0x11, 0x11, 0x0E, 0x11, 0x11, 0x0E], - "9": [0x0E, 0x11, 0x11, 0x0F, 0x01, 0x02, 0x0C], - ":": [0x00, 0x0C, 0x0C, 0x00, 0x0C, 0x0C, 0x00], - ";": [0x00, 0x0C, 0x0C, 0x00, 0x0C, 0x04, 0x08], - "<": [0x02, 0x04, 0x08, 0x10, 0x08, 0x04, 0x02], - "=": [0x00, 0x00, 0x1F, 0x00, 0x1F, 0x00, 0x00], - ">": [0x08, 0x04, 0x02, 0x01, 0x02, 0x04, 0x08], - "?": [0x0E, 0x11, 0x01, 0x02, 0x04, 0x00, 0x04], - "@": [0x0E, 0x11, 0x17, 0x15, 0x17, 0x10, 0x0E], - "A": [0x0E, 0x11, 0x11, 0x1F, 0x11, 0x11, 0x11], - "B": [0x1E, 0x11, 0x11, 0x1E, 0x11, 0x11, 0x1E], - "C": [0x0E, 0x11, 0x10, 0x10, 0x10, 0x11, 0x0E], - "D": [0x1C, 0x12, 0x11, 0x11, 0x11, 0x12, 0x1C], - "E": [0x1F, 0x10, 0x10, 0x1E, 0x10, 0x10, 0x1F], - "F": [0x1F, 0x10, 0x10, 0x1E, 0x10, 0x10, 0x10], - "G": [0x0E, 0x11, 0x10, 0x17, 0x11, 0x11, 0x0F], - "H": [0x11, 0x11, 0x11, 0x1F, 0x11, 0x11, 0x11], - "I": [0x0E, 0x04, 0x04, 0x04, 0x04, 0x04, 0x0E], - "J": [0x01, 0x01, 0x01, 0x01, 0x01, 0x11, 0x0E], - "K": [0x11, 0x12, 0x14, 0x18, 0x14, 0x12, 0x11], - "L": [0x10, 0x10, 0x10, 0x10, 0x10, 0x10, 0x1F], - "M": [0x11, 0x1B, 0x15, 0x15, 0x11, 0x11, 0x11], - "N": [0x11, 0x19, 0x15, 0x13, 0x11, 0x11, 0x11], - "O": [0x0E, 0x11, 0x11, 0x11, 0x11, 0x11, 0x0E], - "P": [0x1E, 0x11, 0x11, 0x1E, 0x10, 0x10, 0x10], - "Q": [0x0E, 0x11, 0x11, 0x11, 0x15, 0x12, 0x0D], - "R": [0x1E, 0x11, 0x11, 0x1E, 0x14, 0x12, 0x11], - "S": [0x0F, 0x10, 0x10, 0x0E, 0x01, 0x01, 0x1E], - "T": [0x1F, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04], - "U": [0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x0E], - "V": [0x11, 0x11, 0x11, 0x11, 0x11, 0x0A, 0x04], - "W": [0x11, 0x11, 0x11, 0x15, 0x15, 0x15, 0x0A], - "X": [0x11, 0x11, 0x0A, 0x04, 0x0A, 0x11, 0x11], - "Y": [0x11, 0x11, 0x0A, 0x04, 0x04, 0x04, 0x04], - "Z": [0x1F, 0x01, 0x02, 0x04, 0x08, 0x10, 0x1F], - "[": [0x0E, 0x08, 0x08, 0x08, 0x08, 0x08, 0x0E], - "\\": [0x00, 0x10, 0x08, 0x04, 0x02, 0x01, 0x00], - "]": [0x0E, 0x02, 0x02, 0x02, 0x02, 0x02, 0x0E], - "^": [0x04, 0x0A, 0x11, 0x00, 0x00, 0x00, 0x00], - "_": [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x1F], - "`": [0x08, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00], - "a": [0x00, 0x00, 0x0E, 0x01, 0x0F, 0x11, 0x0F], - "b": [0x10, 0x10, 0x1E, 0x11, 0x11, 0x11, 0x1E], - "c": [0x00, 0x00, 0x0F, 0x10, 0x10, 0x10, 0x0F], - "d": [0x01, 0x01, 0x0F, 0x11, 0x11, 0x11, 0x0F], - "e": [0x00, 0x00, 0x0E, 0x11, 0x1F, 0x10, 0x0E], - "f": [0x06, 0x08, 0x1E, 0x08, 0x08, 0x08, 0x08], - "g": [0x00, 0x0F, 0x11, 0x11, 0x0F, 0x01, 0x1E], - "h": [0x10, 0x10, 0x1E, 0x11, 0x11, 0x11, 0x11], - "i": [0x04, 0x00, 0x0C, 0x04, 0x04, 0x04, 0x0E], - "j": [0x02, 0x00, 0x06, 0x02, 0x02, 0x12, 0x0C], - "k": [0x10, 0x10, 0x12, 0x14, 0x18, 0x14, 0x12], - "l": [0x0C, 0x04, 0x04, 0x04, 0x04, 0x04, 0x0E], - "m": [0x00, 0x00, 0x1A, 0x15, 0x15, 0x15, 0x15], - "n": [0x00, 0x00, 0x1E, 0x11, 0x11, 0x11, 0x11], - "o": [0x00, 0x00, 0x0E, 0x11, 0x11, 0x11, 0x0E], - "p": [0x00, 0x00, 0x1E, 0x11, 0x11, 0x1E, 0x10], - "q": [0x00, 0x00, 0x0F, 0x11, 0x11, 0x0F, 0x01], - "r": [0x00, 0x00, 0x16, 0x09, 0x08, 0x08, 0x08], - "s": [0x00, 0x00, 0x0F, 0x10, 0x0E, 0x01, 0x1E], - "t": [0x08, 0x08, 0x1E, 0x08, 0x08, 0x08, 0x06], - "u": [0x00, 0x00, 0x11, 0x11, 0x11, 0x13, 0x0D], - "v": [0x00, 0x00, 0x11, 0x11, 0x11, 0x0A, 0x04], - "w": [0x00, 0x00, 0x11, 0x15, 0x15, 0x15, 0x0A], - "x": [0x00, 0x00, 0x11, 0x0A, 0x04, 0x0A, 0x11], - "y": [0x00, 0x00, 0x11, 0x11, 0x0F, 0x01, 0x1E], - "z": [0x00, 0x00, 0x1F, 0x02, 0x04, 0x08, 0x1F], - "{": [0x06, 0x08, 0x08, 0x04, 0x08, 0x08, 0x06], - "|": [0x04, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04], - "}": [0x0C, 0x02, 0x02, 0x04, 0x02, 0x02, 0x0C], - "~": [0x00, 0x00, 0x08, 0x15, 0x02, 0x00, 0x00], -} - -_GLYPH_WIDTH = 5 -_GLYPH_HEIGHT = 7 -_TEXT_SCALE = 6 # render each font pixel as a 6x6 block -_TEXT_MARGIN = 60 # horizontal padding when wrapping -_TEXT_COLOR = (255, 255, 255) -_STROKE_COLOR = (0, 0, 0) -_STROKE_WIDTH = 3 # outline thickness in pixels -_SHADOW_OFFSET = 4 # drop shadow shift in pixels (down-right) -_SHADOW_DARKEN = 0.55 # shadow keeps 55% of the background color - - -def _random_light_color(rng: random.Random) -> Tuple[int, int, int]: - """A random pastel: any hue, low saturation, high value.""" - hue = rng.random() - saturation = rng.uniform(0.25, 0.55) - value = rng.uniform(0.82, 0.95) - r, g, b = hsv_to_rgb(hue, saturation, value) - return int(r * 255), int(g * 255), int(b * 255) - - -def _lerp(a: int, b: int, t: float) -> int: - return int(round(a + (b - a) * t)) - - -def _text_width(text: str) -> int: - """Pixel width of ``text`` at the rendered scale (spaces count too).""" - if not text: - return 0 - return (len(text) * (_GLYPH_WIDTH + 1) - 1) * _TEXT_SCALE - - -def _wrap_title(title: str, max_width: int) -> List[str]: - """Word-wrap ``title`` into lines that fit ``max_width`` pixels.""" - words = title.split() - if not words: - return [] - lines: List[str] = [] - current = "" - for word in words: - candidate = f"{current} {word}".strip() - if _text_width(candidate) <= max_width or not current: - current = candidate - else: - lines.append(current) - current = word - if current: - lines.append(current) - return lines - - -def _set_pixel(pixels: List[List[Tuple[int, int, int]]], x: int, y: int, - color: Tuple[int, int, int], darken: Optional[float] = None) -> None: - """Set one pixel: solid ``color``, or darken the existing pixel by ``darken``.""" - if not 0 <= y < len(pixels): - return - if not 0 <= x < len(pixels[y]): - return - if darken is None: - pixels[y][x] = color - else: - r, g, b = pixels[y][x] - pixels[y][x] = (int(r * darken), int(g * darken), int(b * darken)) - - -def _render_line(pixels: List[List[Tuple[int, int, int]]], text: str, x0: int, y0: int, - color: Tuple[int, int, int] = (0, 0, 0), - darken: Optional[float] = None) -> None: - """Blit one line of bitmap text onto the pixel grid in place. - - With ``darken``, each glyph pixel darkens whatever is underneath it - instead of painting a solid color (used for the drop shadow, which - stays tinted by the gradient behind it). - """ - for char_index, char in enumerate(text): - glyph = _FONT.get(char) - if glyph is None: - continue - x_off = x0 + char_index * (_GLYPH_WIDTH + 1) * _TEXT_SCALE - for gy, bits in enumerate(glyph): - for gx in range(_GLYPH_WIDTH): - if not bits & (1 << (_GLYPH_WIDTH - 1 - gx)): - continue - for dy in range(_TEXT_SCALE): - for dx in range(_TEXT_SCALE): - _set_pixel(pixels, - x_off + gx * _TEXT_SCALE + dx, - y0 + gy * _TEXT_SCALE + dy, - color, darken) - - -def _encode_png(width: int, height: int, pixels: List[List[Tuple[int, int, int]]]) -> bytes: - """Encode an RGB pixel grid as a PNG using only the stdlib.""" - def chunk(chunk_type: bytes, data: bytes) -> bytes: - payload = chunk_type + data - return (struct.pack(">I", len(data)) + payload - + struct.pack(">I", zlib.crc32(payload) & 0xFFFFFFFF)) - - raw = b"".join( - b"\x00" + bytes(channel for pixel in scanline for channel in pixel) - for scanline in pixels - ) - ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8-bit RGB - return (b"\x89PNG\r\n\x1a\n" - + chunk(b"IHDR", ihdr) - + chunk(b"IDAT", zlib.compress(raw, 9)) - + chunk(b"IEND", b"")) - - -def generate_cover(title: str, path: Path, width: int = COVER_WIDTH, - height: int = COVER_HEIGHT, seed: Optional[int] = None) -> Optional[Path]: - """Write a gradient cover PNG with ``title`` centered on it. - - ``seed`` makes the gradient reproducible (used by tests). Returns the - path on success, or None when the PNG cannot be written (the audiobook - still gets tags, just without cover art). - """ - rng = random.Random(seed) - top_color = _random_light_color(rng) - bottom_color = _random_light_color(rng) - - pixels = [] - for y in range(height): - t = y / max(1, height - 1) - color = (_lerp(top_color[0], bottom_color[0], t), - _lerp(top_color[1], bottom_color[1], t), - _lerp(top_color[2], bottom_color[2], t)) - pixels.append([color] * width) - - lines = _wrap_title((title or "").strip(), width - 2 * _TEXT_MARGIN) - if lines: - line_height = _GLYPH_HEIGHT * _TEXT_SCALE + _TEXT_SCALE * 3 - total_height = len(lines) * line_height - y_start = max(0, (height - total_height) // 2) - for line_index, line in enumerate(lines): - x0 = (width - _text_width(line)) // 2 - y0 = y_start + line_index * line_height - # Paint order: drop shadow (the outlined glyph's full silhouette - # — glyph dilated by the stroke width — shifted down-right and - # darkened, tinted by the gradient underneath), then the black - # outline traced around the glyph, then the solid white text. - for dx in range(-_STROKE_WIDTH, _STROKE_WIDTH + 1): - for dy in range(-_STROKE_WIDTH, _STROKE_WIDTH + 1): - _render_line(pixels, line, x0 + dx + _SHADOW_OFFSET, - y0 + dy + _SHADOW_OFFSET, darken=_SHADOW_DARKEN) - for dx in range(-_STROKE_WIDTH, _STROKE_WIDTH + 1): - for dy in range(-_STROKE_WIDTH, _STROKE_WIDTH + 1): - _render_line(pixels, line, x0 + dx, y0 + dy, - color=_STROKE_COLOR) - _render_line(pixels, line, x0, y0, color=_TEXT_COLOR) - - try: - path.write_bytes(_encode_png(width, height, pixels)) - except OSError as exc: - logger.warning("Could not write cover image %s: %s", path, exc) - return None - logger.info("Generated cover: %s (gradient %s -> %s)", path, top_color, bottom_color) - return path diff --git a/converter/extractors.py b/converter/extractors.py deleted file mode 100644 index a564333..0000000 --- a/converter/extractors.py +++ /dev/null @@ -1,328 +0,0 @@ -"""Text extraction from book files (TXT, PDF, EPUB) and text/HTML cleaning.""" - -import codecs -import logging -import re -import zipfile -from html import unescape -from pathlib import Path -from typing import List, NamedTuple - -try: - from bs4 import BeautifulSoup - BS4_AVAILABLE = True -except ImportError: - BS4_AVAILABLE = False - -logger = logging.getLogger(__name__) - - -class Section(NamedTuple): - """A titled chunk of a book (e.g. an EPUB chapter).""" - - title: str - text: str - - -class Book(NamedTuple): - """A book's metadata plus its titled sections.""" - - title: str - author: str - sections: List[Section] - - -def extract_text(file_path: Path) -> str: - """Extract text from a book file based on its extension.""" - extension = file_path.suffix.lower() - if extension == ".txt": - return _extract_txt(file_path) - if extension == ".pdf": - return _extract_pdf(file_path) - if extension == ".epub": - return extract_epub(file_path) - raise ValueError(f"Unsupported file format: {extension}") - - -def extract_sections(file_path: Path) -> List[Section]: - """Extract the book's text as titled sections (chapters). - - EPUB files are split on their spine documents so they can be converted - one chapter at a time. TXT and PDF files have no chapter structure and - always yield a single section. - """ - if file_path.suffix.lower() == ".epub": - chapters = _extract_epub_chapters(file_path) - if not chapters: - raise RuntimeError("All EPUB extraction methods failed") - return chapters - - return [Section(file_path.stem, extract_text(file_path))] - - -def extract_book(file_path: Path) -> Book: - """Extract sections plus book-level metadata (title, author). - - EPUB and PDF files carry embedded metadata; missing fields (and TXT - files, which have none) fall back to the file stem for the title and - an empty author. - """ - title, author = "", "" - extension = file_path.suffix.lower() - if extension == ".epub": - title, author = _epub_metadata(file_path) - elif extension == ".pdf": - title, author = _pdf_metadata(file_path) - return Book(title or file_path.stem, author.strip(), extract_sections(file_path)) - - -def _epub_metadata(file_path: Path) -> tuple: - """Return (title, author) from an EPUB's Dublin Core metadata.""" - try: - import ebooklib - from ebooklib import epub - - book = epub.read_epub(str(file_path)) - title = _first_dc_value(book.get_metadata("DC", "title")) - author = _first_dc_value(book.get_metadata("DC", "creator")) - return title, author - except Exception as exc: - logger.warning("Could not read EPUB metadata: %s", exc) - return "", "" - - -def _pdf_metadata(file_path: Path) -> tuple: - """Return (title, author) from a PDF's document info dictionary.""" - try: - from pypdf import PdfReader - - reader = PdfReader(str(file_path)) - info = reader.metadata or {} - title = str(info.get("/Title") or "") - author = str(info.get("/Author") or "") - return title, author - except Exception as exc: - logger.warning("Could not read PDF metadata: %s", exc) - return "", "" - - -def _first_dc_value(entries) -> str: - """First value of an ebooklib DC metadata list: [(value, ...), ...].""" - if not entries: - return "" - value = entries[0][0] - return str(value).strip() if value else "" - - -def _extract_epub_chapters(file_path: Path) -> List[Section]: - """Return one Section per EPUB spine document (chapter), in reading order.""" - import ebooklib - - book = None - for method in (_read_epub_ebooklib, _read_epub_zipfile, _read_epub_manual): - try: - book = method(file_path) - except Exception as exc: - logger.warning("EPUB chapter method %s failed: %s", method.__name__, exc) - continue - if book: - break - - if book is None: - return [] - - chapters = [] - for title, text in book: - cleaned = clean_html(text) - if cleaned.strip(): - chapters.append(Section(title or file_path.stem, cleaned)) - return chapters - - -def _toc_titles(book) -> dict: - """Flatten an ebooklib TOC into a ``{href: title}`` mapping.""" - titles = {} - - def walk(nodes) -> None: - for node in nodes: - if isinstance(node, (tuple, list)): - walk(node[1] if len(node) > 1 else []) - continue - href = getattr(node, "href", None) - title = getattr(node, "title", None) - if href and title: - titles[href.split("#")[0]] = title - - walk(book.toc) - return titles - - -def _read_epub_ebooklib(file_path: Path): - """Read EPUB spine documents as (title, html) pairs via ebooklib.""" - import ebooklib - from ebooklib import epub - - book = epub.read_epub(str(file_path)) - titles = _toc_titles(book) - items = [] - for entry in book.spine: - item_id = entry[0] if isinstance(entry, (tuple, list)) else entry - try: - item = book.get_item_with_id(item_id) - except Exception as exc: - logger.debug("Skipping EPUB spine item %r: %s", item_id, exc) - continue - if not item or item.get_type() != ebooklib.ITEM_DOCUMENT: - continue - if isinstance(item, epub.EpubNav): - continue - content = item.get_body_content() - if content: - if isinstance(content, bytes): - content = content.decode("utf-8", errors="ignore") - title = (titles.get(item.file_name) - or titles.get(item.get_name()) - or getattr(item, "title", None) - or item.get_name()) - items.append((title, str(content))) - return items - - -def _read_epub_zipfile(file_path: Path): - """Read EPUB HTML members as (title, html) pairs, ordered by filename.""" - items = [] - with zipfile.ZipFile(file_path, "r") as epub_zip: - for file_name in sorted(epub_zip.namelist(), key=_natural_key): - if file_name.lower().endswith((".html", ".xhtml", ".htm")): - try: - content = epub_zip.read(file_name).decode("utf-8", errors="ignore") - items.append((Path(file_name).stem, content)) - except Exception as exc: - logger.debug("Skipping EPUB member %r: %s", file_name, exc) - return items - - -def _read_epub_manual(file_path: Path): - """Last-resort read of any markup-looking EPUB member.""" - skipped_extensions = (".jpg", ".jpeg", ".png", ".gif", ".css", ".js") - items = [] - with zipfile.ZipFile(file_path, "r") as epub_zip: - for file_name in sorted(epub_zip.namelist(), key=_natural_key): - if file_name.lower().endswith(skipped_extensions): - continue - try: - content = epub_zip.read(file_name).decode("utf-8", errors="ignore") - if "<" in content and len(content.strip()) > 100: - items.append((Path(file_name).stem, content)) - except Exception as exc: - logger.debug("Skipping EPUB member %r: %s", file_name, exc) - return items - - -def clean_text(text: str) -> str: - """Normalize whitespace and strip standalone page numbers. - - Page numbers are removed only when they appear as a short number alone on - its own line (before whitespace collapsing), so inline numbers like - "42 years", "1,000" or "3.5" are preserved. - """ - if not text: - return "" - # Standalone page numbers (digits alone on a line) must go BEFORE the - # newline-collapsing step below. - text = re.sub(r"(?m)^\s*\d{1,4}\s*$", " ", text) - text = re.sub(r"\s+", " ", text) - return text.strip() - - -def clean_html(html_content: str) -> str: - """Strip markup, scripts and styles from HTML content.""" - if not html_content: - return "" - - if BS4_AVAILABLE: - try: - soup = BeautifulSoup(html_content, "html.parser") - for tag in soup(["script", "style"]): - tag.decompose() - text = soup.get_text(separator=" ") - return re.sub(r"\s+", " ", text).strip() - except Exception as exc: - logger.debug("BeautifulSoup cleaning failed, falling back to regex: %s", exc) - - # Fallback regex cleaning - html_content = re.sub(r"<style[^>]*>.*?</style>", "", html_content, flags=re.DOTALL | re.IGNORECASE) - html_content = re.sub(r"<script[^>]*>.*?</script>", "", html_content, flags=re.DOTALL | re.IGNORECASE) - html_content = re.sub(r"<[^>]+>", " ", html_content) - html_content = unescape(html_content) - html_content = re.sub(r"\s+", " ", html_content) - return html_content.strip() - - -def extract_epub(file_path: Path) -> str: - """Extract the book's text from EPUB, trying several methods in order.""" - chapters = _extract_epub_chapters(file_path) - if not chapters: - raise RuntimeError("All EPUB extraction methods failed") - return "\n\n".join(section.text for section in chapters) - - -def _natural_key(name: str): - """Sort key that orders numeric runs numerically (chapter2 before chapter10).""" - return [int(part) if part.isdigit() else part.lower() - for part in re.split(r"(\d+)", name)] - - -def _extract_txt(file_path: Path) -> str: - """Extract from TXT, handling BOMs and common encodings (latin-1 is the catch-all). - - UTF-16 files without a BOM are detected via NUL bytes; otherwise they would - silently decode as NUL-interleaved UTF-8 or cp1252/latin-1 garbage. - """ - data = file_path.read_bytes() - - if data.startswith((codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE)): - return clean_text(data.decode("utf-32")) - if data.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)): - return clean_text(data.decode("utf-16")) - if data.startswith(codecs.BOM_UTF8): - return clean_text(data.decode("utf-8-sig")) - - # No BOM: UTF-16 without BOM is common on Windows; detect via NUL bytes. - sample = data[:4096] - even_nuls = sum(1 for i, b in enumerate(sample) if b == 0 and i % 2 == 0) - odd_nuls = sum(1 for i, b in enumerate(sample) if b == 0 and i % 2 == 1) - if even_nuls or odd_nuls: - encoding = "utf-16-be" if even_nuls > odd_nuls else "utf-16-le" - return clean_text(data.decode(encoding)) - - for encoding in ("utf-8", "cp1252", "latin-1"): - try: - return clean_text(data.decode(encoding)) - except UnicodeError: - continue - raise ValueError(f"Could not decode text file: {file_path}") - - -def _extract_pdf(file_path: Path) -> str: - """Extract from PDF.""" - from pypdf import PdfReader - - text = "" - with open(file_path, "rb") as file: - pdf_reader = PdfReader(file) - total_pages = len(pdf_reader.pages) - logger.info("PDF has %d pages", total_pages) - - for page_num, page in enumerate(pdf_reader.pages, 1): - try: - page_text = page.extract_text() or "" - if page_text.strip(): - text += f"\n\n{page_text}" - if page_num % 10 == 0: - logger.debug("Extracted %d/%d pages", page_num, total_pages) - except Exception as exc: - logger.warning("Failed to extract page %d: %s", page_num, exc) - - logger.info("Extracted text from %d pages, %d characters total", total_pages, len(text)) - return clean_text(text) diff --git a/converter/tts.py b/converter/tts.py deleted file mode 100644 index f5d54e2..0000000 --- a/converter/tts.py +++ /dev/null @@ -1,1305 +0,0 @@ -"""Client wrappers for the TTS backends. - -QwenTTSClient talks to the Qwen3-TTS demo server (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). -AudioCppTTSClient talks to the audiocpp_server from the audio.cpp -repository, which can host any TTS model family audio.cpp supports -(Qwen3-TTS, Higgs Audio, VoxCPM2, IndexTTS2, ...) through one OpenAI-style -API; the family is detected from the server at startup (see the -"audio.cpp backend" sections of the README). -""" - -import contextlib -import io -import json -import logging -import random -import shutil -import sys -import tempfile -import threading -import time -import urllib.error -import urllib.parse -import urllib.request -import wave -from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple - -from . import config -from .audio import concat_audio_files -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) - -# TTS backends (re-exported for the CLI and the converter orchestrator). -BACKEND_QWEN = "qwen" -BACKEND_FASTER = "faster" -BACKEND_AUDIOCPP = "audiocpp" -BACKENDS = (BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER) - -# 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", -} - -# Qwen display names -> ISO 639-1 codes, for audio.cpp families whose -# language request option takes a code instead of a display name. "Auto" -# has no code and maps to None so the field is omitted and the server -# applies its own default. -LANGUAGE_ISO_CODES = { - "Chinese": "zh", - "English": "en", - "German": "de", - "Italian": "it", - "Portuguese": "pt", - "Spanish": "es", - "Japanese": "ja", - "Korean": "ko", - "French": "fr", - "Russian": "ru", -} - -# --- audio.cpp model families --------------------------------------------- -# -# audiocpp_server exposes the same OpenAI-style API for every TTS family it -# hosts; families only differ in a few request conventions, captured here as -# profiles. Families that are not listed use the default profile below. - -# How the "language" request field is expressed by a family. -AUDIOCPP_LANG_DISPLAY = "display" # Qwen display names, e.g. "English" -AUDIOCPP_LANG_ISO = "iso" # ISO 639-1 codes, e.g. "en" -AUDIOCPP_LANG_OMIT = "omit" # no language field; the model detects it - -# The only family with a built-in speaker mode (CustomVoice speaker names -# plus the INSTRUCT style prompt). Every other family is clone-only: the -# voice comes from a server-side preset requested with --voice. -AUDIOCPP_FAMILY_QWEN3_TTS = "qwen3_tts" - -# Server model entry tasks this client can synthesize audiobooks with, -# taken from GET /v1/models (the "task" field of each entry; servers that -# predate the field reported TTS models only, so a missing task is treated -# as "tts"). "vdes" entries are voice design models: the voice is described -# with --instructions instead of coming from a speaker or a reference clip. -# Entries with any other task (asr, vc, diar, ...) are rejected at connect -# time with a hint to pick a synthesis entry. -AUDIOCPP_TASK_TTS = "tts" -AUDIOCPP_TASK_VDES = "vdes" -AUDIOCPP_SYNTHESIS_TASKS = (AUDIOCPP_TASK_TTS, "clon", AUDIOCPP_TASK_VDES) - - -class AudioCppFamilyProfile: - """Request conventions of one audio.cpp model family.""" - - def __init__(self, language_style: str = AUDIOCPP_LANG_OMIT, - sends_instructions: bool = False, - builtin_speakers: bool = False): - self.language_style = language_style - self.sends_instructions = sends_instructions - self.builtin_speakers = builtin_speakers - - -# Generic profile for families not listed in AUDIOCPP_FAMILY_PROFILES: -# clone-only, no style instructions, and no language field (the model -# detects the language itself). Describes higgs_audio_tts, voxcpm2, -# fish_audio, dots_tts, dramabox, omnivoice, outetts, glm_tts, miotts, -# moss_tts_*, pocket_tts, vibevoice, ... as well as families added to -# audio.cpp after this table was written. -AUDIOCPP_DEFAULT_FAMILY_PROFILE = AudioCppFamilyProfile() - -AUDIOCPP_FAMILY_PROFILES = { - AUDIOCPP_FAMILY_QWEN3_TTS: AudioCppFamilyProfile( - language_style=AUDIOCPP_LANG_DISPLAY, - sends_instructions=True, - builtin_speakers=True, - ), - # Families whose language option takes a code (e.g. "en") instead of - # a Qwen display name; otherwise clone-only like the default profile. - "chatterbox": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "confucius4_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "index_tts2": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "magpie_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "supertonic": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), -} - -# 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 _resolve_request_seed() -> int: - """Resolve the seed sent with every request. - - Returns config.SEED as-is, or (with CONSTANT_SEED and SEED < 0) one - random value drawn per run, meant to be reused for every request so - the voice stays consistent across chunk boundaries. Without - CONSTANT_SEED, -1 is returned so the server re-samples the voice on - every generation. - """ - seed = config.SEED - if config.CONSTANT_SEED and seed < 0: - seed = random.randrange(2 ** 31) - return seed - - -def speaker_display_name() -> str: - """Return the display name for the configured custom speaker.""" - return SPEAKER_DISPLAY_NAMES.get( - config.SPEAKER.lower(), config.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 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. - """ - if value is None: - raise ValueError("Language must not be None") - candidate = value.strip() - if not candidate: - raise ValueError("Language must not be empty") - for name in TTS_LANGUAGES: - if candidate.lower() == name.lower(): - return name - alias = TTS_LANGUAGE_ALIASES.get(candidate.lower()) - if alias: - return alias - raise ValueError( - f"Unknown language: {value!r}. Expected one of " - f"{', '.join(TTS_LANGUAGES)} (or an alias: " - f"{', '.join(sorted(TTS_LANGUAGE_ALIASES))})." - ) - - -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 - - -def whisper_backend_available() -> Optional[str]: - """Return the name of an importable Whisper backend, or None. - - Checks faster_whisper first (preferred), then the openai-whisper - package, without importing the heavy model code: a bare import probe - is enough to tell whether the package is installed in the current - environment. Used by the make_audiocpp_server_json tool to warn when - neither is present (e.g. the wrong conda environment is active). - """ - for backend in ("faster_whisper", "whisper"): - try: - __import__(backend) - except ImportError: - continue - return backend - return None - - -# 150 wpm is a typical spoken pace; used only to size the HTTP request -# timeout for long audio.cpp generations (not as a correctness check). -_ESTIMATED_WORDS_PER_MINUTE = 150 - - -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 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 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. - - Returns the generated chunk file's path, or None when all attempts - failed. - """ - 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, label: Optional[str] = None): - """Print a periodic "still working" message while a request generates. - - ``label`` overrides the default "Chunk {chunk_num}" subject, for - backends that send one request per chapter without client-side - chunking (the audio.cpp default) where "chunk" would be misleading. - """ - stop = threading.Event() - subject = label if label is not None else f"Chunk {chunk_num}" - - def _beat(): - start = time.time() - while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): - elapsed = time.time() - start - print(f"[...] {subject} 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 demo server.""" - - 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 VOICE_MODES: - raise ValueError( - 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 - # Seed sent with every request: config.SEED as-is, or (with - # CONSTANT_SEED and SEED < 0) one random value drawn per run and - # reused for every request so the voice stays consistent across - # chunk boundaries. Without CONSTANT_SEED, -1 is forwarded so the - # server re-samples the voice on every generation. - self._seed = _resolve_request_seed() - if language is None: - language = config.LANGUAGE - # Validate before connecting so bad values fail fast without a server. - self.language = normalize_language(language) - self.client = None - self.api_info: Dict[str, Any] = {} - self.clone_client = None - self.clone_api_info: Dict[str, Any] = {} - self._ref_audio_filedata: Optional[Dict[str, Any]] = None - self._connect() - - # ------------------------------------------------------------------ - # Connection - # ------------------------------------------------------------------ - - def _connect(self) -> None: - api_url = config.CLONE_API_URL if self.voice_mode == VOICE_MODE_CLONE else config.QWEN_API_URL - try: - 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.CLONE_API_URL, clone=True) - print(f"[OK] Connected to Voice Clone API at {config.CLONE_API_URL}") - self._resolve_reference_text() - else: - self._init_client(config.QWEN_API_URL, clone=False) - print("[OK] Connected to Qwen API") - except Exception as exc: - raise RuntimeError( - f"Qwen API initialization failed at {api_url}: {exc}. " - "Make sure the Qwen demo server is running and reachable, and that your " - "installed Qwen3-TTS version matches this converter's API expectations " - "(voice clone requires the Base-model demo: Qwen/Qwen3-TTS-12Hz-1.7B-Base)." - ) from exc - - def _resolve_reference_text(self) -> None: - """Resolve the reference transcript: explicit text, then local - transcription, then x-vector-only mode.""" - if not self.voice_clone_ref_text and self.voice_clone_ref_audio: - if self.skip_transcription: - print("[INFO] Skipping reference audio transcription (--no-transcription).") - else: - print("[INFO] Transcribing reference audio for voice cloning...") - self.voice_clone_ref_text = self.transcribe_audio(self.voice_clone_ref_audio) or "" - if not self.voice_clone_ref_text: - print("[WARNING] No reference text available; using x-vector-only clone mode (lower quality).") - print(' Pass --transcription "..." for higher-quality in-context cloning.') - else: - print(f"[OK] Reference text:\n{self.voice_clone_ref_text}") - - def _init_client(self, url: str, clone: bool = False) -> None: - """Initialize a Gradio client and store its API metadata. - - gradio_client prints its usage info directly to stdout while the - client is created and its API metadata loaded, so stdout is swapped - for a buffer for the whole process; the captured text is re-emitted - at DEBUG level for troubleshooting. - """ - from gradio_client import Client - - logger.info("Connecting to Qwen API at %s...", url) - old_stdout = sys.stdout - captured = io.StringIO() - sys.stdout = captured - try: - try: - client = Client(url, httpx_kwargs={"timeout": config.API_TIMEOUT}) - except TypeError: - # Older gradio_client versions don't support httpx_kwargs. - client = Client(url) - if clone: - self.clone_client = client - self.clone_api_info = self._load_api_info(client) - else: - self.client = client - self.api_info = self._load_api_info(client) - finally: - sys.stdout = old_stdout - usage_info = captured.getvalue().strip() - if usage_info: - logger.debug("Gradio client output for %s:\n%s", url, usage_info) - logger.info("Connected to Qwen API") - - @staticmethod - def _load_api_info(client) -> Dict[str, Any]: - """Load available API metadata from the Gradio app.""" - try: - return client.view_api(return_format="dict") - except Exception as exc: - logger.warning("Unable to read API metadata: %s", exc) - return {} - - def _resolve_api_name(self, *candidates: str, api_info: Optional[Dict[str, Any]] = None) -> str: - """Return the first available api_name from candidate list.""" - info = api_info if api_info is not None else self.api_info - named_endpoints = info.get("named_endpoints", {}) - for candidate in candidates: - if candidate in named_endpoints: - return candidate - return candidates[0] - - def _endpoint_accepts_param(self, api_name: str, param_name: str, - api_info: Optional[Dict[str, Any]] = None) -> bool: - """Check whether endpoint input schema includes the given parameter.""" - info = api_info if api_info is not None else self.api_info - endpoint = info.get("named_endpoints", {}).get(api_name, {}) - parameters = endpoint.get("parameters", []) - return any(parameter.get("parameter_name") == param_name for parameter in parameters) - - # ------------------------------------------------------------------ - # Reference audio transcription (voice clone) - # ------------------------------------------------------------------ - - def transcribe_audio(self, audio_path: str) -> Optional[str]: - """Transcribe reference audio locally using an optional Whisper backend.""" - return transcribe_reference_audio(audio_path) - - # ------------------------------------------------------------------ - # Chunk generation - # ------------------------------------------------------------------ - - def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder. - - The text is split into sub-requests of at most - ``config.CHUNK_SIZE`` words each (the book-level chunker - normally guarantees this already; the split is defense in depth - against pathological input such as a punctuation-free run of - text), and the audio files returned for the sub-requests are - concatenated into one chunk file. - """ - try: - sub_texts = split_into_chunks(text, max_words=config.CHUNK_SIZE) - if not sub_texts: - raise RuntimeError("No text to synthesize") - - output_path: Optional[Path] = None - with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ - self._chunk_heartbeat(chunk_num): - part_paths = [ - self._generate_sub_request(sub_text, parts_dir, sub_num, - len(sub_texts), chunk_num) - for sub_num, sub_text in enumerate(sub_texts, 1) - ] - if len(part_paths) == 1: - suffix = part_paths[0].suffix or ".wav" - output_path = self._chunk_path(chunk_num, suffix) - shutil.copy2(part_paths[0], output_path) - else: - output_path = self._chunk_path(chunk_num, ".wav") - concat_audio_files(part_paths, output_path) - - logger.debug("Chunk %d generated successfully (%d sub-request(s))", - chunk_num, len(sub_texts)) - return str(output_path) - - except Exception as exc: - logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) - return None - - def _generate_sub_request(self, text: str, parts_dir: str, sub_num: int, - sub_total: int, chunk_num: int) -> Path: - """Run one API generation for ``text``; returns the downloaded audio.""" - if sub_total > 1: - logger.info("Chunk %d: oversized input split into %d requests " - "(sub-request %d/%d)", chunk_num, sub_total, sub_num, sub_total) - if self.voice_mode == VOICE_MODE_CUSTOM: - result = self._generate_custom_voice(text) - elif self.voice_mode == VOICE_MODE_CLONE: - result = self._generate_voice_clone(text) - else: - raise ValueError(f"Unknown voice mode: {self.voice_mode}") - - if not isinstance(result, (tuple, list)) or not result: - raise RuntimeError("Qwen API returned an invalid result") - - audio_path = result[0] # First element is the audio file path - if not isinstance(audio_path, (str, Path)) or not audio_path: - raise RuntimeError("Qwen API did not return an audio file path") - - source = Path(audio_path) - if not source.exists(): - raise RuntimeError(f"Generated audio file not found: {audio_path}") - - destination = Path(parts_dir) / f"part_{sub_num:02d}{source.suffix or '.wav'}" - shutil.copy2(source, destination) - - return destination - - # ------------------------------------------------------------------ - # API payloads - # ------------------------------------------------------------------ - - def _generate_custom_voice(self, text: str) -> Tuple: - """Generate audio using CustomVoice mode.""" - custom_api = self._resolve_api_name("/run_instruct", "/run_custom_voice", "/generate_custom_voice") - if custom_api == "/run_instruct": - payload = dict( - text=text, - lang_disp=self.language, - spk_disp=speaker_display_name(), - instruct=config.INSTRUCT, - ) - else: - payload = dict( - text=text, - language=self.language, - speaker=config.SPEAKER, - instruct=config.INSTRUCT, - ) - if self._endpoint_accepts_param(custom_api, "model_id_cv"): - payload["model_id_cv"] = CUSTOM_VOICE_MODEL_ID - elif self._endpoint_accepts_param(custom_api, "model_size"): - payload["model_size"] = MODEL_SIZE - - if self._endpoint_accepts_param(custom_api, "seed"): - payload["seed"] = self._seed - - return self.client.predict(**payload, api_name=custom_api) - - def _ref_audio_payload(self) -> Dict[str, Any]: - """Gradio file payload for the reference audio (built once, reused).""" - if self._ref_audio_filedata is None: - from gradio_client import handle_file - self._ref_audio_filedata = handle_file(self.voice_clone_ref_audio) - return self._ref_audio_filedata - - def _generate_voice_clone(self, text: str) -> Tuple: - """Generate audio using Voice Clone mode.""" - if not Path(self.voice_clone_ref_audio).exists(): - raise FileNotFoundError(f"Reference audio not found: {self.voice_clone_ref_audio}") - - if self.clone_client is None: - raise RuntimeError("Voice Clone client is not initialized. Is the Base-model demo running?") - - clone_api = self._resolve_api_name("/run_voice_clone", "/generate_voice_clone", - api_info=self.clone_api_info) - use_xvector = config.XVECTOR_ONLY or not self.voice_clone_ref_text - - if clone_api == "/run_voice_clone": - payload = dict( - ref_aud=self._ref_audio_payload(), - ref_txt=self.voice_clone_ref_text, - use_xvec=use_xvector, - text=text, - lang_disp=self.language, - ) - else: - payload = dict( - ref_audio=self._ref_audio_payload(), - ref_text=self.voice_clone_ref_text, - target_text=text, - language=self.language, - use_xvector_only=use_xvector, - ) - optional_params = { - "model_size": MODEL_SIZE, - "seed": self._seed, - } - for name, value in optional_params.items(): - if self._endpoint_accepts_param(clone_api, name, api_info=self.clone_api_info): - 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 Qwen 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_VOICE - self.api_url = (api_url or config.FASTER_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.API_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.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.MAX_RETRIES - 1: - time.sleep(2 + 2 * attempt) - raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after " - f"{config.MAX_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.CHUNK_SIZE) - 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 = self._request_pcm_with_retry( - sub_text, chunk_num, sub_num, len(sub_chunks)) - pcm_parts.append(pcm) - - 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(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 - - -class AudioCppTTSClient(_BaseTTSClient): - """Generates audio chunks through an audio.cpp audiocpp_server. - - Talks to the OpenAI-style HTTP API of audiocpp_server, which hosts TTS - model families through a native ggml runtime (GGUF weights, no Python - serving stack). The server API is family-agnostic; the family and task - of the configured model entry are read from GET /v1/models at startup - and adapt the request payload (language field style, style instructions) - through AUDIOCPP_FAMILY_PROFILES. Three voice modes, all resolved - server-side from the request's "voice"/"instructions" fields: - - - Speaker mode (no ``voice``): Qwen3-TTS only. A built-in CustomVoice - speaker name (e.g. "Vivian") is passed through, plus the INSTRUCT - style prompt. The server must be configured with the CustomVoice - model for this. Families without built-in speakers reject this mode - with a hint to pass --voice (or --instructions, see below). - - Preset mode (``voice=NAME``): a voice configured on the server - (``voice_presets`` or ``voice_dir`` in its config, e.g. a cloning - reference). The name is validated against GET /v1/audio/voices at - startup because an unresolvable name would silently fall back to - plain TTS on a clone-based model instead of failing. When - AUDIOCPP_CLONE_MODEL_ID names a second server entry of the same - family (typically the Qwen Base model), preset requests are routed - to it. Only the entry actually used needs to exist on the server: a - clone-only (Base) server works for --voice runs, while speaker mode - on such a server fails with a hint to pass --voice. - - Voice design (task "vdes" entries, e.g. Qwen3-TTS VoiceDesign): the - voice is described in natural language through ``instructions``, - which is required and sent with every request (no ``voice`` field). - A constant per-run seed keeps the designed voice consistent across - chunk boundaries. - - ``instructions`` also works on non-design entries, where it acts as a - generic style/delivery instruction (voice control): families that read - it (OmniVoice, Qwen3-TTS CustomVoice, ...) shape the voice or delivery - accordingly, and others ignore it. On instruction-conditioned families - without built-in speakers it may replace --voice entirely (the - instruction defines the voice). Extra request options (``--option - KEY=VALUE``, e.g. emotion, voice_id, speed) are forwarded verbatim in - the request's "options" object, which is the server's generic - pass-through for per-model controls. - - Chunking: the server does its own long-form text chunking for every - family (its ``text_chunk_size`` option, with a per-family default), so - by default each chapter is sent as a single request and the audio - comes back already stitched. With ``chunk_text=True`` (the --chunk CLI - flag), text is instead split client-side into CHUNK_SIZE-word - sub-requests, which may needlessly double-chunk — the warning is - printed by the CLI. - - Each response is a complete WAV file, so sub-request audio is - concatenated with the same lossless path used for the Qwen client. - """ - - def __init__(self, voice: Optional[str] = None, language: Optional[str] = None, - api_url: Optional[str] = None, chunk_text: bool = False, - model_id: Optional[str] = None, - instructions: Optional[str] = None, - request_options: Optional[Dict[str, str]] = None): - self.api_url = (api_url or config.AUDIOCPP_API_URL).rstrip("/") - # Per-run model selection: the --model CLI flag overrides config; an - # empty value is resolved at connect time when the server hosts exactly - # one entry, so multi-model servers don't require editing config.py. - self.model_id = (model_id if model_id is not None - else config.AUDIOCPP_MODEL_ID) or "" - self._model_id_explicit = bool(self.model_id) - # Validate before connecting so bad values fail fast without a server. - self.language = normalize_language( - language if language is not None else config.LANGUAGE) - # One seed value per run, reused for every request (see - # _resolve_request_seed). Unlike the Qwen demo, audio.cpp has no - # negative "randomize" seed, so a negative value means "send no seed - # at all" (see _request_wav) and the server randomizes. - self._seed = _resolve_request_seed() - self.preset_mode = bool(voice) - self.voice = voice or speaker_display_name() - # Style/voice-design instruction sent with every request (the CLI - # --instructions flag overrides AUDIOCPP_INSTRUCTIONS in config.py). - # For task "vdes" entries it describes the voice to design; for other - # families it is a generic style instruction when the model reads one. - self.instructions = (instructions if instructions is not None - else config.AUDIOCPP_INSTRUCTIONS or "").strip() - # Free-form per-request options (--option KEY=VALUE) forwarded in the - # request's "options" object; models ignore keys they don't know. - self.request_options: Dict[str, str] = dict(request_options or {}) - # Both set during _connect once the entry's task is known: design_mode - # for "vdes" entries, instruction_voice when a family without built-in - # speakers gets its voice from the instruction alone (no voice field). - self.design_mode = False - self.instruction_voice = False - # When False (default), each chapter is sent as one request and the - # server does its own long-form chunking (text_chunk_size); when True, - # text is split client-side into CHUNK_SIZE-word sub-requests first. - self.chunk_text = bool(chunk_text) - # Family and task of the selected model entry and the family's request - # profile; all are resolved from GET /v1/models during _connect. - self.family = "" - self.task = AUDIOCPP_TASK_TTS - self.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE - self._connect() - - # ------------------------------------------------------------------ - # Connection - # ------------------------------------------------------------------ - - def _connect(self) -> None: - """Health-check the server and resolve the model, family, task, and voice. - - Speaker mode is only offered to families with built-in speakers - (Qwen3-TTS); every other family must select a server-side voice - with --voice or describe one with --instructions, so it fails fast - with a hint instead of silently synthesizing with a random default - voice. Voice design entries (task "vdes") require --instructions - and reject --voice. - """ - self._check_health() - models = self._list_models() - self._auto_pick_model_id(models) - if self.preset_mode: - self._select_model(models) - self._require_model_id(models) - self._resolve_family(models) - self._resolve_task(models) - if self.task not in AUDIOCPP_SYNTHESIS_TASKS: - available = ", ".join(model["id"] for model in models) or "none" - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' has task " - f"'{self.task}'; audiobook.py can only synthesize with TTS " - f"model entries (tasks {', '.join(AUDIOCPP_SYNTHESIS_TASKS)}). " - f"Pick a synthesis entry with --model (available: {available})." - ) - if self.design_mode: - if self.preset_mode: - raise RuntimeError( - f"--voice cannot be used with the voice design model " - f"'{self.model_id}': the voice is described by the " - "--instructions text instead (see README).") - if not self.instructions: - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' (family " - f"'{self.family}') is a voice design model: pass a " - "description of the voice to synthesize with, e.g. " - '--instructions "A warm adult female narrator with a ' - 'British accent" (see README).') - print(f"[OK] Connected to audio.cpp server at {self.api_url} " - f"(model '{self.model_id}', family '{self.family}', " - "voice design)") - print(f"[INFO] Designing the voice from: {self.instructions}") - elif self.preset_mode: - self._check_voice() - print(f"[OK] Connected to audio.cpp server at {self.api_url} " - f"(model '{self.model_id}', family '{self.family}', " - f"voice '{self.voice}')") - elif self.profile.builtin_speakers: - print(f"[OK] Connected to audio.cpp server at {self.api_url} " - f"(model '{self.model_id}', family '{self.family}', " - f"speaker '{self.voice}')") - print("[INFO] Speaker mode expects the server to be configured with the " - "CustomVoice model; with the Base model the speaker name is ignored " - "and a random default voice is used (see README).") - elif self.instructions: - # Families without built-in speakers can still get their voice - # from the instruction alone (e.g. OmniVoice voice design). - self.instruction_voice = True - print(f"[OK] Connected to audio.cpp server at {self.api_url} " - f"(model '{self.model_id}', family '{self.family}', " - "instruction voice)") - print(f"[INFO] Designing the voice from: {self.instructions}") - else: - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' (family " - f"'{self.family}') has no built-in speakers, so its voice " - "must come from the server: rerun with --voice NAME " - "matching a voice_preset or voice_dir entry in the server " - "config, or describe a voice with --instructions for " - "families that support it (see README).") - if self.instructions and not self.design_mode and not self.instruction_voice: - print(f"[INFO] Sending instruction with every request: {self.instructions}") - print("[INFO] Its effect (style, emotion, delivery) depends on the " - "model family; models without instruction support ignore it.") - - def _get_json(self, path: str, timeout: int = 10) -> Dict[str, Any]: - """GET a JSON document from the server.""" - url = f"{self.api_url}{path}" - try: - with urllib.request.urlopen(url, timeout=timeout) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - detail = "" - try: - detail = exc.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass - raise RuntimeError( - f"audio.cpp server returned HTTP {exc.code} for {path}: {detail}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"audio.cpp request failed for {path}: {exc.reason}") from exc - - def _check_health(self) -> None: - """Verify the server is reachable and reports healthy.""" - try: - payload = self._get_json("/health") - except Exception as exc: - raise RuntimeError( - f"audio.cpp server not reachable at {self.api_url}: {exc}. " - "Start audiocpp_server first (see the 'audio.cpp backend' " - "section of the README)." - ) from exc - if payload.get("status") != "ok": - raise RuntimeError( - f"The audio.cpp server at {self.api_url} reports status " - f"{payload.get('status')!r} instead of 'ok'") - - def _list_models(self) -> List[Dict[str, str]]: - """Fetch the (id, family, task) triples reported by the server.""" - try: - payload = self._get_json("/v1/models") - except Exception as exc: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} did not answer " - f"/v1/models: {exc}") from exc - entries = payload.get("data") or [] - models: List[Dict[str, str]] = [] - for entry in entries: - if isinstance(entry, dict) and entry.get("id"): - models.append({ - "id": entry["id"], - "family": entry.get("family") or "", - "task": entry.get("task") or "", - }) - return models - - def _auto_pick_model_id(self, models: List[Dict[str, str]]) -> None: - """Resolve an empty model id when the server hosts exactly one entry. - - Multi-model servers generated with several lazily-loaded entries can - be used without editing converter/config.py: leave AUDIOCPP_MODEL_ID - (and ``--model``) unset, and the single hosted entry is chosen - automatically. With more than one entry an explicit choice is required - (via ``--model`` or AUDIOCPP_MODEL_ID), since guessing would risk - synthesizing a whole book with the wrong family. - """ - if self.model_id: - return - if len(models) == 1: - self.model_id = models[0]["id"] - logger.info( - "AUDIOCPP_MODEL_ID is unset; using the only server entry '%s'", - self.model_id) - else: - logger.debug( - "AUDIOCPP_MODEL_ID is unset and the server hosts %d entries; " - "an explicit --model or config id is required", - len(models)) - - def _require_model_id(self, models: List[Dict[str, str]]) -> None: - """Verify the model id chosen for this run exists on the server. - - Speaker mode needs AUDIOCPP_MODEL_ID (the CustomVoice entry). - Preset mode validates whichever id _select_model resolved, so a - server hosting only a cloning model works for --voice. - """ - model_ids = [model["id"] for model in models] - if self.model_id and self.model_id in model_ids: - return - configured = ", ".join(model_ids) or "none" - if not self.model_id: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} hosts {len(model_ids)} " - f"model entries ({configured}); audiobook.py needs to know which " - "one to use. Pass --model <id> when converting, or set " - "AUDIOCPP_MODEL_ID in converter/config.py to one of them " - "(see README)." - ) - if self.preset_mode: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} has no model id " - f"'{self.model_id}' or clone model id " - f"'{config.AUDIOCPP_CLONE_MODEL_ID}' (configured: {configured}). " - "Add a TTS model entry for the family you want to the server " - "config and match AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID " - "in converter/config.py to its id, or select it per run with " - "--model (see README)." - ) - raise RuntimeError( - f"The audio.cpp server at {self.api_url} has no model id " - f"'{self.model_id}' (configured: {configured}). Speaker mode needs " - "the Qwen3-TTS CustomVoice model: add a qwen3_tts model entry to " - "the server config and match AUDIOCPP_MODEL_ID in converter/config.py to its " - "id (or pass --model), or rerun with --voice to use a voice preset " - "on any TTS model (see README)." - ) - - def _select_model(self, models: List[Dict[str, str]]) -> None: - """Pick the model for preset (cloning) requests. - - Defaults to the primary model id. When AUDIOCPP_CLONE_MODEL_ID is - configured and present on the server, preset requests are routed - to it instead, so one server can host the CustomVoice model for - speaker mode and the Base model for cloning (Qwen3-TTS setups). - A clone id that names a model of a different family is ignored - with a warning, since preset requests must synthesize with the - family the run is configured for. - """ - clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID - if not clone_model_id or clone_model_id == self.model_id: - return - families = {model["id"]: model["family"] for model in models} - if clone_model_id not in families: - # A qwen3_tts primary without its clone entry silently degrades - # (presets are ignored on the CustomVoice model), so that case - # keeps the warning; single-model servers of other families are - # the normal configuration and only get a debug note. - primary_is_qwen = (families.get(self.model_id) - or AUDIOCPP_FAMILY_QWEN3_TTS) \ - == AUDIOCPP_FAMILY_QWEN3_TTS - if primary_is_qwen: - logger.warning( - "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " - "server; preset requests use '%s' instead", - clone_model_id, self.model_id) - else: - logger.debug( - "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " - "server; preset requests use '%s' instead", - clone_model_id, self.model_id) - return - primary_family = families.get(self.model_id) - clone_family = families[clone_model_id] - if primary_family and clone_family and primary_family != clone_family: - logger.warning( - "AUDIOCPP_CLONE_MODEL_ID %r hosts family %r, but " - "AUDIOCPP_MODEL_ID %r hosts %r; preset requests stay on " - "'%s'. Point both ids at the same model entry in " - "converter/config.py (single-model servers use the same id " - "for both)", - clone_model_id, clone_family, self.model_id, primary_family, - self.model_id) - return - self.model_id = clone_model_id - - def _resolve_family(self, models: List[Dict[str, str]]) -> None: - """Resolve the selected model's family and its request profile. - - The family comes from GET /v1/models. Servers that predate the - family field served Qwen3-TTS only, so a missing family is treated - as qwen3_tts, which also preserves this client's legacy behavior - against those versions. - """ - entry = next( - (model for model in models if model["id"] == self.model_id), None) - family = (entry["family"] if entry is not None else "") or "" - if not family: - family = AUDIOCPP_FAMILY_QWEN3_TTS - logger.debug("Model '%s' reported no family; assuming qwen3_tts", - self.model_id) - self.family = family - self.profile = AUDIOCPP_FAMILY_PROFILES.get( - family, AUDIOCPP_DEFAULT_FAMILY_PROFILE) - if family not in AUDIOCPP_FAMILY_PROFILES: - logger.info( - "audio.cpp family '%s' has no dedicated profile; using the " - "generic profile (voice cloning via --voice, model-detected " - "language)", family) - - def _resolve_task(self, models: List[Dict[str, str]]) -> None: - """Resolve the selected model's task (tts, clon, vdes, ...) and set - design mode for voice design entries. - - The task comes from GET /v1/models and is fixed per server entry by - its server.json config (a VoiceDesign model must be hosted with - "task": "vdes"). Servers that predate the task field hosted plain - TTS models, so a missing task is treated as tts. - """ - entry = next( - (model for model in models if model["id"] == self.model_id), None) - task = (entry["task"] if entry is not None else "") or "" - if not task: - task = AUDIOCPP_TASK_TTS - logger.debug("Model '%s' reported no task; assuming tts", - self.model_id) - self.task = task - self.design_mode = task == AUDIOCPP_TASK_VDES - - def _check_voice(self) -> None: - """Verify the requested voice is available on the server. - - A voice name that matches no server preset or voice-library wav - would be passed through to the model as a cached voice id; on the - Base (cloning) model that is silently ignored and plain TTS audio - comes back, so preset names are validated up front. When the - voices endpoint cannot be queried, validation is skipped with a - warning rather than blocking the run. - """ - query = urllib.parse.urlencode({"model": self.model_id}) - try: - payload = self._get_json(f"/v1/audio/voices?{query}") - except Exception as exc: - logger.warning("Could not list server voices; skipping voice " - "validation: %s", exc) - return - voices = payload.get("voices") or [] - if self.voice not in voices: - available = ", ".join(str(v) for v in voices) or "none" - raise RuntimeError( - f"Voice '{self.voice}' is not available on the audio.cpp server " - f"(available: {available}). Configure it as a voice_preset or " - "voice_dir entry in the server config, or pass a listed name " - "with --voice (see README)." - ) - - # ------------------------------------------------------------------ - # HTTP requests - # ------------------------------------------------------------------ - - def _request_wav(self, text: str) -> bytes: - """POST one sub-chunk and return the raw WAV bytes. - - The request timeout scales with the text length when a whole - chapter is sent in one request (no client-side chunking), since a - long chapter means many minutes of audio generated in one go. - """ - url = f"{self.api_url}/v1/audio/speech" - payload: Dict[str, Any] = { - "model": self.model_id, - "input": text, - } - # Design models take no voice field (the voice comes from the - # instruction); instruction-voice runs on families without built-in - # speakers omit it too, since no speaker or preset was requested. - if not self.design_mode and not self.instruction_voice: - payload["voice"] = self.voice - if self.profile.language_style == AUDIOCPP_LANG_DISPLAY: - payload["language"] = self.language - elif self.profile.language_style == AUDIOCPP_LANG_ISO: - iso_code = LANGUAGE_ISO_CODES.get(self.language) - if iso_code: - payload["language"] = iso_code - else: - # "Auto": no code to send, so let the server pick its default. - logger.debug("%s: no language code for %r; omitted from request", - self.family, self.language) - if self._seed >= 0: - # audio.cpp has no negative "randomize" seed; a negative seed - # means "let the server randomize", so the field is omitted. - payload["seed"] = self._seed - if self.instructions: - # Explicit voice-design or style instruction (required for task - # "vdes" entries; a Ctrl/style control on families that read it). - payload["instructions"] = self.instructions - elif not self.preset_mode and config.INSTRUCT \ - and self.profile.sends_instructions: - # Style instruction for the Qwen3-TTS CustomVoice speakers; - # ignored by the Base (cloning) model and other families. - payload["instructions"] = config.INSTRUCT - if self.request_options: - # Generic per-model controls (--option KEY=VALUE): forwarded - # verbatim; the model ignores keys it does not know. - payload["options"] = dict(self.request_options) - request = urllib.request.Request( - url, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, method="POST") - timeout = config.API_TIMEOUT - if not self.chunk_text: - # Estimated audio duration at 150 wpm, doubled plus a minute of - # slack, bounded below by the configured per-request timeout. - estimated_seconds = 60.0 * len(text.split()) / _ESTIMATED_WORDS_PER_MINUTE - timeout = max(timeout, int(estimated_seconds * 2) + 60) - try: - with urllib.request.urlopen(request, timeout=timeout) as response: - wav = 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"audio.cpp server returned HTTP {exc.code}: {detail}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"audio.cpp request failed: {exc.reason}") from exc - if len(wav) < 12 or wav[:4] != b"RIFF" or wav[8:12] != b"WAVE": - raise RuntimeError("audio.cpp server returned audio that is not a WAV file") - return wav - - def _request_wav_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.MAX_RETRIES): - try: - return self._request_wav(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.MAX_RETRIES - 1: - time.sleep(2 + 2 * attempt) - raise RuntimeError(f"Sub-chunk {sub_num}/{sub_total} failed after " - f"{config.MAX_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. - - By default the whole text goes out as a single request and the - server does its own long-form chunking (see the class docstring). - With ``chunk_text=True`` (--chunk), the text is split into - sub-requests of at most ``config.CHUNK_SIZE`` words each; each - sub-request returns a complete WAV file and the parts are - concatenated into one chunk file. - """ - try: - if self.chunk_text: - sub_texts = split_into_chunks(text, max_words=config.CHUNK_SIZE) - elif text.strip(): - sub_texts = [text] - else: - sub_texts = [] - if not sub_texts: - raise RuntimeError("No text to synthesize") - - output_path: Optional[Path] = None - with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ - self._chunk_heartbeat( - chunk_num, - label=None if self.chunk_text else "Request"): - part_paths = [] - for sub_num, sub_text in enumerate(sub_texts, 1): - wav = self._request_wav_with_retry( - sub_text, chunk_num, sub_num, len(sub_texts)) - destination = Path(parts_dir) / f"part_{sub_num:02d}.wav" - destination.write_bytes(wav) - part_paths.append(destination) - if len(part_paths) == 1: - output_path = self._chunk_path(chunk_num, ".wav") - shutil.copy2(part_paths[0], output_path) - else: - output_path = self._chunk_path(chunk_num, ".wav") - concat_audio_files(part_paths, output_path) - - logger.debug("Chunk %d generated successfully (%d sub-request(s))", - chunk_num, len(sub_texts)) - return str(output_path) - - except Exception as exc: - logger.error("audio.cpp chunk processing failed for chunk %d: %s", - chunk_num, exc) - return None |