"""Orchestrates book-to-audiobook conversion.""" import glob import logging import re import shutil import sys import threading import time from collections import Counter from datetime import datetime from pathlib import Path from typing import Callable, Dict, List, Optional, Tuple import logging_kit from . import audio, chunking, config, cover, extractors from .audio import TrackMeta from .clients import ( BACKENDS, BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, ConversionCancelled, MODEL_SIZE, VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN, VOICE_MODES, AudioCppTTSClient, FasterTTSClient, QwenTTSClient, normalize_language, speaker_display_name_for, ) logger = logging.getLogger(__name__) # Folders, resolved from the project root so the converter runs from any # working directory. User-facing dirs (input/, output/) come from config # (INPUT_DIR/OUTPUT_DIR; relative paths resolve against the project root); # scratch/log dirs live under the app/ container. BASE_DIR = Path(__file__).resolve().parent.parent.parent APP_DIR = BASE_DIR / "app" def resolve_dir(value, default: str) -> Path: """Resolve a configured directory VALUE (str/Path) to a Path. Blank values fall back to DEFAULT ("input"/"output"); relative paths resolve against the project root so the converter runs from any working directory, and "~" expands to the home directory. """ path = Path(str(value or "").strip() or default).expanduser() return path if path.is_absolute() else BASE_DIR / path BOOKS_FOLDER = resolve_dir(config.INPUT_DIR, "input") AUDIOBOOKS_FOLDER = resolve_dir(config.OUTPUT_DIR, "output") CHUNKS_FOLDER = APP_DIR / "chunks" # Per-chunk scratch audio, cleaned per book # The converter's log stream (audiobook_YYYYMMDD.log); naming/retention # policy lives in logging_kit. LOGS_FOLDER = logging_kit.LOG_DIR DEBUG_FOLDER = APP_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 and file-only traceback dumps out of the console (file only).""" return not record.name.startswith( ("httpx", "httpcore", logging_kit.TRACEBACK_LOGGER)) def setup_logging(debug: bool = False, console: bool = True) -> None: """Configure logging to a dated file and (optionally) 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. CONSOLE=False (the TUI run view owns the screen) keeps every record in the file only. Safe to call repeatedly in one process (the hub calls it once per conversion run): force=True replaces the previous handlers instead of silently keeping them. """ 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) handlers = [file_handler] if console: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.DEBUG if debug else logging.WARNING) console_handler.addFilter(_console_log_filter) handlers.append(console_handler) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=handlers, force=True, ) if debug: logging.getLogger("converter").setLevel(logging.DEBUG) else: logging.getLogger("converter").setLevel(logging.INFO) 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 voice_mode_for(backend: str, voice: Optional[str] = None, clone: Optional[str] = None, instructions: Optional[str] = None) -> str: """The voice mode a run with these options would use. Mirrors the choice ``audiobook.convert`` makes from the same inputs (faster always clones; audiocpp clones through a server-side voice; qwen designs with instructions, clones only with a reference .wav, and uses a built-in speaker otherwise), so the hub can run the pre-flight overwrite checks against exactly the output names the conversion will produce. """ if backend == BACKEND_FASTER: return VOICE_MODE_CLONE if backend == BACKEND_AUDIOCPP: return VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM if (instructions or "").strip(): return VOICE_MODE_DESIGN return VOICE_MODE_CLONE if clone else VOICE_MODE_CUSTOM 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 _overwrite_message(existing: List[Path], output_name: str) -> str: """The overwrite question for the files in EXISTING.""" if len(existing) == 1: return (f"{existing[0].name} already exists. Convert anyway " "and overwrite it?") return (f"{len(existing)} output files for '{output_name}' already exist " f"(e.g. {existing[0].name}). Convert anyway and overwrite them?") def prompt_overwrite(existing: List[Path], output_name: str, confirm: Optional[Callable[[str, bool], bool]] = None) -> 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. CONFIRM, when given, replaces the console ``input()`` prompt: it is called once with (message, default) and must return the answer — the hub passes a TUI yes/no dialog so the questions are asked inside the menu instead of the console. """ message = _overwrite_message(existing, output_name) if confirm is not None: return confirm(message, True) 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.""" # Class-level default so a partially-constructed instance (tests build # these with __new__) behaves like a plain console run. _progress = None def __init__(self, voice_mode: str = VOICE_MODE_CUSTOM, voice_clone_ref_audio: Optional[str] = None, voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, speed: float = 1.0, single_file: bool = False, output_format: str = config.AUDIO_FORMAT, language: Optional[str] = None, backend: str = None, voice: Optional[str] = None, debug: bool = False, model_id: Optional[str] = None, instructions: Optional[str] = None, request_options: Optional[Dict[str, str]] = None, api_url: Optional[str] = None, progress: Optional[Callable[[dict], None]] = None, cancel=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 is None: raise ValueError("backend is required (pass --backend)") 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) # Output file names the book being converted will produce (filled in # by convert_book; reported on the book_done/book_failed events). self.current_outputs: List[str] = [] # 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() # Interactive reporting (the TUI run view): PROGRESS receives an # event dict per state change and turns the clients' console prints # off (quiet is set at construction so connect-time lines respect it # too); CANCEL (a threading.Event) stops the run between requests. quiet = progress is not None 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(chunks_dir=CHUNKS_FOLDER, voice=voice, api_url=api_url, quiet=quiet) elif backend == BACKEND_AUDIOCPP: # --voice picks the voice: a built-in speaker name on the # CustomVoice entry, or a server-side preset (cloning) # elsewhere. model_id picks the server entry per run # (auto-selected on single-entry servers); instructions # describe or style the voice, request_options pass # per-model controls through to the server. self.tts = AudioCppTTSClient(chunks_dir=CHUNKS_FOLDER, voice=voice, language=self.language, model_id=model_id, instructions=instructions, request_options=self.request_options, api_url=api_url, quiet=quiet) else: # Qwen: the voice mode picks the request shape (built-in # speaker, clone from a reference .wav, or a designed voice); # --voice names the built-in speaker in speaker mode and # instructions describe the voice in design mode. self.tts = QwenTTSClient( chunks_dir=CHUNKS_FOLDER, 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, instructions=self.instructions, api_url=api_url, quiet=quiet, voice=voice, ) self._progress = progress self.tts.cancel = cancel def _emit(self, event: dict) -> None: """Send one progress event (a no-op without a progress callback).""" if self._progress is not None: self._progress(event) def _say(self, message: str) -> None: """Print a console progress line unless the run view owns the screen.""" if self._progress is None: print(message) def _check_cancelled(self) -> None: """Raise ConversionCancelled when the run's cancel event is set.""" cancel = getattr(getattr(self, "tts", None), "cancel", None) if isinstance(cancel, threading.Event) and cancel.is_set(): raise ConversionCancelled("Cancelled by user") 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.backend == BACKEND_QWEN and self.voice_mode == VOICE_MODE_DESIGN \ and not (self.instructions or "").strip(): raise ValueError( "Voice Design mode requires a voice description. " "Use --instructions \"...\" to describe the voice to synthesize with." ) if self.backend == BACKEND_QWEN and self.voice_mode == VOICE_MODE_CUSTOM \ and not (self.voice or "").strip(): raise ValueError( "CustomVoice mode requires a speaker. Use --voice SPEAKER " "(e.g. Vivian) to pick one, or --clone / --instructions " "for the other 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 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 (for audiocpp's speaker mode, the selected built-in CustomVoice speaker). 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 "default" elif backend == BACKEND_AUDIOCPP: if voice: narrator = voice elif instructions: # The voice comes from the instruction, not a speaker name. narrator = "designed" else: # Unreachable in a valid run (the audiocpp client refuses a # speaker-capable entry without --voice); keep a stable tag # for the pre-flight of runs that will fail at connect time. narrator = "narrator" elif voice_mode == VOICE_MODE_DESIGN: # Qwen's VoiceDesign model: the voice is described by an # instruction and has no speaker name. narrator = "designed" elif voice_mode == VOICE_MODE_CLONE: narrator = Path(voice_clone_ref_audio).stem else: narrator = speaker_display_name_for(voice or "") 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. Raises ConversionCancelled when the run's cancel event is set. """ logger.info("Converting: %s", file_path.name) start_time = time.time() try: # Reset per book (an instance may have been built without # __init__, e.g. in tests): no outputs until the plan is known. self.current_outputs = [] # Start from a clean scratch folder so a previous crash can never # affect this run audio.cleanup_chunks(CHUNKS_FOLDER) 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()}" # The output files this book will produce (single final file, # or one per chapter). Reported on the book_done/book_failed # events so the run view can list them in its summary. if self.output_format == "m4b" or self.single_file \ or len(sections) == 1: self.current_outputs = [f"{stem}.{self.output_format}"] else: self.current_outputs = [ f"{stem}_{index:02d}_" f"{self._sanitize_filename(section.title)}." f"{self.output_format}" for index, section in enumerate(sections, 1)] # --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: self._say(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): self._check_cancelled() self._emit({"kind": "chapter", "index": index, "total": len(sections)}) 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 ConversionCancelled: raise except Exception as exc: logger.error("Conversion failed: %s", exc) logging_kit.log_traceback() return False finally: # Always cleanup, even on failure or interrupt audio.cleanup_chunks(CHUNKS_FOLDER) 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): self._check_cancelled() self._emit({"kind": "chapter", "index": index, "total": total_chapters}) chapter_path = CHUNKS_FOLDER / f"chapter_{index:04d}.wav" title = (section.title or "").strip() or f"Chapter {index}" if self._progress is None: 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, chunks_dir=CHUNKS_FOLDER, 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. Raises ConversionCancelled when the run's cancel event is set (between chunks). """ total_chunks = len(chunks) if self._progress is None: 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): self._check_cancelled() 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) self._say(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks) self._emit({"kind": "chunk_done", "chunk": chunk_num, "total": total_chunks, "seconds": time.time() - request_start}) else: logger.error("Chunk %d/%d failed; aborting the remaining chunks", chunk_num, total_chunks) self._emit({"kind": "chunk_failed", "chunk": chunk_num, "total": total_chunks}) break except ConversionCancelled: raise except Exception as exc: results[chunk_num] = None logger.error("Chunk %d/%d error: %s; aborting the remaining chunks", chunk_num, total_chunks, exc) self._emit({"kind": "chunk_failed", "chunk": chunk_num, "total": total_chunks, "error": str(exc)}) break successful_chunks = sum(1 for path in results.values() if path) if self._progress is None: 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 CHUNK_SIZE-word TTS requests.""" return chunking.split_into_chunks(text) 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) 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") self._say(f"[INFO] Processing {total_chunks} chunks via {backend}...") self._emit({"kind": "chunks", "total": total_chunks}) 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, chunks_dir=CHUNKS_FOLDER, 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) self._say(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted " f"({successful_chunks}/{total_chunks} chunks)") 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 ConversionCancelled: raise except Exception as exc: logger.error("Conversion failed: %s", exc) logging_kit.log_traceback() return False def _print_banner(self) -> None: """Print the startup summary for the selected backend.""" self._say("=" * 70) self._say("TTS AUDIOBOOK GENERATOR") self._say("=" * 70) self._say(f"Books folder: {BOOKS_FOLDER}") self._say(f"Output folder: {AUDIOBOOKS_FOLDER}") if self.backend == BACKEND_FASTER: self._say(f"Faster TTS endpoint: {config.FASTER_API_URL}") self._say("Backend: faster (voice cloning, reference configured on server)") self._say(f"Voice: {self.voice}") elif self.backend == BACKEND_AUDIOCPP: self._say(f"audio.cpp endpoint: {config.AUDIOCPP_API_URL}") self._say(f"Model id: {self.tts.model_id}") self._say(f"Model family: {getattr(self.tts, 'family', 'unknown')}") if getattr(self.tts, "preset_mode", False): self._say("Backend: audio.cpp (voice cloning, reference configured on server)") self._say(f"Voice: {self.tts.voice}") elif getattr(self.tts, "speaker_mode", False): self._say("Backend: audio.cpp (custom voice, built-in speaker)") self._say(f"Speaker: {self.tts.voice}") elif self.instructions: self._say("Backend: audio.cpp (voice from --instructions description)") self._say(f"Instruction: {self.instructions}") if self.request_options: self._say(f"Request options: {self.request_options}") self._say(f"Language: {self.language}") else: tts_client = getattr(self, "tts", None) api_url = (getattr(tts_client, "api_url", None) or config.QWEN_API_URL) self._say(f"Qwen API endpoint: {api_url}") self._say(f"Voice mode: {self.voice_mode}") self._say(f"Model size: {MODEL_SIZE} (always)") if self.voice_mode == VOICE_MODE_CUSTOM: self._say(f"Speaker: {self.voice}") self._say(f"Language: {self.language}") elif self.voice_mode == VOICE_MODE_CLONE: self._say(f"Reference audio: {Path(self.voice_clone_ref_audio).name}") self._say(f"Language: {self.language}") elif self.voice_mode == VOICE_MODE_DESIGN: self._say("Backend: qwen-tts (voice from --instructions description)") self._say(f"Instruction: {self.instructions}") self._say(f"Language: {self.language}") self._say(f"Output format: {self.output_format}") if self.single_file and self.output_format != "m4b": self._say("Chapter mode: single file (--single-file)") if abs(self.speed - 1.0) >= audio.SPEED_EPSILON: self._say(f"Playback speed: {self.speed:g}x") if self.debug: self._say(f"Debug dumps (per-chunk text + raw audio): {DEBUG_FOLDER}") self._say("=" * 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, confirm: Optional[Callable[[str, bool], bool]] = None, book_files: Optional[List[Path]] = None, output_name: 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. CONFIRM replaces the console ``input()`` prompt (the hub passes a TUI yes/no dialog). BOOK_FILES overrides the books-folder scan with an explicit list (a single --input-file book; still filtered to supported formats), and OUTPUT_NAME overrides the computed output name with a verbatim base name (--output-file's stem, no narrator tag or stem-collision suffix). Both default to the directory-scan behavior. """ if book_files is None: book_files = sorted( f for f in BOOKS_FOLDER.iterdir() if f.is_file() and f.suffix.lower() in SUPPORTED_FORMATS ) else: book_files = sorted( f for f in book_files 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") # Compute the output name each book would produce. An explicit # name (--output-file) is used verbatim for the single book; # otherwise names carry the narrator tag and a stem-collision # suffix when two books share a stem (e.g. dune.txt + dune.epub). if output_name is not None: names = [(book_files[0], output_name)] else: stem_counts: Dict[str, int] = Counter(book_file.stem for book_file in book_files) narrator_tag = AudiobookConverter.compute_narrator_tag( backend, voice, voice_mode, voice_clone_ref_audio, instructions) names = [] for book_file in book_files: name = book_file.stem if stem_counts[book_file.stem] > 1: name = f"{book_file.stem}_{book_file.suffix.lstrip('.')}" names.append((book_file, f"{name}_{narrator_tag}")) # Ask every overwrite question up front, before any conversion # starts, so the rest of the run is unattended. planned: List[Tuple[Path, str]] = [] for book_file, name in names: existing = find_existing_outputs(name, output_format) if existing and not prompt_overwrite(existing, name, confirm=confirm): print(f"[INFO] Skipping {book_file.name} (existing output kept)") continue planned.append((book_file, name)) return book_files, planned # ------------------------------------------------------------------ # Main conversion loop # ------------------------------------------------------------------ def run(self) -> bool: """Main conversion process. Returns True if all books converted. Raises ConversionCancelled when the run's cancel event is set. """ 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: self._say(f"[INFO] No supported files found in {BOOKS_FOLDER}") self._say(f"Supported formats: {', '.join(SUPPORTED_FORMATS)}") self._say("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " f"to {BOOKS_FOLDER} and run again.") self._emit({"kind": "done", "ok": 0, "total": 0}) return True if not planned: self._say("[INFO] Nothing to convert (all books skipped)") self._emit({"kind": "done", "ok": 0, "total": 0}) return True self._say(f"[INFO] Converting {len(planned)} of {len(book_files)} book(s)") results = {} cancelled = False for index, (book_file, output_name) in enumerate(planned, 1): self._check_cancelled() self._emit({"kind": "book", "index": index, "total": len(planned), "name": book_file.name}) try: success = self.convert_book(book_file, output_name=output_name) results[book_file.name] = success self._emit({"kind": "book_done", "name": book_file.name, "ok": bool(success), "files": list(getattr(self, "current_outputs", []))}) except ConversionCancelled: self._emit({"kind": "cancelled"}) logger.info("Conversion cancelled by user at %s", book_file.name) cancelled = True break except KeyboardInterrupt: self._say("\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 self._emit({"kind": "book_failed", "name": book_file.name, "error": str(exc), "files": list(getattr(self, "current_outputs", []))}) if not results.get(book_file.name): logger.error("Conversion of %s failed; aborting the remaining books", book_file.name) break successful = sum(results.values()) total = len(results) # A cancelled run is not a successful run on either path (the TUI # event consumer and the console summary report it consistently). ok = not cancelled and total > 0 and successful == total self._emit({"kind": "done", "ok": successful, "total": total or len(planned), "cancelled": cancelled}) if self._progress is not None: return ok 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 ok