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,
    BACKEND_SGLOMNI,
    ConversionCancelled,
    MODEL_SIZE,
    VOICE_MODE_CLONE,
    VOICE_MODE_CUSTOM,
    VOICE_MODE_DESIGN,
    VOICE_MODES,
    AudioCppTTSClient,
    FasterTTSClient,
    QwenTTSClient,
    SgOmniTTSClient,
    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"]

# Device names Windows cannot use as a file name (with or without an
# extension); sanitized output names matching these get a prefix.
_WINDOWS_RESERVED_NAMES = frozenset(
    {"CON", "PRN", "AUX", "NUL"}
    | {f"COM{i}" for i in range(1, 10)}
    | {f"LPT{i}" for i in range(1, 10)})


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,
                   model: 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;
    sglomni resolves from the selected model's capability — a design model
    takes instructions, a clone-capable model clones when a reference .wav
    is given and otherwise synthesizes its default voice, and a
    speaker-capable model takes a preset name; 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 backend == BACKEND_SGLOMNI:
        from backends.sglomni.catalog import entry_by_key
        entry = entry_by_key(model or "")
        if entry is not None:
            if entry.capability == "design":
                return VOICE_MODE_DESIGN
            if entry.capability == "clone":
                return VOICE_MODE_CLONE if clone else VOICE_MODE_CUSTOM
            return VOICE_MODE_CUSTOM
        # Unresolved model (the caller resolves it later): the qwen-style
        # heuristic is the closest pre-flight approximation.
    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 ChunkClampCancelled(Exception):
    """The chunk-cap popup's Cancel answer: the run stops unstarted."""


def chunk_clamp_needed(entry) -> bool:
    """True when ENTRY cannot narrate a full CHUNK_SIZE sub-request.

    A catalog entry declares ``chunk_words`` when its engine's admission
    window caps one request below what the configured CHUNK_SIZE can
    narrate (Higgs: the server pins each request's prompt plus
    generation at 4096 tokens).
    """
    return (entry is not None and entry.chunk_words is not None
            and entry.chunk_words < config.CHUNK_SIZE)


def chunk_clamp_message(entry) -> List[str]:
    """The popup text for a chunk_words-capped ENTRY (short lines)."""
    return [
        f"{entry.label} can narrate at most ~{entry.chunk_words} words "
        "per request: the server caps",
        "each request's prompt plus generation at a fixed window. "
        "Longer sub-chunks",
        "may cut off mid-sentence.",
    ]


