path: root/app/converter/converter.py

"""Orchestrates book-to-audiobook conversion."""

import glob
import logging
import re
import shutil
import sys
import threading
import time
from collections import Counter
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(
        logging_kit.stream_path("audiobook", LOGS_FOLDER),
        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,
                 unload_models: Optional[bool] = 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 {})
        # audio.cpp only: force unloading previously-loaded server models
        # at connect time (None follows the AUDIOCPP_UNLOAD_MODELS setting).
        self.unload_models = unload_models
        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. unload_models
            # forces a pre-run model unload when not None ("All" runs).
            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,
                                         unload_models=unload_models)
        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 <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 (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(" ", "_")

    @staticmethod
    def compute_model_tag(model_id: Optional[str]) -> str:
        """Model id used in output file names, without a server connection.

        "All (multiple generation)" runs name every output with the
        generating model's id so the per-model files never collide
        (e.g. ``dune_qwen3_tts_1_7b_base_q8_0_Vivian.m4b``). Pure (no I/O,
        no server) so the pre-flight can compute the exact output names a
        run would produce before spending time connecting to a TTS server.
        """
        return AudiobookConverter._sanitize_filename(
            model_id or "", fallback="model").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,
                             name_tag: 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. NAME_TAG, when
        given, is inserted between the book stem and the narrator tag
        ("All (multiple generation)" runs pass the sanitized model id, so
        each model's outputs are named and planned separately).
        """
        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('.')}"
                tag = f"{name_tag}_{narrator_tag}" if name_tag else narrator_tag
                names.append((book_file, f"{name}_{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