def prompt_chunk_clamp(entry, ask=None) -> Optional[int]:
    """The per-run sub-request word cap for ENTRY (None = no clamp).

    Models whose catalog entry declares ``chunk_words`` below CHUNK_SIZE
    (Higgs) cannot narrate a full sub-chunk in one request; the popup
    offers to clamp CHUNK_SIZE for this run, keep it ("try anyway",
    risking mid-chunk truncation), or cancel the run. ASK, when given,
    replaces the console prompt: it is called with (message lines, words)
    and returns "clamp" or "anyway" (Cancel raises inside the callback —
    the hub maps it to returning to the Generate form). A closed stdin
    (non-interactive run) clamps: unattended runs keep working and never
    produce silently truncated audio.
    """
    if not chunk_clamp_needed(entry):
        return None
    lines = chunk_clamp_message(entry)
    if ask is not None:
        answer = ask(lines, entry.chunk_words)
        return entry.chunk_words if answer == "clamp" else None
    for line in lines:
        print(line)
    while True:
        try:
            answer = input(
                f"Set Chunk to {entry.chunk_words} for this run, try "
                "anyway, or cancel? [S/t/c]: ").strip().lower()
        except EOFError:
            print(f"\n[WARNING] No interactive input available; clamping "
                  f"sub-requests to {entry.chunk_words} words for this run")
            return entry.chunk_words
        if answer in ("", "s", "set"):
            return entry.chunk_words
        if answer in ("t", "try"):
            return None
        if answer in ("c", "cancel"):
            raise ChunkClampCancelled(
                "Conversion cancelled at the chunk-size prompt")
        print("Please answer 's', 't', or 'c'.")


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,
                 chunk_size: Optional[int] = 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)
        # The run's model selection (audio.cpp: a server entry id, sglomni:
        # the resolved catalog key, None elsewhere) — the startup banner
        # reports it.
        self.model_id = model_id
        # 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, cancel=cancel)
        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,
                                         cancel=cancel)
        elif backend == BACKEND_SGLOMNI:
            # SGLang-Omni hosts one model per server process; MODEL_ID
            # names the catalog entry. Managed runs (no API_URL) require
            # the model's weights on disk (resolved here, with the
            # actionable message when they are not); remote runs accept
            # any catalog key — the external server has its own weights.
            # The client resolves the request shape from the entry's
            # voice capability (preset speaker / per-request clone /
            # described-voice design) at connect time.
            from backends.sglomni import models as sg_models
            if api_url is None:
                entry = sg_models.resolve_model(model_id)
            else:
                from backends.sglomni.catalog import entry_by_key
                entry = entry_by_key((model_id or "").strip())
                if entry is None:
                    raise RuntimeError(
                        f"Unknown SGLang-Omni model {model_id!r} — pick a "
                        "catalog key for --model (see the backend docs).")
            model_id = entry.key
            self.model_id = model_id
            # CHUNK_SIZE carries a per-run override (the pre-flight chunk
            # popup's clamp for models whose engine caps one request below
            # a full sub-chunk); None follows the config.CHUNK_SIZE setting.
            self.chunk_size = chunk_size
            self.tts = SgOmniTTSClient(
                chunks_dir=CHUNKS_FOLDER, model=model_id, voice=voice,
                ref_audio=voice_clone_ref_audio,
                ref_text=voice_clone_ref_text,
                skip_transcription=skip_transcription,
                instructions=instructions, language=self.language,
                api_url=api_url, chunk_size=self.chunk_size,
                quiet=quiet, cancel=cancel)
        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,
                cancel=cancel,
            )
        self._progress = progress
        # The converter's own handle on the run's cancel event (also passed
        # to the client, so a cancel during connect-time work is honored).
        self._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(self, "_cancel", None)
        if not isinstance(cancel, threading.Event):
            # Converters built without __init__ (tests): fall back to the
            # client's event, the pre-constructor-arg wiring.
            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.

        Reserved Windows device names (CON, NUL, COM1, ...) are suffixed
        so the resulting name is writable on every platform.
        """
        cleaned = re.sub(r'[\\/:*?"<>|]', " ", name)
        cleaned = re.sub(r"\s+", " ", cleaned).strip().strip(".")
        cleaned = cleaned[:80] or fallback
        if cleaned.upper() in _WINDOWS_RESERVED_NAMES:
            return f"{fallback}_{cleaned}"
        return cleaned

    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 backend == BACKEND_SGLOMNI:
            if voice_mode == VOICE_MODE_CLONE and voice_clone_ref_audio:
                narrator = Path(voice_clone_ref_audio).stem
            elif voice_mode == VOICE_MODE_DESIGN:
                narrator = "designed"
            else:
                # A preset name on speaker-capable models, or the server's
                # built-in default voice (clone models without a reference).
                narrator = voice or "default"
        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 — each with its speed-adjusted copy when
            # SPEED != 1.0, see audio.combine_chunks). Reported on the
            # book_done/book_failed events so the run view can list them
            # in its summary.
            speed_tag = ("" if abs(self.speed - 1.0) < audio.SPEED_EPSILON
                         else f"_{self.speed:g}")
            if self.output_format == "m4b" or self.single_file \
                    or len(sections) == 1:
                self.current_outputs = [f"{stem}{speed_tag}."
                                        f"{self.output_format}"]
            else:
                self.current_outputs = [
                    f"{stem}_{index:02d}_"
                    f"{self._sanitize_filename(section.title)}{speed_tag}."
                    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_SGLOMNI: "SGLang-Omni 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}")
        elif self.backend == BACKEND_SGLOMNI:
            entry = getattr(self.tts, "entry", None)
            api_url = getattr(self.tts, "api_url", None) \
                or config.SGLOMNI_API_URL
            self._say(f"SGLang-Omni endpoint: {api_url}")
            self._say(f"Model: {getattr(entry, 'label', self.model_id or '?')}"
                      f" ({getattr(entry, 'repo', '')})")
            if getattr(entry, "capability", None) == "design":
                self._say("Backend: SGLang-Omni (voice from --instructions "
                          "description)")
                self._say(f"Instruction: {self.instructions}")
            elif getattr(entry, "capability", None) == "clone":
                if self.voice_clone_ref_audio:
                    self._say("Backend: SGLang-Omni (voice cloning from a "
                              "reference clip)")
                    self._say(f"Reference audio: "
                              f"{Path(self.voice_clone_ref_audio).name}")
                else:
                    self._say("Backend: SGLang-Omni (model's default voice)")
            else:
                self._say("Backend: SGLang-Omni (built-in preset voice)")
                self._say(f"Voice: {self.voice or 'default'}")
            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)")

        # Per-book outcome ({book file name: ok}), published on the
        # instance so multi-book orchestrators (the "All" run) can count
        # partial success after run() returns False (a failed book aborts
        # the rest, but earlier books still count).
        self.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)
                self.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")
                self.results[book_file.name] = False
                break
            except Exception as exc:
                logger.error("Unexpected error: %s", exc)
                self.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 self.results.get(book_file.name):
                logger.error("Conversion of %s failed; aborting the remaining books",
                             book_file.name)
                break

        successful = sum(self.results.values())
        total = len(self.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 self.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