refactor(webui): replace inline get_pipeline/resolve_voice_target closures with domain modules

- Replace get_pipeline() closure with PipelinePool from domain/pipeline_factory
- Replace resolve_voice_target() closure with domain function from voice_utils
- Remove dead _load_pipeline() function and unused is_plugin_registered import
- Add 33 tests for resolve_voice_target and PipelinePool
- Add 10 regression tests verifying domain extraction preserves behavior
- 1131 tests pass (+61 new)

.gitattributes

@@ -0,0 +1,15 @@
+*.py text eol=lf
+*.md text eol=lf
+*.yml text eol=lf
+*.yaml text eol=lf
+*.toml text eol=lf
+*.json text eol=lf
+*.txt text eol=lf
+*.html text eol=lf
+*.css text eol=lf
+*.js text eol=lf
+*.sh text eol=lf
+*.cfg text eol=lf
+*.ini text eol=lf
+*.svg text eol=lf
+*.j2 text eol=lf

.github/workflows/test_pip.yml

@@ -1,7 +1,9 @@
-name: pip install
+name: CI
-run-name: pip install
+run-name: CI
-on: 
+
+on:
   push:
+    branches: [main]
     paths:
       - '**.py'
       - 'pyproject.toml'
@@ -11,23 +13,41 @@ on:
       - 'pyproject.toml'
       - '.github/workflows/**'
   workflow_dispatch:
+
 jobs:
-  install-and-run:
+  test:
     strategy:
       matrix:
-        os: [ubuntu-latest, macos-latest, windows-latest]
+        os: [ubuntu-latest, macos-14, windows-latest]
         python-version: ['3.12']
       fail-fast: false
-    continue-on-error: true
     runs-on: ${{ matrix.os }}
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v7
+
       - name: Set up Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
-      - name: Install from repository
+
-        run: python -m pip install .
+      - name: Install uv
-      #- name: Run abogen
+        uses: astral-sh/setup-uv@v8.3.1
-      #  run: abogen
+        with:
+          enable-cache: true
+          prune-cache: false
+          cache-dependency-glob: pyproject.toml
+
+      - name: Install system dependencies (Ubuntu)
+        if: runner.os == 'Linux'
+        run: sudo apt-get update && sudo apt-get install -y libegl1
+
+      - name: Install dependencies
+        run: uv pip install --system .[dev]
+        env:
+          UV_LINK_MODE: copy
+
+      - name: Run tests
+        env:
+          QT_QPA_PLATFORM: offscreen
+        run: pytest tests/ -v --tb=short

.github/workflows/test_publish.yml

@@ -18,7 +18,7 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v7
 
     - name: Login to Github Container Registry
       # Only if we need to push an image

README.md

@@ -38,8 +38,6 @@ This method handles everything automatically - installing all dependencies inclu
 #### <b>OPTION 2: Install using uv</b>
 First, [install uv](https://docs.astral.sh/uv/getting-started/installation/) if you haven't already.
 
-The CUDA extras install both GPU-accelerated Kokoro (via PyTorch) and Supertonic (via onnxruntime-gpu).
-
 ```bash
 # For NVIDIA GPUs (CUDA 12.8) - Recommended
 uv tool install --python 3.12 abogen[cuda] --extra-index-url https://download.pytorch.org/whl/cu128 --index-strategy unsafe-best-match
@@ -67,9 +65,6 @@ venv\Scripts\activate
 # We need to use an older version of PyTorch (2.8.0) until this issue is fixed: https://github.com/pytorch/pytorch/issues/166628
 pip install torch==2.8.0+cu128 torchvision==0.23.0+cu128 torchaudio==2.8.0 --index-url https://download.pytorch.org/whl/cu128
 
-# Also install onnxruntime-gpu for Supertonic GPU acceleration:
-pip install onnxruntime-gpu
-
 # For AMD GPUs:
 # Not supported yet, because ROCm is not available on Windows. Use Linux if you have AMD GPU.
 
@@ -178,7 +173,7 @@ Abogen offers **two interfaces**, but currently they have different feature sets
 
 | Command | Interface | Features |
 |---------|-----------|----------|
-| `abogen` | PyQt6 Desktop GUI | Stable core features + **Supertonic TTS**|
+| `abogen` | PyQt6 Desktop GUI | Stable core features |
 | `abogen-web` | Flask Web UI | Core features + **Supertonic TTS**, **LLM Normalization**, **Audiobookshelf Integration** and more! |
 
 > **Note:** The Web UI is under active development. We are working to integrate these new features into the PyQt desktop app. until then, the Web UI provides the most feature-rich experience.
@@ -412,18 +407,18 @@ When Audiobookshelf sits behind Nginx Proxy Manager (NPM), make sure the API pat
 1. Create a **Proxy Host** that points to your ABS container or host (default forward port `13378`).
 2. Under the **SSL** tab, enable your certificate and tick **Force SSL** if you want HTTPS only.
 3. In the **Advanced** tab, append the snippet below so bearer tokens, client IPs, and large uploads survive the proxy hop:
-```nginx
+   ```nginx
-proxy_set_header Host $host;
+  proxy_set_header Host $host;
-proxy_set_header X-Real-IP $remote_addr;
+  proxy_set_header X-Real-IP $remote_addr;
-proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+  proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-proxy_set_header X-Forwarded-Proto $scheme;
+  proxy_set_header X-Forwarded-Proto $scheme;
-proxy_set_header X-Forwarded-Host $host;
+  proxy_set_header X-Forwarded-Host $host;
-proxy_set_header X-Forwarded-Port $server_port;
+  proxy_set_header X-Forwarded-Port $server_port;
-proxy_set_header Authorization $http_authorization;
+  proxy_set_header Authorization $http_authorization;
-client_max_body_size 5g;
+  client_max_body_size 5g;
-proxy_read_timeout 300s;
+  proxy_read_timeout 300s;
-proxy_connect_timeout 300s;
+  proxy_connect_timeout 300s;
-  ```
+   ```
 4. Disable **Block Common Exploits** (it strips Authorization headers in some NPM builds).
 5. Enable **Websockets Support** on the main proxy screen (Audiobookshelf uses it for the web UI, and it keeps the reverse proxy configuration consistent).
 6. If you publish Audiobookshelf under a path prefix (for example `/abs`), add a **Custom Location** with `Location: /abs/` and set the **Forward Path** to `/`. That rewrite strips the `/abs` prefix before traffic reaches Audiobookshelf so `/abs/api/...` on the internet becomes `/api/...` on the backend. Use the same prefixed URL in Abogen’s “Base URL” field.

WINDOWS_INSTALL.bat

@@ -323,13 +323,6 @@ if /I "%IS_NVIDIA%"=="true" (
             pause
             exit /b
         )
-        echo Installing onnxruntime-gpu for Supertonic GPU acceleration...
-        %PYTHON_CONSOLE_PATH% -m uv pip install --system onnxruntime-gpu
-        if errorlevel 1 (
-            echo Failed to install onnxruntime-gpu.
-            pause
-            exit /b
-        )
     ) else (
         echo CUDA is available on NVIDIA GPU.
     )
@@ -355,13 +348,6 @@ if /I "%IS_NVIDIA%"=="true" (
             pause
             exit /b
         )
-        echo Installing onnxruntime-gpu for Supertonic GPU acceleration...
-        %PYTHON_CONSOLE_PATH% -m uv pip install --system onnxruntime-gpu
-        if errorlevel 1 (
-            echo Failed to install onnxruntime-gpu.
-            pause
-            exit /b
-        )
     )
 )
 

abogen/assets/settings.svg

@@ -1,31 +1,31 @@
-<?xml version="1.0" encoding="utf-8"?>
+<?xml version="1.0" encoding="utf-8"?>
-
+
-<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
+<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
-<!-- Uploaded to: SVG Repo, www.svgrepo.com, Generator: SVG Repo Mixer Tools -->
+<!-- Uploaded to: SVG Repo, www.svgrepo.com, Generator: SVG Repo Mixer Tools -->
-<svg height="800px" width="800px" version="1.1" id="_x32_" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" 
+<svg height="800px" width="800px" version="1.1" id="_x32_" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" 
-	 viewBox="0 0 512 512"  xml:space="preserve">
+	 viewBox="0 0 512 512"  xml:space="preserve">
-<style type="text/css">
+<style type="text/css">
-	.st0{fill:#808080;}
+	.st0{fill:#808080;}
-</style>
+</style>
-<g>
+<g>
-	<path class="st0" d="M502.325,307.303l-39.006-30.805c-6.215-4.908-9.665-12.429-9.668-20.348c0-0.084,0-0.168,0-0.252
+	<path class="st0" d="M502.325,307.303l-39.006-30.805c-6.215-4.908-9.665-12.429-9.668-20.348c0-0.084,0-0.168,0-0.252
-		c-0.014-7.936,3.44-15.478,9.667-20.396l39.007-30.806c8.933-7.055,12.093-19.185,7.737-29.701l-17.134-41.366
+		c-0.014-7.936,3.44-15.478,9.667-20.396l39.007-30.806c8.933-7.055,12.093-19.185,7.737-29.701l-17.134-41.366
-		c-4.356-10.516-15.167-16.86-26.472-15.532l-49.366,5.8c-7.881,0.926-15.656-1.966-21.258-7.586
+		c-4.356-10.516-15.167-16.86-26.472-15.532l-49.366,5.8c-7.881,0.926-15.656-1.966-21.258-7.586
-		c-0.059-0.06-0.118-0.119-0.177-0.178c-5.597-5.602-8.476-13.36-7.552-21.225l5.799-49.363
+		c-0.059-0.06-0.118-0.119-0.177-0.178c-5.597-5.602-8.476-13.36-7.552-21.225l5.799-49.363
-		c1.328-11.305-5.015-22.116-15.531-26.472L337.004,1.939c-10.516-4.356-22.646-1.196-29.701,7.736l-30.805,39.005
+		c1.328-11.305-5.015-22.116-15.531-26.472L337.004,1.939c-10.516-4.356-22.646-1.196-29.701,7.736l-30.805,39.005
-		c-4.908,6.215-12.43,9.665-20.349,9.668c-0.084,0-0.168,0-0.252,0c-7.935,0.014-15.477-3.44-20.395-9.667L204.697,9.675
+		c-4.908,6.215-12.43,9.665-20.349,9.668c-0.084,0-0.168,0-0.252,0c-7.935,0.014-15.477-3.44-20.395-9.667L204.697,9.675
-		c-7.055-8.933-19.185-12.092-29.702-7.736L133.63,19.072c-10.516,4.356-16.86,15.167-15.532,26.473l5.799,49.366
+		c-7.055-8.933-19.185-12.092-29.702-7.736L133.63,19.072c-10.516,4.356-16.86,15.167-15.532,26.473l5.799,49.366
-		c0.926,7.881-1.964,15.656-7.585,21.257c-0.059,0.059-0.118,0.118-0.178,0.178c-5.602,5.598-13.36,8.477-21.226,7.552
+		c0.926,7.881-1.964,15.656-7.585,21.257c-0.059,0.059-0.118,0.118-0.178,0.178c-5.602,5.598-13.36,8.477-21.226,7.552
-		l-49.363-5.799c-11.305-1.328-22.116,5.015-26.472,15.531L1.939,174.996c-4.356,10.516-1.196,22.646,7.736,29.701l39.006,30.805
+		l-49.363-5.799c-11.305-1.328-22.116,5.015-26.472,15.531L1.939,174.996c-4.356,10.516-1.196,22.646,7.736,29.701l39.006,30.805
-		c6.215,4.908,9.665,12.429,9.668,20.348c0,0.084,0,0.167,0,0.251c0.014,7.935-3.44,15.477-9.667,20.395L9.675,307.303
+		c6.215,4.908,9.665,12.429,9.668,20.348c0,0.084,0,0.167,0,0.251c0.014,7.935-3.44,15.477-9.667,20.395L9.675,307.303
-		c-8.933,7.055-12.092,19.185-7.736,29.701l17.134,41.365c4.356,10.516,15.168,16.86,26.472,15.532l49.366-5.799
+		c-8.933,7.055-12.092,19.185-7.736,29.701l17.134,41.365c4.356,10.516,15.168,16.86,26.472,15.532l49.366-5.799
-		c7.882-0.926,15.656,1.965,21.258,7.586c0.059,0.059,0.118,0.119,0.178,0.178c5.597,5.603,8.476,13.36,7.552,21.226l-5.799,49.364
+		c7.882-0.926,15.656,1.965,21.258,7.586c0.059,0.059,0.118,0.119,0.178,0.178c5.597,5.603,8.476,13.36,7.552,21.226l-5.799,49.364
-		c-1.328,11.305,5.015,22.116,15.532,26.472l41.366,17.134c10.516,4.356,22.646,1.196,29.701-7.736l30.804-39.005
+		c-1.328,11.305,5.015,22.116,15.532,26.472l41.366,17.134c10.516,4.356,22.646,1.196,29.701-7.736l30.804-39.005
-		c4.908-6.215,12.43-9.665,20.348-9.669c0.084,0,0.168,0,0.251,0c7.936-0.014,15.478,3.44,20.396,9.667l30.806,39.007
+		c4.908-6.215,12.43-9.665,20.348-9.669c0.084,0,0.168,0,0.251,0c7.936-0.014,15.478,3.44,20.396,9.667l30.806,39.007
-		c7.055,8.933,19.185,12.093,29.701,7.736l41.366-17.134c10.516-4.356,16.86-15.168,15.532-26.472l-5.8-49.366
+		c7.055,8.933,19.185,12.093,29.701,7.736l41.366-17.134c10.516-4.356,16.86-15.168,15.532-26.472l-5.8-49.366
-		c-0.926-7.881,1.965-15.656,7.586-21.257c0.059-0.059,0.119-0.119,0.178-0.178c5.602-5.597,13.36-8.476,21.225-7.552l49.364,5.799
+		c-0.926-7.881,1.965-15.656,7.586-21.257c0.059-0.059,0.119-0.119,0.178-0.178c5.602-5.597,13.36-8.476,21.225-7.552l49.364,5.799
-		c11.305,1.328,22.117-5.015,26.472-15.531l17.134-41.365C514.418,326.488,511.258,314.358,502.325,307.303z M281.292,329.698
+		c11.305,1.328,22.117-5.015,26.472-15.531l17.134-41.365C514.418,326.488,511.258,314.358,502.325,307.303z M281.292,329.698
-		c-39.68,16.436-85.172-2.407-101.607-42.087c-16.436-39.68,2.407-85.171,42.087-101.608c39.68-16.436,85.172,2.407,101.608,42.088
+		c-39.68,16.436-85.172-2.407-101.607-42.087c-16.436-39.68,2.407-85.171,42.087-101.608c39.68-16.436,85.172,2.407,101.608,42.088
-		C339.815,267.771,320.972,313.262,281.292,329.698z"/>
+		C339.815,267.771,320.972,313.262,281.292,329.698z"/>
-</g>
+</g>
 </svg>

abogen/constants.py

@@ -29,57 +29,6 @@ LANGUAGE_DESCRIPTIONS = {
     "z": "Mandarin Chinese",
 }
 
-# Mapping from Kokoro single-letter language codes to ISO 3166-1 alpha-2 country codes
-# Used for loading flag icons
-KOKORO_LANG_TO_COUNTRY = {
-    "a": "us",  # American English -> United States
-    "b": "gb",  # British English -> United Kingdom
-    "e": "es",  # Spanish -> Spain
-    "f": "fr",  # French -> France
-    "h": "in",  # Hindi -> India
-    "i": "it",  # Italian -> Italy
-    "j": "jp",  # Japanese -> Japan
-    "p": "br",  # Brazilian Portuguese -> Brazil
-    "z": "cn",  # Mandarin Chinese -> China
-}
-
-# Mapping from Supertonic ISO 639-1 language codes to ISO 3166-1 alpha-2 country codes
-# Used for loading flag icons in the Supertonic language picker
-SUPERTONIC_LANG_TO_COUNTRY = {
-    "en": "gb",
-    "ko": "kr",
-    "ja": "jp",
-    "ar": "ae",
-    "bg": "bg",
-    "cs": "cz",
-    "da": "dk",
-    "de": "de",
-    "el": "gr",
-    "es": "es",
-    "et": "ee",
-    "fi": "fi",
-    "fr": "fr",
-    "hi": "in",
-    "hr": "hr",
-    "hu": "hu",
-    "id": "id",
-    "it": "it",
-    "lt": "lt",
-    "lv": "lv",
-    "nl": "nl",
-    "pl": "pl",
-    "pt": "pt",
-    "ro": "ro",
-    "ru": "ru",
-    "sk": "sk",
-    "sl": "si",
-    "sv": "se",
-    "tr": "tr",
-    "uk": "ua",
-    "vi": "vn",
-    "na": "na",
-}
-
 # Supported sound formats
 SUPPORTED_SOUND_FORMATS = [
     "wav",
@@ -114,64 +63,6 @@ SUPPORTED_INPUT_FORMATS = [
 # 384 if self.lang_code in 'ab':
 SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = list(LANGUAGE_DESCRIPTIONS.keys())
 
-# Voice and sample text constants
-VOICES_INTERNAL = [
-    "af_alloy",
-    "af_aoede",
-    "af_bella",
-    "af_heart",
-    "af_jessica",
-    "af_kore",
-    "af_nicole",
-    "af_nova",
-    "af_river",
-    "af_sarah",
-    "af_sky",
-    "am_adam",
-    "am_echo",
-    "am_eric",
-    "am_fenrir",
-    "am_liam",
-    "am_michael",
-    "am_onyx",
-    "am_puck",
-    "am_santa",
-    "bf_alice",
-    "bf_emma",
-    "bf_isabella",
-    "bf_lily",
-    "bm_daniel",
-    "bm_fable",
-    "bm_george",
-    "bm_lewis",
-    "ef_dora",
-    "em_alex",
-    "em_santa",
-    "ff_siwis",
-    "hf_alpha",
-    "hf_beta",
-    "hm_omega",
-    "hm_psi",
-    "if_sara",
-    "im_nicola",
-    "jf_alpha",
-    "jf_gongitsune",
-    "jf_nezumi",
-    "jf_tebukuro",
-    "jm_kumo",
-    "pf_dora",
-    "pm_alex",
-    "pm_santa",
-    "zf_xiaobei",
-    "zf_xiaoni",
-    "zf_xiaoxiao",
-    "zf_xiaoyi",
-    "zm_yunjian",
-    "zm_yunxi",
-    "zm_yunxia",
-    "zm_yunyang",
-]
-
 # Voice and sample text mapping
 SAMPLE_VOICE_TEXTS = {
     "a": "This is a sample of the selected voice.",

abogen/domain/audio_buffer.py

@@ -0,0 +1,172 @@
+"""Audio buffer operations for audiobook generation.
+
+This module provides core audio buffer manipulation functions including:
+- Silence generation
+- Audio mixing
+- Audio normalization
+- Audio buffer resizing
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+import numpy as np
+
+# Standard sample rate used throughout the application
+SAMPLE_RATE = 24000
+
+
+def create_silence(duration_seconds: float) -> np.ndarray:
+    """Create a silence audio buffer.
+    
+    Args:
+        duration_seconds: Duration of silence in seconds.
+        
+    Returns:
+        Numpy array of float32 zeros with length = duration_seconds * SAMPLE_RATE.
+        Returns empty array if duration is <= 0.
+    """
+    if duration_seconds <= 0:
+        return np.array([], dtype="float32")
+    
+    samples = int(round(duration_seconds * SAMPLE_RATE))
+    if samples <= 0:
+        return np.array([], dtype="float32")
+    
+    return np.zeros(samples, dtype="float32")
+
+
+def mix_audio(
+    target: np.ndarray,
+    source: np.ndarray,
+    start_sample: int,
+    end_sample: Optional[int] = None,
+) -> np.ndarray:
+    """Mix source audio into target buffer at specified position.
+    
+    This performs additive mixing (target += source). The target buffer
+    is extended if necessary to accommodate the source audio.
+    
+    Args:
+        target: The target audio buffer to mix into.
+        source: The source audio buffer to mix.
+        start_sample: Starting sample index in target buffer.
+        end_sample: Optional end sample index. If None, calculated from source length.
+        
+    Returns:
+        The target buffer (possibly extended). If target was extended, returns new array.
+    """
+    if source.size == 0:
+        return target
+    
+    if end_sample is None:
+        end_sample = start_sample + len(source)
+    
+    # Extend target buffer if needed
+    if end_sample > len(target):
+        new_length = end_sample
+        new_target = np.concatenate([
+            target,
+            np.zeros(new_length - len(target), dtype="float32")
+        ])
+        target = new_target
+    
+    # Perform the mix (additive)
+    target[start_sample:end_sample] += source
+    return target
+
+
+def normalize_audio(
+    audio: np.ndarray,
+    target_peak: float = 1.0,
+) -> np.ndarray:
+    """Normalize audio buffer to prevent clipping.
+    
+    If the audio exceeds the target peak (default 1.0), it is scaled down
+    proportionally to prevent distortion.
+    
+    Args:
+        audio: Input audio buffer.
+        target_peak: Target maximum amplitude (default 1.0).
+        
+    Returns:
+        Normalized audio buffer (new array, original is not modified).
+    """
+    if audio.size == 0:
+        return audio.copy()
+    
+    max_amplitude = float(np.abs(audio).max())
+    
+    if max_amplitude <= target_peak:
+        return audio.copy()
+    
+    # Scale down to prevent clipping
+    scale_factor = target_peak / max_amplitude
+    return (audio * scale_factor).astype("float32")
+
+
+def ensure_buffer_size(
+    buffer: np.ndarray,
+    min_samples: int,
+) -> np.ndarray:
+    """Ensure audio buffer is at least min_samples long.
+    
+    If buffer is shorter, it is extended with zeros.
+    
+    Args:
+        buffer: Input audio buffer.
+        min_samples: Minimum required length in samples.
+        
+    Returns:
+        Buffer of at least min_samples length (new array if extended).
+    """
+    if len(buffer) >= min_samples:
+        return buffer
+    
+    new_buffer = np.zeros(min_samples, dtype="float32")
+    new_buffer[:len(buffer)] = buffer
+    return new_buffer
+
+
+def concatenate_audio(*buffers: np.ndarray) -> np.ndarray:
+    """Concatenate multiple audio buffers.
+    
+    Args:
+        *buffers: Audio buffers to concatenate.
+        
+    Returns:
+        Single concatenated audio buffer.
+    """
+    non_empty = [b for b in buffers if b.size > 0]
+    if not non_empty:
+        return np.array([], dtype="float32")
+    return np.concatenate(non_empty)
+
+
+def audio_duration(audio: np.ndarray, sample_rate: int = SAMPLE_RATE) -> float:
+    """Calculate duration of audio buffer in seconds.
+    
+    Args:
+        audio: Audio buffer.
+        sample_rate: Sample rate in Hz (default SAMPLE_RATE).
+        
+    Returns:
+        Duration in seconds.
+    """
+    return len(audio) / sample_rate
+
+
+def samples_for_duration(duration_seconds: float, sample_rate: int = SAMPLE_RATE) -> int:
+    """Calculate number of samples for a given duration.
+    
+    Args:
+        duration_seconds: Duration in seconds.
+        sample_rate: Sample rate in Hz (default SAMPLE_RATE).
+        
+    Returns:
+        Number of samples (rounded to nearest integer), or 0 if duration is <= 0.
+    """
+    if duration_seconds <= 0:
+        return 0
+    return int(round(duration_seconds * sample_rate))

abogen/domain/audio_helpers.py

@@ -0,0 +1,118 @@
+"""Audio helper utilities.
+
+Functions for building ffmpeg commands, converting audio formats,
+and applying chapter metadata to MP4 files.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import numpy as np
+
+
+SAMPLE_RATE = 24000
+
+
+def build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str]] = None) -> list[str]:
+    from abogen.infrastructure.exporters import ExportService
+
+    base = [
+        "ffmpeg",
+        "-y",
+        "-f",
+        "f32le",
+        "-ar",
+        str(SAMPLE_RATE),
+        "-ac",
+        "1",
+        "-i",
+        "pipe:0",
+    ]
+    if fmt == "mp3":
+        base += ["-c:a", "libmp3lame", "-qscale:a", "2"]
+    elif fmt == "opus":
+        base += ["-c:a", "libopus", "-b:a", "24000"]
+    elif fmt == "m4b":
+        base += ["-c:a", "aac", "-q:a", "2", "-movflags", "+faststart+use_metadata_tags"]
+    else:
+        base += ["-c:a", "copy"]
+
+    if metadata:
+        svc = ExportService()
+        base.extend(svc._metadata_to_ffmpeg_args(metadata))
+    base.append(str(path))
+    return base
+
+
+def to_float32(audio_segment) -> np.ndarray:
+    if audio_segment is None:
+        return np.zeros(0, dtype="float32")
+
+    tensor = audio_segment
+    if hasattr(tensor, "detach"):
+        tensor = tensor.detach()
+    if hasattr(tensor, "cpu"):
+        try:
+            tensor = tensor.cpu()
+        except Exception:
+            pass
+    if hasattr(tensor, "numpy"):
+        return np.asarray(tensor.numpy(), dtype="float32").reshape(-1)
+    return np.asarray(tensor, dtype="float32").reshape(-1)
+
+
+def apply_m4b_chapters_with_mutagen(
+    audio_path: Path,
+    chapters: List[Dict[str, Any]],
+) -> bool:
+    """Apply chapter atoms to an MP4/M4B file using mutagen.
+
+    Returns True if chapters were written, False otherwise.
+    Raises ImportError if mutagen is not installed.
+    """
+    if not chapters:
+        return False
+
+    from fractions import Fraction
+    from mutagen.mp4 import MP4, MP4Chapter  # type: ignore[import]
+
+    mp4 = MP4(str(audio_path))
+
+    chapter_objects: List[MP4Chapter] = []
+    for index, entry in enumerate(sorted(chapters, key=lambda item: float(item.get("start") or 0.0))):
+        start_raw = entry.get("start")
+        if start_raw is None:
+            continue
+        try:
+            start_seconds = max(0.0, float(start_raw))
+        except (TypeError, ValueError):
+            continue
+
+        title_value = entry.get("title")
+        title_text = str(title_value) if title_value else f"Chapter {index + 1}"
+
+        start_fraction = Fraction(int(round(start_seconds * 1000)), 1000)
+        chapter_atom = MP4Chapter(start_fraction, title_text)
+
+        end_raw = entry.get("end")
+        if end_raw is not None:
+            try:
+                end_seconds = float(end_raw)
+            except (TypeError, ValueError):
+                end_seconds = None
+            if end_seconds is not None and end_seconds > start_seconds:
+                chapter_atom.end = Fraction(int(round(end_seconds * 1000)), 1000)
+
+        chapter_objects.append(chapter_atom)
+
+    if not chapter_objects:
+        return False
+
+    from typing import cast
+
+    mp4.chapters = cast(Any, chapter_objects)
+    mp4.save()
+
+    return True

abogen/domain/audio_sink.py

@@ -0,0 +1,131 @@
+"""Audio sink abstraction for unified audio output.
+
+Provides a context-manager-based abstraction for writing audio data
+to various output formats (WAV, FLAC via soundfile; compressed via ffmpeg).
+
+Usage:
+    with open_audio_sink(path, "wav") as sink:
+        sink.write(audio_data)
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Optional
+
+import numpy as np
+
+from abogen.domain.audio_buffer import SAMPLE_RATE
+from abogen.domain.audio_helpers import build_ffmpeg_command
+
+
+@dataclass(frozen=True)
+class AudioSink:
+    """Represents an open audio output target."""
+
+    write: Callable[[np.ndarray], None]
+    close: Callable[[], None]
+
+    def __enter__(self) -> AudioSink:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.close()
+
+
+def _ensure_ffmpeg() -> None:
+    """Ensure static ffmpeg binaries are on PATH."""
+    import static_ffmpeg  # type: ignore
+
+    ffmpeg_cache_root = _get_ffmpeg_cache_root()
+    platform_cache = os.path.join(ffmpeg_cache_root, sys.platform)
+    os.makedirs(platform_cache, exist_ok=True)
+    try:
+        import static_ffmpeg.run as static_ffmpeg_run  # type: ignore
+
+        static_ffmpeg_run.LOCK_FILE = os.path.join(ffmpeg_cache_root, "lock.file")
+    except Exception:
+        pass
+    static_ffmpeg.add_paths(weak=True, download_dir=platform_cache)
+
+
+def _get_ffmpeg_cache_root() -> str:
+    from abogen.infrastructure.cache import get_internal_cache_path
+
+    return get_internal_cache_path("ffmpeg")
+
+
+def open_audio_sink(
+    path: Path,
+    fmt: str,
+    *,
+    metadata: Optional[dict[str, str]] = None,
+    cancel_check: Optional[Callable[[], bool]] = None,
+    extra_ffmpeg_args: Optional[list[str]] = None,
+    ffmpeg_cmd: Optional[list[str]] = None,
+) -> AudioSink:
+    """Open an audio output sink for writing raw float32 PCM samples.
+
+    Args:
+        path: Output file path.
+        fmt: Output format ("wav", "flac", "mp3", "opus", "m4b").
+        metadata: Optional metadata dict (ignored when ffmpeg_cmd is provided).
+        cancel_check: Optional callable; if it returns True, writes are silently skipped.
+        extra_ffmpeg_args: Optional extra args inserted after ffmpeg header (ignored when ffmpeg_cmd is provided).
+        ffmpeg_cmd: Optional pre-built ffmpeg command list (for m4b with cover art etc.).
+
+    Returns:
+        AudioSink with write() and close() methods.
+    """
+    fmt = fmt.lower()
+
+    if fmt in {"wav", "flac"}:
+        import soundfile as sf
+
+        soundfile_obj = sf.SoundFile(
+            path,
+            mode="w",
+            samplerate=SAMPLE_RATE,
+            channels=1,
+            format=fmt.upper(),
+        )
+
+        def _write_wav(data: np.ndarray) -> None:
+            if cancel_check and cancel_check():
+                return
+            soundfile_obj.write(data)
+
+        def _close_wav() -> None:
+            soundfile_obj.close()
+
+        return AudioSink(write=_write_wav, close=_close_wav)
+
+    # Compressed formats: pipe through ffmpeg
+    _ensure_ffmpeg()
+
+    if ffmpeg_cmd is not None:
+        cmd = list(ffmpeg_cmd)
+    else:
+        cmd = build_ffmpeg_command(path, fmt, metadata=metadata)
+        if extra_ffmpeg_args:
+            cmd[2:2] = extra_ffmpeg_args
+
+    process = subprocess.Popen(
+        cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
+    )
+
+    def _write_compressed(data: np.ndarray) -> None:
+        if (cancel_check and cancel_check()) or process.stdin is None or process.stdin.closed:
+            return
+        process.stdin.write(data.tobytes())
+
+    def _close_compressed() -> None:
+        if process.stdin and not process.stdin.closed:
+            process.stdin.close()
+        process.wait()
+
+    return AudioSink(write=_write_compressed, close=_close_compressed)

abogen/domain/chapter_classification.py

@@ -0,0 +1,131 @@
+"""Heuristics for classifying chapters as content vs. supplements.
+
+A 'supplement' is any non-story material that a listener would typically
+skip: title page, copyright, table of contents, acknowledgements, etc.
+The scoring functions return a float; higher ⇒ more likely to be a
+supplement.  ``should_preselect_chapter`` turns that score into a
+boolean suitable for a web form default.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List, Tuple
+
+# Compiled once at module load – these are immutable.
+
+_SUPPLEMENT_TITLE_PATTERNS: List[Tuple[re.Pattern[str], float]] = [
+    (re.compile(r"\btitle\s+page\b"), 3.0),
+    (re.compile(r"\bcopyright\b"), 2.4),
+    (re.compile(r"\btable\s+of\s+contents\b"), 2.8),
+    (re.compile(r"\bcontents\b"), 2.0),
+    (re.compile(r"\backnowledg(e)?ments?\b"), 2.0),
+    (re.compile(r"\bdedication\b"), 2.0),
+    (re.compile(r"\babout\s+the\s+author(s)?\b"), 2.4),
+    (re.compile(r"\balso\s+by\b"), 2.0),
+    (re.compile(r"\bpraise\s+for\b"), 2.0),
+    (re.compile(r"\bcolophon\b"), 2.2),
+    (re.compile(r"\bpublication\s+data\b"), 2.2),
+    (re.compile(r"\btranscriber'?s?\s+note\b"), 2.2),
+    (re.compile(r"\bglossary\b"), 2.2),
+    (re.compile(r"\bindex\b"), 2.0),
+    (re.compile(r"\bbibliograph(y|ies)\b"), 2.0),
+    (re.compile(r"\breferences\b"), 1.8),
+    (re.compile(r"\bappendix\b"), 1.9),
+]
+
+_CONTENT_TITLE_PATTERNS: List[re.Pattern[str]] = [
+    re.compile(r"\bchapter\b"),
+    re.compile(r"\bbook\b"),
+    re.compile(r"\bpart\b"),
+    re.compile(r"\bsection\b"),
+    re.compile(r"\bscene\b"),
+    re.compile(r"\bprologue\b"),
+    re.compile(r"\bepilogue\b"),
+    re.compile(r"\bintroduction\b"),
+    re.compile(r"\bstory\b"),
+]
+
+_SUPPLEMENT_TEXT_KEYWORDS: List[Tuple[str, float]] = [
+    ("copyright", 1.2),
+    ("all rights reserved", 1.1),
+    ("isbn", 0.9),
+    ("library of congress", 1.0),
+    ("table of contents", 1.0),
+    ("dedicated to", 0.8),
+    ("acknowledg", 0.8),
+    ("printed in", 0.6),
+    ("permission", 0.6),
+    ("publisher", 0.5),
+    ("praise for", 0.9),
+    ("also by", 0.9),
+    ("glossary", 0.8),
+    ("index", 0.8),
+    ("newsletter", 3.2),
+    ("mailing list", 2.6),
+    ("sign-up", 2.2),
+]
+
+
+def supplement_score(title: str, text: str, index: int) -> float:
+    """Return a score indicating how likely *title*/*text* is a supplement.
+
+    Higher values ⇒ more likely to be non-story material (title page,
+    copyright, acknowledgements, etc.).
+    """
+    normalized_title = (title or "").lower()
+    score = 0.0
+
+    for pattern, weight in _SUPPLEMENT_TITLE_PATTERNS:
+        if pattern.search(normalized_title):
+            score += weight
+
+    for pattern in _CONTENT_TITLE_PATTERNS:
+        if pattern.search(normalized_title):
+            score -= 2.0
+
+    stripped_text = (text or "").strip()
+    length = len(stripped_text)
+    if length <= 150:
+        score += 0.9
+    elif length <= 400:
+        score += 0.6
+    elif length <= 800:
+        score += 0.35
+
+    lowercase_text = stripped_text.lower()
+    for keyword, weight in _SUPPLEMENT_TEXT_KEYWORDS:
+        if keyword in lowercase_text:
+            score += weight
+
+    if index == 0 and score > 0:
+        score += 0.25
+
+    return score
+
+
+def should_preselect_chapter(
+    title: str,
+    text: str,
+    index: int,
+    total_count: int,
+) -> bool:
+    """Return True if the chapter should be *enabled* by default in the form.
+
+    A single chapter is always preselected.  For multi-chapter books, the
+    chapter is preselected when its supplement score is below 1.9.
+    """
+    if total_count <= 1:
+        return True
+    score = supplement_score(title, text, index)
+    return score < 1.9
+
+
+def ensure_at_least_one_chapter_enabled(chapters: List[Dict[str, Any]]) -> None:
+    """Mutate *chapters* in-place so that at least one has ``enabled=True``."""
+    if not chapters:
+        return
+    if any(chapter.get("enabled") for chapter in chapters):
+        return
+    best_index = max(range(len(chapters)), key=lambda idx: chapters[idx].get("characters", 0))
+    chapters[best_index]["enabled"] = True

abogen/domain/chapter_overrides.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Tuple
+
+from abogen.text_extractor import ExtractedChapter
+from abogen.domain.voice_utils import coerce_truthy
+
+
+def apply_chapter_overrides(
+    extracted: List[ExtractedChapter],
+    overrides: List[Dict[str, Any]],
+) -> Tuple[List[ExtractedChapter], Dict[str, str], List[str]]:
+    if not overrides:
+        return [], {}, []
+
+    selected: List[ExtractedChapter] = []
+    metadata_updates: Dict[str, str] = {}
+    diagnostics: List[str] = []
+
+    for position, payload in enumerate(overrides):
+        if not isinstance(payload, dict):
+            diagnostics.append(
+                f"Skipped chapter override at position {position + 1}: unsupported payload type {type(payload).__name__}."
+            )
+            continue
+
+        enabled = coerce_truthy(payload.get("enabled", True))
+        payload["enabled"] = enabled
+        if not enabled:
+            continue
+
+        metadata_payload = payload.get("metadata") or {}
+        if isinstance(metadata_payload, dict):
+            for key, value in metadata_payload.items():
+                if value is None:
+                    continue
+                metadata_updates[str(key)] = str(value)
+
+        base: Optional[ExtractedChapter] = None
+        idx_candidate = payload.get("index")
+        idx_normalized: Optional[int] = None
+        if isinstance(idx_candidate, int):
+            idx_normalized = idx_candidate
+        elif isinstance(idx_candidate, str):
+            try:
+                idx_normalized = int(idx_candidate)
+            except ValueError:
+                idx_normalized = None
+        if idx_normalized is not None and 0 <= idx_normalized < len(extracted):
+            base = extracted[idx_normalized]
+            payload["index"] = idx_normalized
+
+        if base is None:
+            source_title = payload.get("source_title")
+            if isinstance(source_title, str):
+                base = next((chapter for chapter in extracted if chapter.title == source_title), None)
+
+        if base is None:
+            candidate_title = payload.get("title")
+            if isinstance(candidate_title, str):
+                base = next((chapter for chapter in extracted if chapter.title == candidate_title), None)
+
+        text_override = payload.get("text")
+        if text_override is not None:
+            text_value = str(text_override)
+        elif base is not None:
+            text_value = base.text
+        else:
+            diagnostics.append(
+                f"Skipped chapter override at position {position + 1}: no text provided and no matching source chapter found."
+            )
+            continue
+
+        title_override = payload.get("title")
+        if title_override is not None:
+            title_value = str(title_override)
+        elif base is not None:
+            title_value = base.title
+        else:
+            title_value = f"Chapter {position + 1}"
+
+        if base and not payload.get("source_title"):
+            payload["source_title"] = base.title
+
+        payload["title"] = title_value
+        payload["text"] = text_value
+        payload["characters"] = len(text_value)
+        payload.setdefault("order", payload.get("order", position))
+
+        selected.append(ExtractedChapter(title=title_value, text=text_value))
+
+    return selected, metadata_updates, diagnostics

abogen/domain/chapter_titles.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+import re
+from typing import List, Tuple
+
+
+_HEADING_SANITIZE_RE = re.compile(r"[^a-z0-9]+")
+_HEADING_NUMBER_PREFIX_RE = re.compile(
+    r"^\s*(?P<number>(?:\d+|[ivxlcdm]+))(?P<suffix>(?:[\s.:;-].*)?)$",
+    re.IGNORECASE,
+)
+_ACRONYM_ALLOWLIST = {
+    "AI", "API", "CPU", "DIY", "GPU", "HTML", "HTTP", "HTTPS", "ID",
+    "JSON", "MP3", "MP4", "M4B", "NASA", "OCR", "PDF", "SQL", "TV",
+    "TTS", "UK", "UN", "UFO", "OK", "URL", "USA", "US", "VR",
+}
+_ROMAN_NUMERAL_CHARS = frozenset("IVXLCDM")
+_CAPS_WORD_RE = re.compile(r"[A-Z][A-Z0-9'\u2019-]*")
+
+
+def simplify_heading_text(text: str) -> str:
+    raw = str(text or "").strip().lower()
+    if not raw:
+        return ""
+    simplified = _HEADING_SANITIZE_RE.sub("", raw)
+    if simplified.startswith("chapter"):
+        simplified = simplified[7:]
+    return simplified
+
+
+def headings_equivalent(left: str, right: str) -> bool:
+    simple_left = simplify_heading_text(left)
+    simple_right = simplify_heading_text(right)
+    if not simple_left or not simple_right:
+        return False
+    if simple_left == simple_right:
+        return True
+    if simple_right.startswith(simple_left):
+        return True
+    if simple_left.startswith(simple_right):
+        return True
+    if len(simple_left) > 5 and simple_left in simple_right:
+        return True
+    return False
+
+
+def strip_duplicate_heading_line(text: str, heading: str) -> Tuple[str, bool]:
+    source_text = str(text or "")
+    if not source_text:
+        return source_text, False
+    normalized_heading = simplify_heading_text(heading)
+    if not normalized_heading:
+        return source_text, False
+    lines = source_text.splitlines()
+    new_lines: List[str] = []
+    removed = False
+    for line in lines:
+        stripped = line.strip()
+        if not removed and stripped:
+            if headings_equivalent(stripped, heading):
+                removed = True
+                continue
+        new_lines.append(line)
+    if not removed:
+        return source_text, False
+    while new_lines and not new_lines[0].strip():
+        new_lines.pop(0)
+    return "\n".join(new_lines), True
+
+
+def normalize_caps_word(word: str) -> str:
+    upper = word.upper()
+    letters = [char for char in upper if char.isalpha()]
+    if not letters:
+        return word
+    if upper in _ACRONYM_ALLOWLIST:
+        return word
+    if len(letters) <= 1:
+        return word
+    if all(char in _ROMAN_NUMERAL_CHARS for char in letters) and len(letters) <= 7:
+        return word
+
+    parts = re.split(r"(['\-\u2019])", word)
+    normalized_parts: List[str] = []
+    for part in parts:
+        if part in {"'", "-", "\u2019"}:
+            normalized_parts.append(part)
+            continue
+        if not part:
+            continue
+        normalized_parts.append(part[0].upper() + part[1:].lower())
+    return "".join(normalized_parts) or word
+
+
+def normalize_chapter_opening_caps(text: str) -> Tuple[str, bool]:
+    if not text:
+        return text, False
+
+    leading_len = len(text) - len(text.lstrip())
+    leading = text[:leading_len]
+    working = text[leading_len:]
+    if not working:
+        return text, False
+
+    builder: List[str] = []
+    pos = 0
+    changed = False
+
+    while pos < len(working):
+        char = working[pos]
+        if char in "\r\n":
+            builder.append(working[pos:])
+            pos = len(working)
+            break
+        if char.isspace():
+            builder.append(char)
+            pos += 1
+            continue
+        if char.islower():
+            builder.append(working[pos:])
+            pos = len(working)
+            break
+        if not char.isalpha():
+            builder.append(char)
+            pos += 1
+            continue
+
+        match = _CAPS_WORD_RE.match(working, pos)
+        if not match:
+            builder.append(char)
+            pos += 1
+            continue
+
+        word = match.group(0)
+        if any(ch.islower() for ch in word):
+            builder.append(working[pos:])
+            pos = len(working)
+            break
+
+        normalized = normalize_caps_word(word)
+        if normalized != word:
+            changed = True
+        builder.append(normalized)
+        pos = match.end()
+
+    if pos < len(working):
+        builder.append(working[pos:])
+
+    if not changed:
+        return text, False
+
+    return leading + "".join(builder), True
+
+
+def format_spoken_chapter_title(title: str, index: int, apply_prefix: bool) -> str:
+    base = str(title or "").strip()
+    if not base:
+        return f"Chapter {index}" if apply_prefix else ""
+    if not apply_prefix:
+        return base
+    lowered = base.lower()
+    if lowered.startswith("chapter") and (len(lowered) == 7 or not lowered[7].isalpha()):
+        return base
+    match = _HEADING_NUMBER_PREFIX_RE.match(base)
+    if match:
+        number = match.group("number") or ""
+        suffix = match.group("suffix") or ""
+        cleaned_suffix = suffix.lstrip(" .,:;-_ \t\u2013\u2014\u00b7\u2022")
+        if cleaned_suffix:
+            return f"Chapter {number}. {cleaned_suffix}"
+        return f"Chapter {number}"
+    return base
+
+
+def apply_chapter_text_transforms(
+    text: str,
+    *,
+    heading_text: str,
+    raw_title: str,
+    strip_heading: bool,
+    normalize_caps: bool,
+) -> Tuple[str, bool, bool]:
+    """Strip duplicate heading and normalize opening caps.
+
+    Returns ``(text, heading_removed, caps_changed)``.
+    The caller is responsible for state updates (pending flags, logging,
+    dict mutation, ``continue``).
+    """
+    heading_removed = False
+    caps_changed = False
+
+    if strip_heading and heading_text:
+        text, heading_removed = strip_duplicate_heading_line(text, heading_text)
+        if not heading_removed and raw_title:
+            match = _HEADING_NUMBER_PREFIX_RE.match(raw_title)
+            if match:
+                number = match.group("number")
+                if number:
+                    text, heading_removed = strip_duplicate_heading_line(text, number)
+
+    if normalize_caps and text:
+        text, caps_changed = normalize_chapter_opening_caps(text)
+
+    return text, heading_removed, caps_changed

abogen/domain/chunk_utils.py

@@ -0,0 +1,75 @@
+"""Chunk processing utilities.
+
+Functions for grouping chunks, recording override usage, and selecting
+text for TTS synthesis.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any, Dict, Iterable, Mapping, Optional
+
+from abogen.pronunciation_store import increment_usage
+
+
+def safe_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def group_chunks_by_chapter(chunks: Iterable[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]:
+    grouped: Dict[int, List[Dict[str, Any]]] = defaultdict(list)
+    for entry in chunks or []:
+        if not isinstance(entry, dict):
+            continue
+        try:
+            chapter_index = int(entry.get("chapter_index", 0))
+        except (TypeError, ValueError):
+            chapter_index = 0
+        grouped[chapter_index].append(dict(entry))
+
+    for chapter_index, items in grouped.items():
+        items.sort(key=lambda payload: safe_int(payload.get("chunk_index")))
+
+    return grouped
+
+
+def record_override_usage(
+    job: Any,
+    usage_counter: Mapping[str, int],
+    token_map: Mapping[str, str],
+) -> None:
+    if not usage_counter:
+        return
+
+    language = getattr(job, "language", "") or "a"
+    for normalized, amount in usage_counter.items():
+        if amount <= 0:
+            continue
+        token_value = token_map.get(normalized, normalized)
+        try:
+            increment_usage(language=language, token=token_value, amount=int(amount))
+        except Exception:  # pragma: no cover - defensive logging
+            job.add_log(f"Failed to record usage for override {token_value}", level="warning")
+
+
+def chunk_text_for_tts(entry: Mapping[str, Any]) -> str:
+    """Choose the best source text for synthesis.
+
+    We must prefer the raw chunk text (``text`` / ``original_text``) so
+    manual/pronunciation overrides can match against the original tokens
+    (e.g. censored words like ``Unfu*k``).  ``normalized_text`` may have
+    already been run through ``normalize_for_pipeline``, which can remove
+    punctuation and prevent overrides from triggering.
+    """
+
+    if not isinstance(entry, Mapping):
+        return ""
+    return str(
+        entry.get("text")
+        or entry.get("original_text")
+        or entry.get("normalized_text")
+        or ""
+    ).strip()

abogen/domain/device.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import platform as _platform
+
+
+def select_device() -> str:
+    """Return the best available compute device (``"mps"``, ``"cuda"``, or ``"cpu"``).
+
+    Checks ``torch`` availability at runtime so this can be called from
+    any context without requiring torch at import time.
+    """
+    try:
+        import torch  # type: ignore[import-not-found]
+    except Exception:
+        return "cpu"
+
+    system = _platform.system()
+    if system == "Darwin" and _platform.processor() == "arm":
+        try:
+            if torch.backends.mps.is_available():  # type: ignore[union-attr]
+                return "mps"
+        except Exception:
+            pass
+        return "cpu"
+
+    try:
+        if torch.cuda.is_available():  # type: ignore[union-attr]
+            return "cuda"
+    except Exception:
+        pass
+    return "cpu"

abogen/domain/file_type.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Tuple
+
+from abogen.text_extractor import ExtractedChapter
+
+
+_SIGNIFICANT_LENGTH_THRESHOLDS: Dict[str, int] = {"epub": 1000, "markdown": 500}
+_MIN_SHORT_CONTENT: Dict[str, int] = {"epub": 240, "markdown": 160}
+_STRUCTURAL_KEYWORDS = (
+    "preface",
+    "prologue",
+    "introduction",
+    "foreword",
+    "epilogue",
+    "afterword",
+    "appendix",
+    "acknowledgment",
+    "acknowledgement",
+)
+_STRUCTURAL_MIN_LENGTH = 120
+_MAX_SHORT_CHAPTERS = 2
+
+
+@dataclass
+class ChapterFilterResult:
+    kept: List[ExtractedChapter]
+    skipped: List[Tuple[str, int]]
+
+
+def infer_file_type(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix == ".epub":
+        return "epub"
+    if suffix in {".md", ".markdown"}:
+        return "markdown"
+    if suffix == ".pdf":
+        return "pdf"
+    if suffix == ".txt":
+        return "text"
+    return suffix.lstrip(".") or "text"
+
+
+def looks_structural(title: str) -> bool:
+    lowered = title.strip().lower()
+    if not lowered:
+        return False
+    return any(keyword in lowered for keyword in _STRUCTURAL_KEYWORDS)
+
+
+def chapter_label(file_type: str) -> str:
+    return "chapters" if file_type.lower() in {"epub", "markdown"} else "pages"
+
+
+def auto_select_relevant_chapters(
+    chapters: List[ExtractedChapter],
+    file_type: str,
+) -> ChapterFilterResult:
+    if not chapters:
+        return ChapterFilterResult(kept=[], skipped=[])
+
+    normalized = file_type.lower()
+    threshold = _SIGNIFICANT_LENGTH_THRESHOLDS.get(normalized, 0)
+    min_short = _MIN_SHORT_CONTENT.get(normalized, 0)
+
+    kept: List[ExtractedChapter] = []
+    skipped: List[Tuple[str, int]] = []
+    short_kept = 0
+
+    for chapter in chapters:
+        stripped = chapter.text.strip()
+        length = len(stripped)
+        if length == 0:
+            skipped.append((chapter.title, length))
+            continue
+
+        keep = False
+        if threshold == 0:
+            keep = True
+        elif length >= threshold:
+            keep = True
+        elif not kept:
+            keep = True
+        elif min_short and length >= min_short and short_kept < _MAX_SHORT_CHAPTERS:
+            keep = True
+            short_kept += 1
+        elif looks_structural(chapter.title) and length >= _STRUCTURAL_MIN_LENGTH:
+            keep = True
+
+        if keep:
+            kept.append(chapter)
+        else:
+            skipped.append((chapter.title, length))
+
+    if kept:
+        return ChapterFilterResult(kept=kept, skipped=skipped)
+
+    longest_idx = None
+    longest_length = 0
+    for idx, chapter in enumerate(chapters):
+        stripped = chapter.text.strip()
+        if stripped and len(stripped) > longest_length:
+            longest_length = len(stripped)
+            longest_idx = idx
+
+    if longest_idx is not None:
+        longest = chapters[longest_idx]
+        fallback_skipped = [
+            (chapter.title, len(chapter.text.strip()))
+            for idx, chapter in enumerate(chapters)
+            if idx != longest_idx and chapter.text.strip()
+        ]
+        return ChapterFilterResult(kept=[longest], skipped=fallback_skipped)
+
+    return ChapterFilterResult(kept=[], skipped=skipped)
+
+
+def update_metadata_for_chapter_count(
+    metadata: Dict[str, Any], count: int, file_type: str
+) -> None:
+    if not metadata or count <= 0:
+        return
+
+    label = "Chapters" if file_type.lower() in {"epub", "markdown"} else "Pages"
+    metadata["chapter_count"] = str(count)
+
+    pattern = re.compile(r"\(\d+\s+(Chapters?|Pages?)\)")
+    replacement = f"({count} {label})"
+    for key in ("album", "ALBUM"):
+        value = metadata.get(key)
+        if not isinstance(value, str):
+            continue
+        metadata[key] = pattern.sub(replacement, value)

abogen/domain/metadata_extraction.py

@@ -0,0 +1,191 @@
+"""Metadata extraction and processing utilities.
+
+This module provides functions for extracting metadata from text content
+and generating ffmpeg metadata arguments.
+"""
+
+from __future__ import annotations
+
+import datetime
+import os
+import re
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+
+def extract_metadata_from_text(text: str) -> Dict[str, Optional[str]]:
+    """Extract metadata tags from text content.
+    
+    Looks for tags in format: <<METADATA_KEY:value>>
+    
+    Supported tags:
+    - TITLE, ARTIST, ALBUM, YEAR
+    - ALBUM_ARTIST, COMPOSER, GENRE
+    - COVER_PATH
+    
+    Args:
+        text: Text content to search for metadata tags.
+        
+    Returns:
+        Dictionary with extracted metadata values (None if not found).
+    """
+    metadata = {}
+    
+    patterns = {
+        "title": r"<<METADATA_TITLE:([^>]*)>>",
+        "artist": r"<<METADATA_ARTIST:([^>]*)>>",
+        "album": r"<<METADATA_ALBUM:([^>]*)>>",
+        "year": r"<<METADATA_YEAR:([^>]*)>>",
+        "album_artist": r"<<METADATA_ALBUM_ARTIST:([^>]*)>>",
+        "composer": r"<<METADATA_COMPOSER:([^>]*)>>",
+        "genre": r"<<METADATA_GENRE:([^>]*)>>",
+        "cover_path": r"<<METADATA_COVER_PATH:([^>]*)>>",
+    }
+    
+    for key, pattern in patterns.items():
+        match = re.search(pattern, text)
+        if match:
+            metadata[key] = match.group(1).strip()
+        else:
+            metadata[key] = None
+    
+    return metadata
+
+
+def get_filename_from_path(
+    file_path: str,
+    display_path: Optional[str] = None,
+    from_queue: bool = False,
+) -> str:
+    """Extract filename (without extension) from path.
+    
+    Args:
+        file_path: The file path to extract from.
+        display_path: Optional display path (used if from_queue is False).
+        from_queue: Whether the file is from queue.
+        
+    Returns:
+        Filename without extension.
+    """
+    if from_queue:
+        base_path = file_path
+    else:
+        base_path = display_path if display_path else file_path
+    
+    filename = os.path.splitext(os.path.basename(base_path))[0]
+    return filename
+
+
+def build_ffmpeg_metadata_args(
+    metadata: Dict[str, Optional[str]],
+    filename: str,
+) -> List[str]:
+    """Build ffmpeg metadata arguments from metadata dictionary.
+    
+    Args:
+        metadata: Dictionary with metadata keys and values.
+        filename: Fallback filename for title/album if not specified.
+        
+    Returns:
+        List of ffmpeg metadata arguments.
+    """
+    args = []
+    
+    # Default values
+    defaults = {
+        "title": filename,
+        "artist": "Unknown",
+        "album": filename,
+        "date": str(datetime.datetime.now().year),
+        "album_artist": "Unknown",
+        "composer": "Narrator",
+        "genre": "Audiobook",
+    }
+    
+    # Map of metadata keys to ffmpeg metadata keys
+    key_mapping = {
+        "title": "title",
+        "artist": "artist",
+        "album": "album",
+        "year": "date",  # year -> date for ffmpeg
+        "album_artist": "album_artist",
+        "composer": "composer",
+        "genre": "genre",
+    }
+    
+    for metadata_key, ffmpeg_key in key_mapping.items():
+        value = metadata.get(metadata_key)
+        if value is None:
+            value = defaults.get(metadata_key, "")
+        if value:
+            args.extend(["-metadata", f"{ffmpeg_key}={value}"])
+    
+    return args
+
+
+def extract_metadata_and_build_args(
+    text: str,
+    filename: str,
+    display_path: Optional[str] = None,
+    from_queue: bool = False,
+) -> Tuple[List[str], Optional[str]]:
+    """Extract metadata from text and build ffmpeg arguments.
+    
+    Convenience function that combines extract_metadata_from_text and
+    build_ffmpeg_metadata_args.
+    
+    Args:
+        text: Text content to search for metadata tags.
+        filename: Fallback filename for title/album.
+        display_path: Optional display path.
+        from_queue: Whether the file is from queue.
+        
+    Returns:
+        Tuple of (ffmpeg_metadata_args, cover_path).
+    """
+    metadata = extract_metadata_from_text(text)
+    cover_path = metadata.get("cover_path")
+    
+    # Get actual filename from path
+    actual_filename = get_filename_from_path(
+        file_path=filename,
+        display_path=display_path,
+        from_queue=from_queue,
+    )
+    
+    args = build_ffmpeg_metadata_args(metadata, actual_filename)
+    return args, cover_path
+
+
+def read_text_for_metadata(
+    file_path: str,
+    is_direct_text: bool,
+    direct_text: Optional[str] = None,
+    encoding: Optional[str] = None,
+) -> str:
+    """Read text content for metadata extraction.
+    
+    Args:
+        file_path: Path to file (or text if is_direct_text).
+        is_direct_text: Whether file_path contains direct text.
+        direct_text: Optional direct text (used if is_direct_text).
+        encoding: File encoding (detected if not provided).
+        
+    Returns:
+        Text content for metadata extraction.
+    """
+    if is_direct_text:
+        return direct_text or file_path
+    
+    # Read from file
+    actual_path = direct_text if direct_text else file_path
+    
+    try:
+        if encoding is None:
+            from abogen.utils import detect_encoding
+            encoding = detect_encoding(actual_path)
+        
+        with open(actual_path, "r", encoding=encoding, errors="replace") as f:
+            return f.read()
+    except Exception:
+        return ""

abogen/domain/metadata_helpers.py

@@ -0,0 +1,405 @@
+from __future__ import annotations
+
+import json
+import math
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Mapping, Optional, Tuple
+
+
+_SERIES_NAME_KEYS = (
+    "series",
+    "series_name",
+    "series_title",
+)
+_SERIES_NUMBER_KEYS = (
+    "series_index",
+    "series_position",
+    "series_sequence",
+    "book_number",
+    "series_number",
+)
+_SERIES_NUMBER_RE = re.compile(r"\d+(?:\.\d+)?")
+
+
+def normalize_metadata_map(values: Optional[Mapping[str, Any]]) -> Dict[str, str]:
+    normalized: Dict[str, str] = {}
+    if not values:
+        return normalized
+    for key, value in values.items():
+        if value is None:
+            continue
+        text = str(value).strip()
+        if not text:
+            continue
+        normalized[str(key).casefold()] = text
+    return normalized
+
+
+def format_author_sentence(raw: Optional[str]) -> str:
+    if raw is None:
+        return ""
+    normalized = str(raw).strip()
+    if not normalized:
+        return ""
+    lowered = normalized.casefold()
+    if lowered in {"unknown", "various"}:
+        return ""
+
+    working = normalized.replace("&", " and ")
+    segments = [segment.strip() for segment in working.split(",") if segment.strip()]
+    tokens: List[str] = []
+
+    if segments:
+        for segment in segments:
+            parts = [part.strip() for part in re.split(r"\band\b", segment, flags=re.IGNORECASE) if part.strip()]
+            if parts:
+                tokens.extend(parts)
+            else:
+                tokens.append(segment)
+    else:
+        parts = [part.strip() for part in re.split(r"\band\b", working, flags=re.IGNORECASE) if part.strip()]
+        tokens.extend(parts or [normalized])
+
+    cleaned = [token for token in tokens if token and token.casefold() not in {"unknown", "various"}]
+    if not cleaned:
+        return ""
+    if len(cleaned) == 1:
+        return f"By {cleaned[0]}"
+    if len(cleaned) == 2:
+        return f"By {cleaned[0]} and {cleaned[1]}"
+    return f"By {', '.join(cleaned[:-1])}, and {cleaned[-1]}"
+
+
+def ensure_sentence(text: str) -> str:
+    cleaned = text.strip()
+    if not cleaned:
+        return ""
+    if cleaned[-1] in ".!?":
+        return cleaned
+    return f"{cleaned}."
+
+
+def normalize_series_number(value: Any) -> Optional[str]:
+    text = str(value or "").strip()
+    if not text:
+        return None
+    candidate = text.replace(",", ".")
+    if candidate.replace(".", "", 1).isdigit():
+        if "." in candidate:
+            normalized = candidate.rstrip("0").rstrip(".")
+            return normalized or "0"
+        try:
+            return str(int(candidate))
+        except ValueError:
+            pass
+    match = _SERIES_NUMBER_RE.search(candidate)
+    if not match:
+        return None
+    normalized = match.group(0)
+    if "." in normalized:
+        normalized = normalized.rstrip("0").rstrip(".")
+        return normalized or "0"
+    try:
+        return str(int(normalized))
+    except ValueError:
+        return normalized
+
+
+def extract_series_metadata(values: Mapping[str, str]) -> Tuple[Optional[str], Optional[str]]:
+    series_name: Optional[str] = None
+    for key in _SERIES_NAME_KEYS:
+        raw = values.get(key)
+        if raw:
+            cleaned = str(raw).strip()
+            if cleaned:
+                series_name = cleaned
+                break
+
+    series_number: Optional[str] = None
+    for key in _SERIES_NUMBER_KEYS:
+        raw = values.get(key)
+        if raw is None:
+            continue
+        normalized = normalize_series_number(raw)
+        if normalized:
+            series_number = normalized
+            break
+
+    return series_name, series_number
+
+
+def format_series_sentence(series_name: Optional[str], series_number: Optional[str]) -> str:
+    if not series_name or not series_number:
+        return ""
+    name = series_name.strip()
+    number = series_number.strip()
+    if not name or not number:
+        return ""
+    article = "the " if not name.lower().startswith("the ") else ""
+    phrase = f"Book {number} of {article}{name}"
+    return re.sub(r"\s+", " ", phrase).strip()
+
+
+_PEOPLE_SPLIT_RE = re.compile(r"[;,/&]|\band\b", re.IGNORECASE)
+_LIST_SPLIT_RE = re.compile(r"[;,\n]")
+_SERIES_SEQUENCE_TAG_KEYS: Tuple[str, ...] = (
+    "series_index",
+    "series_position",
+    "series_sequence",
+    "series_number",
+    "seriesnumber",
+    "book_number",
+    "booknumber",
+)
+
+
+def normalize_metadata_casefold(values: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
+    normalized: Dict[str, Any] = {}
+    if not values:
+        return normalized
+    for key, value in values.items():
+        if value is None:
+            continue
+        key_text = str(key).strip().lower()
+        if not key_text:
+            continue
+        if isinstance(value, (list, tuple, set)):
+            normalized[key_text] = value
+        else:
+            text = str(value).strip()
+            if text:
+                normalized[key_text] = text
+    return normalized
+
+
+def split_people_field(raw: Any) -> List[str]:
+    if raw is None:
+        return []
+    if isinstance(raw, (list, tuple, set)):
+        results: List[str] = []
+        for item in raw:
+            results.extend(split_people_field(item))
+        return results
+    text = str(raw or "").strip()
+    if not text:
+        return []
+    tokens = [_token.strip() for _token in _PEOPLE_SPLIT_RE.split(text) if _token.strip()]
+    seen: set[str] = set()
+    ordered: List[str] = []
+    for token in tokens:
+        key = token.casefold()
+        if key in seen:
+            continue
+        seen.add(key)
+        ordered.append(token)
+    return ordered
+
+
+def split_simple_list(raw: Any) -> List[str]:
+    if raw is None:
+        return []
+    if isinstance(raw, (list, tuple, set)):
+        results: List[str] = []
+        for item in raw:
+            results.extend(split_simple_list(item))
+        return results
+    text = str(raw or "").strip()
+    if not text:
+        return []
+    tokens = [_token.strip() for _token in _LIST_SPLIT_RE.split(text) if _token.strip()]
+    seen: set[str] = set()
+    ordered: List[str] = []
+    for token in tokens:
+        key = token.casefold()
+        if key in seen:
+            continue
+        seen.add(key)
+        ordered.append(token)
+    return ordered
+
+
+def first_nonempty(*values: Any) -> Optional[str]:
+    for value in values:
+        if value is None:
+            continue
+        if isinstance(value, (list, tuple, set)):
+            items = list(value)
+            if not items:
+                continue
+            value = items[0]
+        text = str(value).strip()
+        if text:
+            return text
+    return None
+
+
+def extract_year(raw: Optional[str]) -> Optional[int]:
+    if not raw:
+        return None
+    text = str(raw).strip()
+    if not text:
+        return None
+    match = re.search(r"(19|20)\d{2}", text)
+    if match:
+        try:
+            return int(match.group(0))
+        except ValueError:
+            return None
+    try:
+        parsed = int(text)
+    except ValueError:
+        return None
+    if 0 < parsed < 3000:
+        return parsed
+    return None
+
+
+def normalize_series_sequence(raw: Any) -> Optional[str]:
+    if raw is None:
+        return None
+    if isinstance(raw, (int, float)):
+        if isinstance(raw, float) and (math.isnan(raw) or math.isinf(raw)):
+            return None
+        text = str(raw)
+    else:
+        text = str(raw).strip()
+    if not text:
+        return None
+    candidate = text.replace(",", ".")
+    match = _SERIES_NUMBER_RE.search(candidate)
+    if not match:
+        return None
+    normalized = match.group(0)
+    if "." in normalized:
+        normalized = normalized.rstrip("0").rstrip(".")
+        if not normalized:
+            normalized = "0"
+        return normalized
+    try:
+        return str(int(normalized))
+    except ValueError:
+        cleaned = normalized.lstrip("0")
+        return cleaned or "0"
+
+
+def build_audiobookshelf_metadata(
+    tags: Mapping[str, Any],
+    *,
+    language: str = "",
+    filename: str = "",
+) -> Dict[str, Any]:
+    normalized = normalize_metadata_casefold(tags)
+    title = first_nonempty(
+        normalized.get("title"),
+        normalized.get("book_title"),
+        normalized.get("name"),
+        normalized.get("album"),
+        filename,
+    )
+    authors = split_people_field(
+        normalized.get("authors")
+        or normalized.get("author")
+        or normalized.get("album_artist")
+        or normalized.get("artist")
+    )
+    narrators = split_people_field(normalized.get("narrators") or normalized.get("narrator"))
+    description = first_nonempty(
+        normalized.get("description"), normalized.get("summary"), normalized.get("comment")
+    )
+    genres = split_simple_list(normalized.get("genre"))
+    keywords = split_simple_list(normalized.get("tags") or normalized.get("keywords"))
+    lang = first_nonempty(normalized.get("language"), normalized.get("lang")) or language or ""
+    series_name = first_nonempty(
+        normalized.get("series"),
+        normalized.get("series_name"),
+        normalized.get("seriesname"),
+        normalized.get("series_title"),
+        normalized.get("seriestitle"),
+    )
+
+    series_sequence = None
+    for key in _SERIES_SEQUENCE_TAG_KEYS:
+        raw_value = normalized.get(key)
+        seq = normalize_series_sequence(raw_value)
+        if seq:
+            series_sequence = seq
+            break
+    if not series_name:
+        series_sequence = None
+
+    data: Dict[str, Any] = {
+        "title": title,
+        "subtitle": normalized.get("subtitle"),
+        "authors": authors,
+        "narrators": narrators,
+        "description": description,
+        "publisher": normalized.get("publisher"),
+        "genres": genres,
+        "tags": keywords,
+        "language": lang,
+        "publishedYear": extract_year(
+            normalized.get("published")
+            or normalized.get("publication_year")
+            or normalized.get("date")
+            or normalized.get("year")
+        ),
+        "seriesName": series_name,
+        "seriesSequence": series_sequence,
+        "isbn": first_nonempty(normalized.get("isbn"), normalized.get("asin")),
+    }
+    published_date = first_nonempty(
+        normalized.get("published"), normalized.get("publication_date"), normalized.get("date")
+    )
+    if published_date:
+        data["publishedDate"] = published_date
+
+    rating_text = first_nonempty(normalized.get("rating"), normalized.get("my_rating"))
+    if rating_text:
+        try:
+            data["rating"] = float(str(rating_text).strip())
+        except ValueError:
+            pass
+        rating_max_text = first_nonempty(
+            normalized.get("rating_max"), normalized.get("rating_scale")
+        )
+        if rating_max_text:
+            try:
+                data["ratingMax"] = float(str(rating_max_text).strip())
+            except ValueError:
+                pass
+
+    cleaned: Dict[str, Any] = {}
+    for key, value in data.items():
+        if value is None:
+            continue
+        if isinstance(value, str) and not value.strip():
+            continue
+        if isinstance(value, (list, tuple)) and not value:
+            continue
+        cleaned[key] = value
+    return cleaned
+
+
+def load_audiobookshelf_chapters(
+    metadata_path: Path,
+) -> Optional[List[Dict[str, Any]]]:
+    if not metadata_path.exists():
+        return None
+    try:
+        payload = json.loads(metadata_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    chapters = payload.get("chapters")
+    if not isinstance(chapters, list):
+        return None
+    cleaned: List[Dict[str, Any]] = []
+    for entry in chapters:
+        if not isinstance(entry, Mapping):
+            continue
+        title = first_nonempty(entry.get("title"), entry.get("original_title"))
+        start = entry.get("start")
+        end = entry.get("end")
+        if title and start is not None and end is not None:
+            cleaned.append({"title": str(title), "start": start, "end": end})
+    return cleaned or None

abogen/domain/metadata_merge.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+
+def merge_metadata(
+    extracted: Optional[Dict[str, Any]],
+    overrides: Optional[Dict[str, Any]],
+) -> Dict[str, str]:
+    merged: Dict[str, str] = {}
+    if extracted:
+        for key, value in extracted.items():
+            if value is None:
+                continue
+            merged[str(key)] = str(value)
+    if overrides:
+        for key, value in overrides.items():
+            key_str = str(key)
+            if value is None:
+                merged.pop(key_str, None)
+            else:
+                merged[key_str] = str(value)
+    return merged

abogen/domain/normalization.py

@@ -0,0 +1,96 @@
+"""Text normalization convenience helpers.
+
+Provides both the simple ``normalize_text_for_pipeline`` (apostrophe + LLM only)
+and the comprehensive ``prepare_text_for_tts`` that chains all three normalization
+stages used during conversion: heteronym rules → pronunciation rules → pipeline
+normalization.  The latter is the single entry point that both the Web UI and
+PyQt Desktop GUI should use.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Mapping, Optional
+
+from abogen.kokoro_text_normalization import (
+    ApostropheConfig,
+    normalize_for_pipeline as _normalize_for_pipeline,
+)
+from abogen.normalization_settings import (
+    build_apostrophe_config,
+    get_runtime_settings,
+    apply_overrides as _apply_overrides,
+)
+
+_BASE_APOSTROPHE_CONFIG = ApostropheConfig()
+
+
+def normalize_text_for_pipeline(
+    text: str,
+    *,
+    normalization_overrides: Optional[Mapping[str, Any]] = None,
+) -> str:
+    """Normalize text using runtime settings with optional overrides."""
+    runtime_settings = get_runtime_settings()
+    if normalization_overrides:
+        runtime_settings = _apply_overrides(runtime_settings, normalization_overrides)
+    apostrophe_config = build_apostrophe_config(settings=runtime_settings, base=_BASE_APOSTROPHE_CONFIG)
+    return _normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings)
+
+
+def prepare_text_for_tts(
+    text: str,
+    *,
+    heteronym_rules: Optional[List[Dict[str, Any]]] = None,
+    pronunciation_rules: Optional[List[Dict[str, Any]]] = None,
+    normalization_overrides: Optional[Mapping[str, Any]] = None,
+    usage_counter: Optional[Dict[str, int]] = None,
+) -> str:
+    """Apply the full text normalization pipeline before TTS synthesis.
+
+    Chains three stages in order:
+      1. Heteronym sentence rules (context-dependent pronunciation)
+      2. Pronunciation rules (token-level replacements)
+      3. Pipeline normalization (apostrophe handling, LLM normalization)
+
+    This is the **single entry point** that both the Web UI conversion runner
+    and the PyQt conversion thread should call before passing text to the TTS
+    backend.
+
+    Parameters
+    ----------
+    text:
+        Raw text to normalize.
+    heteronym_rules:
+        Compiled heteronym rules from ``compile_heteronym_sentence_rules``.
+    pronunciation_rules:
+        Compiled pronunciation rules from ``compile_pronunciation_rules``.
+    normalization_overrides:
+        User-level overrides for normalization settings (apostrophe mode, etc.).
+    usage_counter:
+        Mutable dict that tracks how many times each pronunciation override was
+        applied.  Passed through to ``apply_pronunciation_rules``.
+
+    Returns
+    -------
+    str
+        Fully normalized text ready for TTS.
+    """
+    from abogen.domain.pronunciation import (
+        apply_heteronym_sentence_rules,
+        apply_pronunciation_rules,
+    )
+
+    result = str(text or "")
+
+    if heteronym_rules:
+        result = apply_heteronym_sentence_rules(result, heteronym_rules)
+
+    if pronunciation_rules:
+        result = apply_pronunciation_rules(result, pronunciation_rules, usage_counter)
+
+    runtime_settings = get_runtime_settings()
+    if normalization_overrides:
+        runtime_settings = _apply_overrides(runtime_settings, normalization_overrides)
+    apostrophe_config = build_apostrophe_config(settings=runtime_settings, base=_BASE_APOSTROPHE_CONFIG)
+
+    return _normalize_for_pipeline(result, config=apostrophe_config, settings=runtime_settings)

abogen/domain/output_paths.py

@@ -0,0 +1,91 @@
+"""Output path resolution utilities.
+
+Pure functions for resolving output directories, building file paths,
+and computing project folder layouts.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable, List, Optional, Tuple
+
+from abogen.text_extractor import ExtractedChapter
+
+
+_OUTPUT_SANITIZE_RE = re.compile(r"[^\w\-_.]+")
+
+
+def slugify(title: str, index: int) -> str:
+    sanitized = re.sub(r"[^\w\-]+", "_", title.lower()).strip("_")
+    if not sanitized:
+        sanitized = f"chapter_{index:02d}"
+    return sanitized[:80]
+
+
+def sanitize_output_stem(name: str) -> str:
+    base = Path(name or "").stem
+    sanitized = _OUTPUT_SANITIZE_RE.sub("_", base).strip("_")
+    return sanitized or "output"
+
+
+def output_timestamp_token() -> str:
+    return datetime.now().strftime("%Y%m%d-%H%M%S")
+
+
+def build_output_path(directory: Path, original_name: str, extension: str) -> Path:
+    sanitized = sanitize_output_stem(original_name)
+    return directory / f"{sanitized}.{extension}"
+
+
+def apply_newline_policy(chapters: List[ExtractedChapter], replace_single_newlines: bool) -> None:
+    if not replace_single_newlines:
+        return
+    newline_regex = re.compile(r"(?<!\n)\n(?!\n)")
+    for chapter in chapters:
+        chapter.text = newline_regex.sub(" ", chapter.text)
+
+
+def resolve_output_directory(
+    *,
+    save_mode: str,
+    stored_path: Path,
+    output_folder: Optional[str],
+    desktop_dir: Optional[Path],
+    user_output_path: Optional[Path],
+    user_cache_outputs: Optional[Path],
+) -> Path:
+    if save_mode == "Save to Desktop" and desktop_dir:
+        return desktop_dir
+    if save_mode == "Save next to input file":
+        return stored_path.parent
+    if save_mode == "Choose output folder" and output_folder:
+        return Path(output_folder)
+    if save_mode == "Use default save location" and user_output_path:
+        return user_output_path
+    return user_cache_outputs or Path(".")
+
+
+def resolve_project_layout(
+    *,
+    original_filename: str,
+    save_as_project: bool,
+    base_dir: Path,
+    timestamp_fn: Callable[[], str] = output_timestamp_token,
+    sanitize_fn: Callable[[str, int], str] = sanitize_output_stem,
+) -> Tuple[Path, Path, Path, Optional[Path]]:
+    sanitized = sanitize_fn(original_filename, 0)
+    folder_name = f"{timestamp_fn()}_{sanitized}"
+    project_root = base_dir / folder_name
+    project_root.mkdir(parents=True, exist_ok=True)
+
+    if save_as_project:
+        audio_dir = project_root / "audio"
+        subtitle_dir = project_root / "subtitles"
+        metadata_dir = project_root / "metadata"
+        for directory in (audio_dir, subtitle_dir, metadata_dir):
+            directory.mkdir(parents=True, exist_ok=True)
+        return project_root, audio_dir, subtitle_dir, metadata_dir
+
+    return project_root, project_root, project_root, None

abogen/domain/progress.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+"""Progress and ETR (estimated time remaining) calculation.
+
+Shared by Web UI and PyQt desktop GUI. Pure math, no UI dependencies.
+"""
+import time
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ProgressTracker:
+    """Tracks character-based progress with ETR calculation.
+
+    Usage:
+        tracker = ProgressTracker(total_chars=50000)
+        # ... as processing occurs:
+        tracker.update(chars_done=5000)
+        print(tracker.etr_str)   # "00:04:30"
+        print(tracker.percent)   # 10
+    """
+    total_chars: int
+    _start_time: float = field(default_factory=time.time, repr=False)
+    _chars_done: int = field(default=0, repr=False)
+
+    def update(self, chars_done: int) -> None:
+        self._chars_done = chars_done
+
+    @property
+    def percent(self) -> int:
+        if self.total_chars <= 0:
+            return 0
+        return min(int(self._chars_done / self.total_chars * 100), 99)
+
+    @property
+    def etr_str(self) -> str:
+        elapsed = time.time() - self._start_time
+        if self._chars_done <= 0 or elapsed <= 0.5:
+            return "Processing..."
+        avg_time_per_char = elapsed / self._chars_done
+        remaining = self.total_chars - self._chars_done
+        if remaining <= 0:
+            return "00:00:00"
+        secs = avg_time_per_char * remaining
+        h = int(secs // 3600)
+        m = int((secs % 3600) // 60)
+        s = int(secs % 60)
+        return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+def calc_etr_str(elapsed: float, done: int, total: int) -> str:
+    """Standalone ETR string calculation (matches PyQt original logic).
+
+    Args:
+        elapsed: seconds since processing started
+        done: items/characters processed so far
+        total: total items/characters to process
+
+    Returns:
+        ETR string like "01:23:45" or "Processing..."
+    """
+    if done <= 0 or elapsed <= 0.5:
+        return "Processing..."
+    avg_time_per_item = elapsed / done
+    remaining = total - done
+    if remaining <= 0:
+        return "00:00:00"
+    secs = avg_time_per_item * remaining
+    h = int(secs // 3600)
+    m = int((secs % 3600) // 60)
+    s = int(secs % 60)
+    return f"{h:02d}:{m:02d}:{s:02d}"

abogen/domain/pronunciation.py

@@ -0,0 +1,261 @@
+"""Pronunciation rule compilation and application.
+
+Pure functions for compiling token-level and sentence-level pronunciation
+overrides into regex patterns, applying them to text, and merging multiple
+override sources with precedence rules.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Iterable, List, Mapping, Optional
+
+from abogen.entity_analysis import normalize_token as normalize_entity_token
+from abogen.entity_analysis import normalize_manual_override_token
+
+
+def compile_pronunciation_rules(
+    overrides: Optional[Iterable[Mapping[str, Any]]],
+) -> List[Dict[str, Any]]:
+    if not overrides:
+        return []
+
+    candidates: List[Dict[str, Any]] = []
+    seen: set[str] = set()
+
+    for entry in overrides:
+        if not isinstance(entry, Mapping):
+            continue
+        pronunciation_value = str(entry.get("pronunciation") or "").strip()
+        if not pronunciation_value:
+            continue
+
+        token_values: List[str] = []
+        token_raw = entry.get("token")
+        if token_raw:
+            token_value = str(token_raw).strip()
+            if token_value:
+                token_values.append(token_value)
+        normalized_raw = entry.get("normalized")
+        if normalized_raw:
+            normalized_value = str(normalized_raw).strip()
+            if normalized_value:
+                token_values.append(normalized_value)
+        if token_raw and not token_values:
+            fallback = normalize_entity_token(str(token_raw))
+            if fallback:
+                token_values.append(fallback)
+
+        if not token_values:
+            continue
+
+        usage_normalized = str(entry.get("normalized") or "").strip()
+        if not usage_normalized and token_values:
+            usage_normalized = normalize_entity_token(token_values[0]) or token_values[0]
+        usage_token = str(entry.get("token") or token_values[0])
+
+        for token_value in token_values:
+            key = token_value.casefold()
+            if key in seen:
+                continue
+            seen.add(key)
+            candidates.append(
+                {
+                    "token": token_value,
+                    "normalized": usage_normalized,
+                    "replacement": pronunciation_value,
+                }
+            )
+
+    if not candidates:
+        return []
+
+    candidates.sort(key=lambda item: len(item["token"]), reverse=True)
+    compiled: List[Dict[str, Any]] = []
+    for candidate in candidates:
+        token_value = candidate["token"]
+        pronunciation_value = candidate["replacement"]
+        escaped = re.escape(token_value)
+        pattern = re.compile(rf"(?i)(?<!\w){escaped}(?P<possessive>'s|\u2019s|\u2019)?(?!\w)")
+        compiled.append(
+            {
+                "pattern": pattern,
+                "replacement": pronunciation_value,
+                "normalized": candidate.get("normalized") or token_value,
+                "token": candidate.get("token") or token_value,
+            }
+        )
+
+    return compiled
+
+
+def compile_heteronym_sentence_rules(
+    overrides: Optional[Iterable[Mapping[str, Any]]],
+) -> List[Dict[str, Any]]:
+    if not overrides:
+        return []
+
+    compiled: List[Dict[str, Any]] = []
+    seen: set[str] = set()
+
+    for entry in overrides:
+        if not isinstance(entry, Mapping):
+            continue
+        sentence = str(entry.get("sentence") or "").strip()
+        if not sentence:
+            continue
+        choice = str(entry.get("choice") or "").strip()
+        if not choice:
+            continue
+
+        replacement_sentence = ""
+        options = entry.get("options")
+        if isinstance(options, list):
+            for opt in options:
+                if not isinstance(opt, Mapping):
+                    continue
+                if str(opt.get("key") or "").strip() == choice:
+                    replacement_sentence = str(opt.get("replacement_sentence") or "").strip()
+                    break
+        if not replacement_sentence:
+            continue
+
+        rule_key = f"{sentence}\n{choice}".casefold()
+        if rule_key in seen:
+            continue
+        seen.add(rule_key)
+
+        parts = [p for p in re.split(r"\s+", sentence) if p]
+        if not parts:
+            continue
+        pattern_text = r"\s+".join(re.escape(p) for p in parts)
+        pattern = re.compile(pattern_text)
+        compiled.append({"pattern": pattern, "replacement": replacement_sentence})
+
+    compiled.sort(key=lambda item: len(item["pattern"].pattern), reverse=True)
+    return compiled
+
+
+def apply_heteronym_sentence_rules(text: str, rules: List[Dict[str, Any]]) -> str:
+    if not text or not rules:
+        return text
+    result = text
+    for rule in rules:
+        pattern = rule["pattern"]
+        replacement = rule["replacement"]
+        result = pattern.sub(replacement, result)
+    return result
+
+
+def apply_pronunciation_rules(
+    text: str,
+    rules: List[Dict[str, Any]],
+    usage_counter: Optional[Dict[str, int]] = None,
+) -> str:
+    if not text or not rules:
+        return text
+
+    result = text
+    for rule in rules:
+        pattern = rule["pattern"]
+        pronunciation_value = rule["replacement"]
+        usage_key = str(rule.get("normalized") or "").strip()
+
+        def _replacement(match: re.Match[str]) -> str:
+            suffix = match.group("possessive") or ""
+            if usage_counter is not None and usage_key:
+                usage_counter[usage_key] = usage_counter.get(usage_key, 0) + 1
+            return pronunciation_value + suffix
+
+        result = pattern.sub(_replacement, result)
+
+    return result
+
+
+def merge_pronunciation_overrides(job: Any) -> List[Dict[str, Any]]:
+    """Return pronunciation override entries, ensuring manual overrides are included.
+
+    Pending jobs keep both ``manual_overrides`` and ``pronunciation_overrides``, but the
+    latter can be stale if the UI didn't resync before enqueue. During conversion,
+    we must merge manual overrides so they always apply (before TTS).
+
+    Precedence: manual overrides win over existing entries for the same normalized key.
+    """
+
+    collected: Dict[str, Dict[str, Any]] = {}
+
+    existing = getattr(job, "pronunciation_overrides", None)
+    if isinstance(existing, list):
+        for entry in existing:
+            if not isinstance(entry, Mapping):
+                continue
+            token_value = str(entry.get("token") or "").strip()
+            pronunciation_value = str(entry.get("pronunciation") or "").strip()
+            if not token_value or not pronunciation_value:
+                continue
+            normalized = str(entry.get("normalized") or "").strip() or normalize_entity_token(token_value)
+            if not normalized:
+                continue
+            collected[normalized] = {
+                "token": token_value,
+                "normalized": normalized,
+                "pronunciation": pronunciation_value,
+                "voice": str(entry.get("voice") or "").strip() or None,
+                "notes": str(entry.get("notes") or "").strip() or None,
+                "context": str(entry.get("context") or "").strip() or None,
+                "source": str(entry.get("source") or "pronunciation"),
+                "language": getattr(job, "language", None),
+            }
+
+    speakers = getattr(job, "speakers", None)
+    if isinstance(speakers, dict):
+        for payload in speakers.values():
+            if not isinstance(payload, Mapping):
+                continue
+            token_value = str(payload.get("token") or "").strip()
+            pronunciation_value = str(payload.get("pronunciation") or "").strip()
+            if not token_value or not pronunciation_value:
+                continue
+            normalized = normalize_entity_token(token_value)
+            if not normalized:
+                continue
+            collected[normalized] = {
+                "token": token_value,
+                "normalized": normalized,
+                "pronunciation": pronunciation_value,
+                "voice": str(
+                    payload.get("resolved_voice")
+                    or payload.get("voice")
+                    or getattr(job, "voice", "")
+                ).strip()
+                or None,
+                "notes": None,
+                "context": None,
+                "source": "speaker",
+                "language": getattr(job, "language", None),
+            }
+
+    manual = getattr(job, "manual_overrides", None)
+    if isinstance(manual, list):
+        for entry in manual:
+            if not isinstance(entry, Mapping):
+                continue
+            token_value = str(entry.get("token") or "").strip()
+            pronunciation_value = str(entry.get("pronunciation") or "").strip()
+            if not token_value or not pronunciation_value:
+                continue
+            normalized = str(entry.get("normalized") or "").strip() or normalize_manual_override_token(token_value)
+            if not normalized:
+                continue
+            collected[normalized] = {
+                "token": token_value,
+                "normalized": normalized,
+                "pronunciation": pronunciation_value,
+                "voice": str(entry.get("voice") or "").strip() or None,
+                "notes": str(entry.get("notes") or "").strip() or None,
+                "context": str(entry.get("context") or "").strip() or None,
+                "source": str(entry.get("source") or "manual"),
+                "language": getattr(job, "language", None),
+            }
+
+    return list(collected.values())

abogen/domain/split_pattern.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+"""Unified split pattern logic extracted from 3 copies."""
+import re
+
+
+PUNCTUATION_SENTENCE = r".!?。！？"
+PUNCTUATION_SENTENCE_COMMA = r".!?,。！？、，"
+
+
+def get_split_pattern(language: str, subtitle_mode: str) -> str:
+    """Get the appropriate split pattern based on language and subtitle mode.
+
+    Args:
+        language: Language code (a, b, e, f, etc.)
+        subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.)
+
+    Returns:
+        Split pattern string
+    """
+    # For English, always use newline splitting only
+    if language in ("a", "b"):
+        return "\n"
+
+    # Determine spacing pattern based on language
+    spacing = r"\s*" if language in ("z", "j") else r"\s+"
+
+    # For CJK languages, when subtitle mode is Disabled or Line, prefer
+    # punctuation-based splitting instead of plain newline splitting.
+    if subtitle_mode in ("Disabled", "Line") and language in ("z", "j"):
+        return rf"(?<=[{PUNCTUATION_SENTENCE}]){spacing}|\n+"
+
+    if subtitle_mode == "Line":
+        return "\n"
+    elif subtitle_mode == "Sentence":
+        return rf"(?<=[{PUNCTUATION_SENTENCE}]){spacing}|\n+"
+    elif subtitle_mode == "Sentence + Comma":
+        return rf"(?<=[{PUNCTUATION_SENTENCE_COMMA}]){spacing}|\n+"
+    else:
+        return r"\n+"

abogen/domain/subtitle_generation.py

@@ -0,0 +1,358 @@
+"""Subtitle generation utilities for audiobook generation.
+
+This module provides functions for processing TTS tokens into subtitle entries
+according to various subtitle modes (Line, Sentence, Sentence + Comma,
+Sentence + Highlighting).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import List, Optional, Tuple
+
+
+# Punctuation constants for sentence splitting
+PUNCTUATION_SENTENCE = ".!?\u061f\u3002\uff01\uff1f"  # .!? .?. ??
+PUNCTUATION_SENTENCE_COMMA = ".!?,\u3001\u061f\u3002\uff01\uff0c\uff1f"  # .!?, ,. ??
+
+
+def process_subtitle_tokens(
+    tokens_with_timestamps: List[dict],
+    subtitle_entries: List[Tuple[float, float, str]],
+    max_subtitle_words: int,
+    subtitle_mode: str,
+    lang_code: str,
+    use_spacy_segmentation: bool = False,
+    fallback_end_time: Optional[float] = None,
+) -> None:
+    """Process TTS tokens into subtitle entries according to the subtitle mode.
+    
+    This function modifies subtitle_entries in-place by appending new entries.
+    
+    Args:
+        tokens_with_timestamps: List of token dictionaries with 'start', 'end', 'text',
+            and 'whitespace' keys.
+        subtitle_entries: List to append subtitle entries to (modified in-place).
+            Each entry is a tuple of (start_time, end_time, text).
+        max_subtitle_words: Maximum number of words per subtitle entry.
+        subtitle_mode: One of "Disabled", "Line", "Sentence", "Sentence + Comma",
+            "Sentence + Highlighting", or a string like "5" for word-count mode.
+        lang_code: Language code for spaCy processing (e.g., "a" for English).
+        use_spacy_segmentation: Whether to use spaCy for sentence boundary detection.
+        fallback_end_time: Fallback end time for the last entry if none is available.
+    """
+    if not tokens_with_timestamps:
+        return
+
+    processed_tokens = tokens_with_timestamps
+
+    # For English with spaCy enabled and sentence-based modes, use spaCy for sentence boundaries
+    # spaCy is disabled when subtitle mode is "Disabled" or "Line"
+    use_spacy_for_english = (
+        use_spacy_segmentation
+        and subtitle_mode not in ["Disabled", "Line"]
+        and lang_code in ["a", "b"]
+        and subtitle_mode in ["Sentence", "Sentence + Comma"]
+    )
+
+    if subtitle_mode == "Sentence + Highlighting":
+        _process_karaoke_highlighting(
+            processed_tokens, subtitle_entries, max_subtitle_words, fallback_end_time
+        )
+    elif subtitle_mode in ["Sentence", "Sentence + Comma", "Line"]:
+        if use_spacy_for_english and subtitle_mode != "Line":
+            _process_spacy_sentences(
+                processed_tokens, subtitle_entries, max_subtitle_words,
+                subtitle_mode, lang_code, fallback_end_time
+            )
+        else:
+            _process_regex_sentences(
+                processed_tokens, subtitle_entries, max_subtitle_words,
+                subtitle_mode, fallback_end_time
+            )
+    else:
+        # Word count-based grouping (e.g., "5" for 5-word groups)
+        _process_word_count(
+            processed_tokens, subtitle_entries, max_subtitle_words,
+            subtitle_mode, fallback_end_time
+        )
+
+
+def _process_karaoke_highlighting(
+    tokens: List[dict],
+    subtitle_entries: List[Tuple[float, float, str]],
+    max_subtitle_words: int,
+    fallback_end_time: Optional[float],
+) -> None:
+    """Process tokens for Sentence + Highlighting mode (karaoke effect)."""
+    separator = rf"[{re.escape(PUNCTUATION_SENTENCE)}]"
+    current_sentence = []
+    word_count = 0
+
+    for token in tokens:
+        current_sentence.append(token)
+        word_count += 1
+
+        # Split sentences based on separator or word count
+        if (
+            re.search(separator, token["text"]) and token.get("whitespace") == " "
+        ) or word_count >= max_subtitle_words:
+            if current_sentence:
+                # Create karaoke subtitle entry for this sentence
+                start_time = current_sentence[0]["start"]
+                end_time = current_sentence[-1]["end"]
+
+                # Generate karaoke text with timing
+                karaoke_text = ""
+                for t in current_sentence:
+                    # Calculate duration in centiseconds
+                    duration = (
+                        t["end"] - t["start"]
+                        if t.get("end") is not None and t.get("start") is not None
+                        else 0.5
+                    )
+                    duration_cs = int(duration * 100)
+                    # Add karaoke effect
+                    karaoke_text += f"{{\\kf{duration_cs}}}{t['text']}{t.get('whitespace', '') or ''}"
+
+                subtitle_entries.append(
+                    (start_time, end_time, karaoke_text.strip())
+                )
+                current_sentence = []
+                word_count = 0
+
+    # Add any remaining tokens as a sentence
+    if current_sentence:
+        start_time = current_sentence[0]["start"]
+        end_time = current_sentence[-1]["end"]
+
+        # Generate karaoke text for remaining tokens
+        karaoke_text = ""
+        for t in current_sentence:
+            duration = t["end"] - t["start"] if t.get("end") and t.get("start") else 0.5
+            duration_cs = int(duration * 100)
+            karaoke_text += f"{{\\kf{duration_cs}}}{t['text']}{t.get('whitespace', '') or ''}"
+        subtitle_entries.append((start_time, end_time, karaoke_text.strip()))
+
+    # Fallback for last entry
+    _apply_fallback_end_time(subtitle_entries, fallback_end_time)
+
+
+def _process_spacy_sentences(
+    tokens: List[dict],
+    subtitle_entries: List[Tuple[float, float, str]],
+    max_subtitle_words: int,
+    subtitle_mode: str,
+    lang_code: str,
+    fallback_end_time: Optional[float],
+) -> None:
+    """Process tokens using spaCy for sentence boundary detection."""
+    try:
+        from abogen.spacy_utils import get_spacy_model
+    except ImportError:
+        # Fall back to regex if spaCy is not available
+        _process_regex_sentences(
+            tokens, subtitle_entries, max_subtitle_words,
+            subtitle_mode, fallback_end_time
+        )
+        return
+
+    nlp = get_spacy_model(lang_code)
+    if not nlp:
+        _process_regex_sentences(
+            tokens, subtitle_entries, max_subtitle_words,
+            subtitle_mode, fallback_end_time
+        )
+        return
+
+    # Build full text and track character positions to token indices
+    full_text = ""
+    for token in tokens:
+        text_part = token["text"] + (token.get("whitespace") or "")
+        full_text += text_part
+
+    # Get sentence boundaries from spaCy
+    doc = nlp(full_text)
+    sentence_boundaries = [sent.end_char for sent in doc.sents]
+
+    # For "Sentence + Comma" mode, also split on commas
+    if subtitle_mode == "Sentence + Comma":
+        comma_positions = [
+            i + 1 for i, c in enumerate(full_text) if c == ","
+        ]
+        sentence_boundaries = sorted(
+            set(sentence_boundaries + comma_positions)
+        )
+
+    # Group tokens by sentence boundaries
+    current_sentence = []
+    word_count = 0
+    current_char_pos = 0
+    boundary_idx = 0
+
+    for token in tokens:
+        current_sentence.append(token)
+        word_count += 1
+        text_len = len(token["text"]) + len(token.get("whitespace") or "")
+        current_char_pos += text_len
+
+        # Check if we've hit a sentence boundary or max words
+        at_boundary = (
+            boundary_idx < len(sentence_boundaries)
+            and current_char_pos >= sentence_boundaries[boundary_idx]
+        )
+        if at_boundary or word_count >= max_subtitle_words:
+            if current_sentence:
+                start_time = current_sentence[0]["start"]
+                end_time = current_sentence[-1]["end"]
+                sentence_text = "".join(
+                    t["text"] + (t.get("whitespace") or "")
+                    for t in current_sentence
+                )
+                subtitle_entries.append(
+                    (start_time, end_time, sentence_text.strip())
+                )
+                current_sentence = []
+                word_count = 0
+            if at_boundary:
+                boundary_idx += 1
+
+    # Add remaining tokens
+    if current_sentence:
+        start_time = current_sentence[0]["start"]
+        end_time = current_sentence[-1]["end"]
+        sentence_text = "".join(
+            t["text"] + (t.get("whitespace") or "")
+            for t in current_sentence
+        )
+        subtitle_entries.append(
+            (start_time, end_time, sentence_text.strip())
+        )
+
+    # Fallback for last entry
+    _apply_fallback_end_time(subtitle_entries, fallback_end_time)
+
+
+def _process_regex_sentences(
+    tokens: List[dict],
+    subtitle_entries: List[Tuple[float, float, str]],
+    max_subtitle_words: int,
+    subtitle_mode: str,
+    fallback_end_time: Optional[float],
+) -> None:
+    """Process tokens using regex for sentence boundary detection."""
+    # Define separator pattern based on mode
+    if subtitle_mode == "Line":
+        separator = r"\n"
+    elif subtitle_mode == "Sentence":
+        # Use punctuation without comma
+        separator = rf"[{re.escape(PUNCTUATION_SENTENCE)}]"
+    else:  # Sentence + Comma
+        # Use punctuation with comma
+        separator = rf"[{re.escape(PUNCTUATION_SENTENCE_COMMA)}]"
+
+    current_sentence = []
+    word_count = 0
+
+    for token in tokens:
+        current_sentence.append(token)
+        word_count += 1
+
+        # Split sentences based on separator or word count
+        if (
+            re.search(separator, token["text"]) and token.get("whitespace") == " "
+        ) or word_count >= max_subtitle_words:
+            if current_sentence:
+                # Create subtitle entry for this sentence
+                start_time = current_sentence[0]["start"]
+                end_time = current_sentence[-1]["end"]
+
+                # Simplified text joining logic
+                sentence_text = ""
+                for t in current_sentence:
+                    sentence_text += t["text"] + (t.get("whitespace") or "")
+
+                subtitle_entries.append(
+                    (start_time, end_time, sentence_text.strip())
+                )
+                current_sentence = []
+                word_count = 0
+
+    # Add any remaining tokens as a sentence
+    if current_sentence:
+        start_time = current_sentence[0]["start"]
+        end_time = current_sentence[-1]["end"]
+
+        # Simplified text joining logic
+        sentence_text = ""
+        for t in current_sentence:
+            sentence_text += t["text"] + (t.get("whitespace") or "")
+        subtitle_entries.append((start_time, end_time, sentence_text.strip()))
+
+    # Fallback for last entry
+    _apply_fallback_end_time(subtitle_entries, fallback_end_time)
+
+
+def _process_word_count(
+    tokens: List[dict],
+    subtitle_entries: List[Tuple[float, float, str]],
+    max_subtitle_words: int,
+    subtitle_mode: str,
+    fallback_end_time: Optional[float],
+) -> None:
+    """Process tokens by counting spaces (word count mode)."""
+    try:
+        word_count = int(subtitle_mode.split()[0])
+        word_count = min(word_count, max_subtitle_words)
+    except (ValueError, IndexError):
+        word_count = 1
+
+    current_group = []
+    space_count = 0
+
+    for token in tokens:
+        current_group.append(token)
+
+        # Count spaces after tokens (in the whitespace field)
+        if token.get("whitespace", "") == " ":
+            space_count += 1
+
+            # Split after counting N spaces
+            if space_count >= word_count:
+                text = "".join(
+                    t["text"] + (t.get("whitespace") or "")
+                    for t in current_group
+                )
+                subtitle_entries.append(
+                    (
+                        current_group[0]["start"],
+                        current_group[-1]["end"],
+                        text.strip(),
+                    )
+                )
+                current_group = []
+                space_count = 0
+
+    # Add any remaining tokens
+    if current_group:
+        text = "".join(
+            t["text"] + (t.get("whitespace") or "") for t in current_group
+        )
+        subtitle_entries.append(
+            (current_group[0]["start"], current_group[-1]["end"], text.strip())
+        )
+
+    # Fallback for last entry
+    _apply_fallback_end_time(subtitle_entries, fallback_end_time)
+
+
+def _apply_fallback_end_time(
+    subtitle_entries: List[Tuple[float, float, str]],
+    fallback_end_time: Optional[float],
+) -> None:
+    """Apply fallback end time to the last entry if needed."""
+    if subtitle_entries and fallback_end_time is not None:
+        last_entry = subtitle_entries[-1]
+        start, end, text = last_entry
+        if end is None or end <= start or end <= 0:
+            subtitle_entries[-1] = (start, fallback_end_time, text)

abogen/domain/title_builder.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Mapping, Optional
+
+from .metadata_helpers import (
+    ensure_sentence,
+    extract_series_metadata,
+    format_author_sentence,
+    format_series_sentence,
+    normalize_metadata_map,
+)
+
+
+def build_title_intro_text(
+    metadata: Optional[Mapping[str, Any]],
+    fallback_basename: str,
+) -> str:
+    """Build the title introduction text from metadata."""
+    normalized = normalize_metadata_map(metadata)
+    fallback_title = Path(fallback_basename).stem if fallback_basename else ""
+    title = (
+        normalized.get("title")
+        or normalized.get("book_title")
+        or normalized.get("album")
+        or fallback_title
+    )
+    if not title:
+        title = fallback_title
+    subtitle = normalized.get("subtitle") or normalized.get("sub_title")
+    if subtitle and title and subtitle.casefold() == title.casefold():
+        subtitle = ""
+
+    author_value = ""
+    for candidate in ("artist", "album_artist", "author", "authors", "writer", "composer"):
+        value = normalized.get(candidate)
+        if value:
+            author_value = value
+            break
+
+    series_name, series_number = extract_series_metadata(normalized)
+    series_sentence = format_series_sentence(series_name, series_number)
+
+    sentences: List[str] = []
+    if series_sentence:
+        sentences.append(ensure_sentence(series_sentence))
+    if title:
+        sentences.append(ensure_sentence(title))
+    if subtitle:
+        sentences.append(ensure_sentence(subtitle))
+    author_sentence = format_author_sentence(author_value)
+    if author_sentence:
+        sentences.append(ensure_sentence(author_sentence))
+    return " ".join(sentences).strip()
+
+
+def build_outro_text(
+    metadata: Optional[Mapping[str, Any]],
+    fallback_basename: str,
+) -> str:
+    """Build the outro/closing text from metadata."""
+    normalized = normalize_metadata_map(metadata)
+    fallback_title = Path(fallback_basename).stem if fallback_basename else ""
+    title = (
+        normalized.get("title")
+        or normalized.get("book_title")
+        or normalized.get("album")
+        or fallback_title
+    )
+    author_value = ""
+    for candidate in ("authors", "author", "album_artist", "artist", "writer", "composer"):
+        value = normalized.get(candidate)
+        if value:
+            author_value = value
+            break
+    author_sentence = format_author_sentence(author_value)
+    authors_fragment = (
+        author_sentence[3:].strip() if author_sentence.lower().startswith("by ") else author_sentence.strip()
+    )
+
+    if title and authors_fragment:
+        closing_line = f"The end of {title} from {authors_fragment}"
+    elif title:
+        closing_line = f"The end of {title}"
+    elif authors_fragment:
+        closing_line = f"The end from {authors_fragment}"
+    else:
+        closing_line = "The end"
+
+    series_name, series_number = extract_series_metadata(normalized)
+    series_sentence = format_series_sentence(series_name, series_number)
+
+    sentences: List[str] = [ensure_sentence(closing_line)]
+    if series_sentence:
+        sentences.append(ensure_sentence(series_sentence))
+
+    return " ".join(sentence for sentence in sentences if sentence).strip()

abogen/domain/tokens.py

@@ -0,0 +1,13 @@
+"""Shared token stubs for TTS processing."""
+
+from __future__ import annotations
+
+
+class FakeToken:
+    """Minimal token stub for languages without per-word token support."""
+
+    def __init__(self, text: str, start: float, end: float):
+        self.text = text
+        self.start_ts = start
+        self.end_ts = end
+        self.whitespace = ""

abogen/domain/voice_loader.py

@@ -0,0 +1,116 @@
+"""Voice loading and caching utilities.
+
+This module provides unified voice loading with caching support for both
+PyQt and WebUI interfaces.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Tuple
+
+from abogen.voice_formulas import get_new_voice
+
+
+class VoiceCache:
+    """Thread-safe voice cache for loaded voice tensors."""
+    
+    def __init__(self):
+        self._cache: Dict[str, Any] = {}
+    
+    def get(self, voice_spec: str) -> Optional[Any]:
+        """Get cached voice by spec."""
+        return self._cache.get(voice_spec)
+    
+    def set(self, voice_spec: str, voice: Any) -> None:
+        """Cache a loaded voice."""
+        self._cache[voice_spec] = voice
+    
+    def contains(self, voice_spec: str) -> bool:
+        """Check if voice is in cache."""
+        return voice_spec in self._cache
+    
+    def clear(self) -> None:
+        """Clear all cached voices."""
+        self._cache.clear()
+    
+    def __contains__(self, voice_spec: str) -> bool:
+        return self.contains(voice_spec)
+
+
+def resolve_voice(
+    voice_spec: str,
+    pipeline: Any,
+    use_gpu: bool,
+    cache: Optional[VoiceCache] = None,
+) -> Any:
+    """Resolve voice spec to actual voice tensor or name.
+    
+    If voice_spec contains '*' (formula), loads the voice using get_new_voice.
+    Otherwise, returns the voice_spec as-is (it's a voice name).
+    
+    Uses optional cache to avoid reloading same voice multiple times.
+    
+    Args:
+        voice_spec: Voice specification (name or formula string with '*').
+        pipeline: TTS pipeline instance for loading formula voices.
+        use_gpu: Whether to use GPU for voice loading.
+        cache: Optional VoiceCache instance for caching loaded voices.
+        
+    Returns:
+        Loaded voice tensor (for formulas) or voice name string.
+    """
+    # Check cache first
+    if cache and cache.contains(voice_spec):
+        return cache.get(voice_spec)
+    
+    # Load voice
+    if "*" in voice_spec:
+        if pipeline is None:
+            return voice_spec
+        loaded_voice = get_new_voice(pipeline, voice_spec, use_gpu)
+    else:
+        loaded_voice = voice_spec
+    
+    # Cache it
+    if cache:
+        cache.set(voice_spec, loaded_voice)
+    
+    return loaded_voice
+
+
+def load_voice_cached(
+    voice_name: str,
+    pipeline: Any,
+    use_gpu: bool,
+    cache: Optional[Dict[str, Any]] = None,
+) -> Any:
+    """Load voice with caching (compatibility wrapper for PyQt).
+    
+    This function maintains backward compatibility with the PyQt interface
+    while using the unified voice loading logic.
+    
+    Args:
+        voice_name: Voice name or formula string.
+        pipeline: TTS pipeline instance.
+        use_gpu: Whether to use GPU.
+        cache: Optional dict to use as cache (instead of VoiceCache).
+        
+    Returns:
+        Loaded voice tensor or voice name string.
+    """
+    # Use dict cache if provided (for backward compatibility)
+    if cache is not None:
+        if voice_name in cache:
+            return cache[voice_name]
+    
+    # Load voice
+    if "*" in voice_name:
+        loaded_voice = get_new_voice(pipeline, voice_name, use_gpu)
+    else:
+        loaded_voice = voice_name
+    
+    # Cache it
+    if cache is not None:
+        cache[voice_name] = loaded_voice
+    
+    return loaded_voice

abogen/domain/voice_resolution.py

@@ -0,0 +1,190 @@
+"""Voice resolution helpers.
+
+Functions for resolving voice specifications, collecting required voice IDs,
+and determining the voice to use for chapters and chunks.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Set
+
+from abogen.tts_plugin.utils import get_voices, get_default_voice
+from abogen.voice_formulas import extract_voice_ids
+from abogen.voice_cache import ensure_voice_assets
+
+
+def spec_to_voice_ids(spec: Any) -> Set[str]:
+    text = str(spec or "").strip()
+    if not text:
+        return set()
+    if text == "__custom_mix":
+        return set()
+    if "*" in text:
+        try:
+            return set(extract_voice_ids(text))
+        except ValueError:
+            return set()
+    if text in get_voices("kokoro"):
+        return {text}
+    return set()
+
+
+def job_voice_fallback(job: Any) -> str:
+    base = str(getattr(job, "voice", "") or "").strip()
+    if base and base != "__custom_mix":
+        return base
+
+    speakers = getattr(job, "speakers", None)
+    if isinstance(speakers, dict):
+        narrator = speakers.get("narrator")
+        if isinstance(narrator, dict):
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = narrator.get(key)
+                candidate = str(value or "").strip()
+                if candidate and candidate != "__custom_mix":
+                    return candidate
+        for payload in speakers.values() or []:
+            if not isinstance(payload, dict):
+                continue
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = payload.get(key)
+                candidate = str(value or "").strip()
+                if candidate and candidate != "__custom_mix":
+                    return candidate
+
+    for chapter in getattr(job, "chapters", []) or []:
+        if not isinstance(chapter, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            candidate = str(chapter.get(key) or "").strip()
+            if candidate and candidate != "__custom_mix":
+                return candidate
+
+    return ""
+
+
+def collect_required_voice_ids(job: Any) -> Set[str]:
+    voices: Set[str] = set()
+    voices.update(spec_to_voice_ids(job.voice))
+    voices.update(spec_to_voice_ids(job_voice_fallback(job)))
+
+    for chapter in getattr(job, "chapters", []) or []:
+        if not isinstance(chapter, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            voices.update(spec_to_voice_ids(chapter.get(key)))
+
+    for chunk in getattr(job, "chunks", []) or []:
+        if not isinstance(chunk, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            voices.update(spec_to_voice_ids(chunk.get(key)))
+
+    speakers = getattr(job, "speakers", {})
+    if isinstance(speakers, dict):
+        for payload in speakers.values() or []:
+            if not isinstance(payload, dict):
+                continue
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                voices.update(spec_to_voice_ids(payload.get(key)))
+
+    voices.update(get_voices("kokoro"))
+    return voices
+
+
+def initialize_voice_cache(job: Any) -> None:
+    try:
+        targets = collect_required_voice_ids(job)
+        downloaded, errors = ensure_voice_assets(
+            targets,
+            on_progress=lambda message: job.add_log(message, level="debug"),
+        )
+    except RuntimeError as exc:
+        job.add_log(f"Voice cache unavailable: {exc}", level="warning")
+        return
+
+    if downloaded:
+        job.add_log(
+            f"Cached {len(downloaded)} voice asset{'s' if len(downloaded) != 1 else ''} locally.",
+            level="info",
+        )
+
+    for voice_id, error in errors.items():
+        job.add_log(f"Failed to cache voice '{voice_id}': {error}", level="warning")
+
+
+def chapter_voice_spec(job: Any, override: Optional[Dict[str, Any]]) -> str:
+    if not override:
+        return job_voice_fallback(job)
+
+    resolved = str(override.get("resolved_voice", "")).strip()
+    if resolved:
+        return resolved
+
+    formula = str(override.get("voice_formula", "")).strip()
+    if formula:
+        return formula
+
+    voice = str(override.get("voice", "")).strip()
+    if voice:
+        return voice
+
+    return job_voice_fallback(job)
+
+
+def chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str:
+    for key in ("resolved_voice", "voice_formula", "voice"):
+        value = chunk.get(key)
+        if value:
+            return str(value)
+
+    speaker_id = chunk.get("speaker_id")
+    speakers = getattr(job, "speakers", None)
+    if isinstance(speakers, dict) and speaker_id in speakers:
+        speaker_entry = speakers.get(speaker_id) or {}
+        if isinstance(speaker_entry, dict):
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = speaker_entry.get(key)
+                if value:
+                    return str(value)
+            profile_formula = speaker_entry.get("voice_formula")
+            if profile_formula:
+                return str(profile_formula)
+
+    profile_name = chunk.get("voice_profile")
+    if profile_name:
+        if isinstance(speakers, dict):
+            speaker_entry = speakers.get(profile_name)
+            if isinstance(speaker_entry, dict):
+                for key in ("resolved_voice", "voice_formula", "voice"):
+                    value = speaker_entry.get(key)
+                    if value:
+                        return str(value)
+
+    if fallback:
+        return fallback
+    return job_voice_fallback(job)
+
+
+def resolve_fallback_voice_spec(
+    base_spec: str,
+    job_voice: str,
+    voice_cache_keys: list[str],
+    provider: str = "kokoro",
+) -> str:
+    """Resolve the voice spec for intro/outro with a priority fallback chain.
+
+    Priority: base_spec → job_voice → first voice_cache key → default voice.
+    ``"__custom_mix"`` is treated as empty (it is not a usable voice spec).
+    """
+    spec = base_spec or job_voice
+    if spec == "__custom_mix":
+        spec = job_voice or ""
+    if not spec:
+        for key in voice_cache_keys:
+            if key and key != "__custom_mix":
+                spec = key.split(":", 1)[-1]
+                break
+    if not spec:
+        spec = get_default_voice(provider)
+    return spec

abogen/domain/voice_utils.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any, Mapping, Optional, Tuple, Set
+
+from abogen.voice_formulas import extract_voice_ids, get_new_voice
+from abogen.tts_plugin.utils import get_voices
+
+
+def infer_provider_from_spec(value: Any, fallback: str = "kokoro") -> str:
+    """Infer TTS provider from voice specification."""
+    raw = str(value or "").strip()
+    if not raw:
+        return fallback
+    if raw.upper() == raw and raw.replace("_", "").isalnum():
+        return "supertonic"
+    if raw == "__custom_mix" or "*" in raw or "+" in raw:
+        return "kokoro"
+    if raw in get_voices("kokoro"):
+        return "kokoro"
+    return fallback
+
+
+def supertonic_voice_from_spec(spec: Any, fallback: str) -> str:
+    """Normalize a voice specification for Supertonic.
+
+    This function only performs Supertonic-specific normalization (uppercase conversion
+    and fallback handling). Backend resolution is handled by the registry.
+    """
+    raw = str(spec or "").strip()
+    fallback_raw = str(fallback or "").strip()
+
+    # Normalize to uppercase for Supertonic voice IDs
+    upper = raw.upper() if raw else ""
+
+    # If empty or contains formula characters, use fallback
+    if not upper or "*" in upper or "+" in upper:
+        upper = fallback_raw.upper() if fallback_raw else ""
+
+    # If still empty, use default Supertonic voice
+    if not upper or "*" in upper or "+" in upper:
+        upper = "M1"
+
+    return upper
+
+
+def split_speaker_reference(value: Any) -> Tuple[Optional[str], str]:
+    """Parse speaker/profile reference from string.
+
+    Expected format: "speaker:name" or "profile:name"
+    Returns (name, original) or (None, original) if not a valid reference.
+    """
+    raw = str(value or "").strip()
+    if not raw or ":" not in raw:
+        return None, raw
+    prefix, remainder = raw.split(":", 1)
+    prefix = prefix.strip().lower()
+    if prefix not in {"speaker", "profile"}:
+        return None, raw
+    name = remainder.strip()
+    return (name or None), raw
+
+
+def formula_from_kokoro_entry(entry: Mapping[str, Any]) -> str:
+    """Build voice formula string from kokoro entry."""
+    voices = entry.get("voices") or []
+    if not voices:
+        return ""
+    total = 0.0
+    parts: list[tuple[str, float]] = []
+    for item in voices:
+        if not isinstance(item, (list, tuple)) or len(item) < 2:
+            continue
+        name = str(item[0] or "").strip()
+        try:
+            weight = float(item[1])
+        except (TypeError, ValueError):
+            continue
+        if name and weight > 0:
+            parts.append((name, weight))
+            total += weight
+
+    if not parts:
+        return ""
+
+    normalized = [(name, weight / total) for name, weight in parts]
+    return " + ".join(f"{name}*{weight:.6f}" for name, weight in normalized)
+
+
+def coerce_truthy(value: Any, default: bool = True) -> bool:
+    """Coerce a value to boolean with default."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.lower() not in {"false", "0", "no", "off", ""}
+    if value is None:
+        return default
+    return bool(value)

abogen/infrastructure/exporters.py

@@ -0,0 +1,448 @@
+from __future__ import annotations
+
+import json
+import logging
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Mapping, Sequence
+
+import static_ffmpeg
+
+from abogen.domain.metadata_helpers import (
+    normalize_metadata_casefold,
+    split_people_field,
+    split_simple_list,
+    first_nonempty,
+    extract_year,
+    normalize_series_sequence,
+    build_audiobookshelf_metadata as _build_abs_metadata,
+    load_audiobookshelf_chapters as _load_abs_chapters,
+    _SERIES_SEQUENCE_TAG_KEYS,
+)
+from abogen.epub3.exporter import build_epub3_package
+from abogen.integrations.audiobookshelf import (
+    AudiobookshelfClient,
+    AudiobookshelfConfig,
+    AudiobookshelfUploadError,
+)
+from abogen.utils import create_process
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ExportConfig:
+    """Configuration for export operations."""
+    ffmpeg_path: str = "ffmpeg"
+    verify_ssl: bool = True
+
+
+class ExportService:
+    """Unified service for audiobook exports (M4B, FFMETADATA, EPUB3, Audiobookshelf)."""
+    
+    def __init__(self, config: Optional[ExportConfig] = None):
+        self.config = config or ExportConfig()
+        static_ffmpeg.add_paths()
+    
+    # ----------------------------------------------------------------------
+    # FFMETADATA
+    # ----------------------------------------------------------------------
+    
+    def render_ffmetadata(
+        self,
+        metadata: Dict[str, Any],
+        chapters: List[Dict[str, Any]],
+    ) -> str:
+        """Render FFMETADATA content."""
+        lines = [";FFMETADATA1"]
+        
+        for key, value in (metadata or {}).items():
+            if value is None:
+                continue
+            key_str = str(key).strip()
+            if not key_str:
+                continue
+            lines.append(f"{key_str}={self._escape_ffmetadata_value(value)}")
+        
+        for chapter in chapters or []:
+            start = chapter.get("start")
+            end = chapter.get("end")
+            if start is None or end is None:
+                continue
+            try:
+                start_ms = max(0, int(round(float(start) * 1000)))
+                end_ms = int(round(float(end) * 1000))
+            except (TypeError, ValueError):
+                continue
+            if end_ms <= start_ms:
+                end_ms = start_ms + 1
+            lines.append("[CHAPTER]")
+            lines.append("TIMEBASE=1/1000")
+            lines.append(f"START={start_ms}")
+            lines.append(f"END={end_ms}")
+            title = chapter.get("title")
+            if title:
+                lines.append(f"title={self._escape_ffmetadata_value(title)}")
+            voice = chapter.get("voice")
+            if voice:
+                lines.append(f"voice={self._escape_ffmetadata_value(voice)}")
+        
+        return "\n".join(lines) + "\n"
+    
+    @staticmethod
+    def _escape_ffmetadata_value(value: Any) -> str:
+        escaped = str(value).replace("\\", "\\\\").replace("\n", "\\n")
+        escaped = escaped.replace("=", "\\=").replace(";", "\\;").replace("#", "\\#")
+        return escaped
+    
+    def write_ffmetadata_file(
+        self,
+        audio_path: Path,
+        metadata: Dict[str, Any],
+        chapters: List[Dict[str, Any]],
+    ) -> Optional[Path]:
+        """Write FFMETADATA file to temp location."""
+        content = self.render_ffmetadata(metadata, chapters)
+        if content.strip() == ";FFMETADATA1":
+            return None
+        
+        directory = audio_path.parent if audio_path.parent.exists() else Path(tempfile.gettempdir())
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            suffix=".ffmeta",
+            delete=False,
+            dir=str(directory),
+        ) as handle:
+            handle.write(content)
+            return Path(handle.name)
+    
+    # ----------------------------------------------------------------------
+    # M4B Export
+    # ----------------------------------------------------------------------
+    
+    def embed_m4b_metadata(
+        self,
+        audio_path: Path,
+        metadata: Dict[str, Any],
+        chapters: List[Dict[str, Any]],
+        cover_path: Optional[Path] = None,
+        cover_mime: Optional[str] = None,
+        log_callback: Optional[callable] = None,
+    ) -> None:
+        """Embed metadata and chapters into M4B file using FFmpeg + Mutagen."""
+        ffmetadata_path = self.write_ffmetadata_file(audio_path, metadata, chapters)
+        
+        metadata_args = self._metadata_to_ffmpeg_args(metadata)
+        
+        cmd = ["ffmpeg", "-y", "-i", str(audio_path)]
+        
+        if ffmetadata_path:
+            cmd.extend(["-f", "ffmetadata", "-i", str(ffmetadata_path)])
+        
+        if cover_path and cover_path.exists():
+            cmd.extend(["-i", str(cover_path)])
+            cmd.extend(["-map", "0:a"])
+            cmd.extend(["-map", "1:v:0", "-c:v:0", "mjpeg", "-disposition:v:0", "attached_pic"])
+            if cover_mime:
+                cmd.extend(["-metadata:s:v:0", f"mimetype={cover_mime}"])
+            cmd.extend(["-metadata:s:v:0", "title=Cover Art"])
+        else:
+            cmd.extend(["-map", "0:a"])
+        
+        cmd.extend(["-c:a", "copy"])
+        
+        if ffmetadata_path:
+            cmd.extend(["-map_metadata", "1", "-map_chapters", "1"])
+        else:
+            cmd.extend(["-map_metadata", "0"])
+        
+        if metadata_args:
+            cmd.extend(metadata_args)
+        
+        cmd.extend(["-movflags", "+faststart+use_metadata_tags"])
+        
+        temp_output = audio_path.with_suffix(audio_path.suffix + ".tmp")
+        if audio_path.suffix.lower() in {".m4b", ".mp4", ".m4a"}:
+            cmd.extend(["-f", "mp4"])
+        cmd.append(str(temp_output))
+        
+        if log_callback:
+            log_callback("Embedding metadata into M4B output")
+        
+        process = create_process(cmd, text=True)
+        return_code = process.wait()
+        
+        if ffmetadata_path and ffmetadata_path.exists():
+            try:
+                ffmetadata_path.unlink()
+            except OSError:
+                pass
+        
+        if return_code != 0:
+            if temp_output.exists():
+                temp_output.unlink(missing_ok=True)
+            raise RuntimeError(f"ffmpeg failed to embed metadata (exit code {return_code})")
+        
+        temp_output.replace(audio_path)
+        
+        if log_callback:
+            log_callback("Embedded metadata and chapters into M4B output", "info")
+        
+        # Apply chapters via Mutagen for better compatibility
+        self._apply_m4b_chapters_mutagen(audio_path, chapters, log_callback)
+    
+    @staticmethod
+    def _metadata_to_ffmpeg_args(metadata: Dict[str, Any]) -> List[str]:
+        args = []
+        for key, value in (metadata or {}).items():
+            if value in (None, ""):
+                continue
+            key_str = str(key).strip()
+            if not key_str:
+                continue
+            normalized_key = key_str.lower()
+            if normalized_key == "year":
+                ffmpeg_key = "date"
+            else:
+                ffmpeg_key = key_str
+            args.extend(["-metadata", f"{ffmpeg_key}={value}"])
+        return args
+    
+    def _apply_m4b_chapters_mutagen(
+        self,
+        audio_path: Path,
+        chapters: List[Dict[str, Any]],
+        log_callback: Optional[callable] = None,
+    ) -> bool:
+        """Apply chapter atoms using Mutagen."""
+        if not chapters:
+            return False
+        
+        try:
+            from fractions import Fraction
+            from mutagen.mp4 import MP4, MP4Chapter
+        except ImportError:
+            if log_callback:
+                log_callback("Unable to write MP4 chapter atoms because mutagen is not installed.", "warning")
+            return False
+        
+        try:
+            mp4 = MP4(str(audio_path))
+        except Exception as exc:
+            if log_callback:
+                log_callback(f"Failed to open m4b for chapter embedding: {exc}", "warning")
+            return False
+        
+        chapter_objects = []
+        for index, entry in enumerate(sorted(chapters, key=lambda item: float(item.get("start") or 0.0))):
+            start_raw = entry.get("start")
+            if start_raw is None:
+                continue
+            try:
+                start_seconds = max(0.0, float(start_raw))
+            except (TypeError, ValueError):
+                continue
+            
+            title_value = entry.get("title")
+            title_text = str(title_value) if title_value else f"Chapter {index + 1}"
+            
+            start_fraction = Fraction(int(round(start_seconds * 1000)), 1000)
+            chapter_atom = MP4Chapter(start_fraction, title_text)
+            
+            end_raw = entry.get("end")
+            if end_raw is not None:
+                try:
+                    end_seconds = float(end_raw)
+                except (TypeError, ValueError):
+                    end_seconds = None
+                if end_seconds is not None and end_seconds > start_seconds:
+                    chapter_atom.end = Fraction(int(round(end_seconds * 1000)), 1000)
+            
+            chapter_objects.append(chapter_atom)
+        
+        if not chapter_objects:
+            return False
+        
+        try:
+            mp4.chapters = chapter_objects
+            mp4.save()
+        except Exception as exc:
+            if log_callback:
+                log_callback(f"Failed to persist MP4 chapter atoms: {exc}", "warning")
+            return False
+        
+        if log_callback:
+            log_callback(f"Applied {len(chapter_objects)} chapter markers via mutagen", "info")
+        return True
+    
+    # ----------------------------------------------------------------------
+    # EPUB3 Export
+    # ----------------------------------------------------------------------
+    
+    def export_epub3(
+        self,
+        output_path: Path,
+        book_id: str,
+        extraction: Any,  # ExtractionResult
+        metadata_tags: Dict[str, Any],
+        chapter_markers: Sequence[Dict[str, Any]],
+        chunk_markers: Sequence[Dict[str, Any]],
+        chunks: Iterable[Dict[str, Any]],
+        audio_path: Path,
+        speaker_mode: str = "single",
+        cover_path: Optional[Path] = None,
+        cover_mime: Optional[str] = None,
+    ) -> Path:
+        """Export EPUB3 with media overlays."""
+        return build_epub3_package(
+            output_path=output_path,
+            book_id=book_id,
+            extraction=extraction,
+            metadata_tags=metadata_tags,
+            chapter_markers=chapter_markers,
+            chunk_markers=chunk_markers,
+            chunks=chunks,
+            audio_path=audio_path,
+            speaker_mode=speaker_mode,
+            cover_image_path=cover_path,
+            cover_image_mime=cover_mime,
+        )
+    
+    # ----------------------------------------------------------------------
+    # Audiobookshelf Integration
+    # ----------------------------------------------------------------------
+    
+    def build_audiobookshelf_metadata(self, job: Any) -> Dict[str, Any]:
+        """Build Audiobookshelf metadata from job."""
+        filename = Path(getattr(job, "original_filename", "") or "").stem or "Audiobook"
+        return _build_abs_metadata(
+            getattr(job, "metadata_tags", {}),
+            language=getattr(job, "language", "") or "",
+            filename=filename,
+        )
+
+    def load_audiobookshelf_chapters(self, job: Any) -> Optional[List[Dict[str, Any]]]:
+        """Load chapters from job artifacts for Audiobookshelf."""
+        metadata_ref = job.result.artifacts.get("metadata") if getattr(job, "result", None) else None
+        if not metadata_ref:
+            return None
+        metadata_path = metadata_ref if isinstance(metadata_ref, Path) else Path(str(metadata_ref))
+        return _load_abs_chapters(metadata_path)
+    
+    def upload_audiobookshelf(
+        self,
+        job: Any,
+        audio_path: Path,
+        subtitle_paths: List[Path],
+        chapters: List[Dict[str, Any]],
+        metadata: Dict[str, Any],
+        cover_path: Optional[Path] = None,
+        config: Optional[AudiobookshelfConfig] = None,
+        log_callback: Optional[callable] = None,
+    ) -> None:
+        """Upload to Audiobookshelf."""
+        if config is None:
+            # Load from job or global config
+            cfg = getattr(job, "_abs_config", None)
+            if cfg is None:
+                from abogen.utils import load_config
+                global_cfg = load_config() or {}
+                abs_cfg = global_cfg.get("audiobookshelf")
+                if isinstance(abs_cfg, Mapping):
+                    config = AudiobookshelfConfig(
+                        base_url=str(abs_cfg.get("base_url") or "").strip(),
+                        api_token=str(abs_cfg.get("api_token") or "").strip(),
+                        library_id=str(abs_cfg.get("library_id") or "").strip(),
+                        collection_id=(str(abs_cfg.get("collection_id") or "").strip() or None),
+                        folder_id=str(abs_cfg.get("folder_id") or "").strip(),
+                        verify_ssl=self._coerce_bool(abs_cfg.get("verify_ssl"), True),
+                        send_cover=self._coerce_bool(abs_cfg.get("send_cover"), True),
+                        send_chapters=self._coerce_bool(abs_cfg.get("send_chapters"), True),
+                        send_subtitles=self._coerce_bool(abs_cfg.get("send_subtitles"), False),
+                        timeout=float(abs_cfg.get("timeout", 3600.0)),
+                    )
+                else:
+                    if log_callback:
+                        log_callback("Audiobookshelf upload skipped: not configured", "warning")
+                    return
+        
+        if not config.base_url or not config.api_token or not config.library_id:
+            if log_callback:
+                log_callback("Audiobookshelf upload skipped: configure base URL, API token, and library ID first", "warning")
+            return
+        if not config.folder_id:
+            if log_callback:
+                log_callback("Audiobookshelf upload skipped: enter folder name or ID in settings", "warning")
+            return
+        
+        if not audio_path.exists():
+            if log_callback:
+                log_callback("Audiobookshelf upload skipped: audio output not found", "warning")
+            return
+        
+        existing_subtitles = [p for p in subtitle_paths if p.exists()] if config.send_subtitles else None
+        chapters_to_send = chapters if config.send_chapters else None
+        
+        client = AudiobookshelfClient(config)
+        
+        display_title = metadata.get("title") or audio_path.stem
+        try:
+            existing_items = client.find_existing_items(display_title, folder_id=config.folder_id)
+        except AudiobookshelfUploadError as exc:
+            if log_callback:
+                log_callback(f"Audiobookshelf lookup failed: {exc}", "error")
+            return
+        
+        if existing_items:
+            if log_callback:
+                log_callback(f"Removing existing Audiobookshelf item(s) for '{display_title}' before upload.", "info")
+            try:
+                client.delete_items(existing_items)
+            except Exception as exc:
+                if log_callback:
+                    log_callback(f"Failed to remove existing item(s): {exc}", "warning")
+        
+        cover_to_send = cover_path
+        if config.send_cover and cover_to_send:
+            if isinstance(cover_to_send, str):
+                cover_to_send = Path(cover_to_send)
+            if not cover_to_send.exists():
+                cover_to_send = None
+        
+        client.upload_audiobook(
+            audio_path,
+            metadata=metadata,
+            cover_path=cover_to_send,
+            chapters=chapters_to_send,
+            subtitles=existing_subtitles,
+        )
+        
+        if log_callback:
+            log_callback("Audiobookshelf upload queued.", "info")
+    
+    # ----------------------------------------------------------------------
+    # Helpers
+    # ----------------------------------------------------------------------
+    
+    @staticmethod
+    def _coerce_bool(value: Any, default: bool = True) -> bool:
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            lowered = value.strip().lower()
+            if lowered in {"true", "1", "yes", "on"}:
+                return True
+            if lowered in {"false", "0", "no", "off"}:
+                return False
+            return default
+        if value is None:
+            return default
+        return bool(value)
+
+
+__all__ = [
+    "ExportConfig",
+    "ExportService",
+]

abogen/infrastructure/subtitle_writer.py

@@ -0,0 +1,303 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+from typing import List, Optional, TextIO
+
+from abogen.subtitle_utils import clean_subtitle_text
+
+
+class SubtitleFormat(Enum):
+    SRT = "srt"
+    ASS = "ass"
+    VTT = "vtt"
+
+
+class SubtitleMode(Enum):
+    DISABLED = "Disabled"
+    LINE = "Line"
+    SENTENCE = "Sentence"
+    SENTENCE_COMMA = "Sentence + Comma"
+    SENTENCE_HIGHLIGHT = "Sentence + Highlighting"
+
+
+class SubtitleAlignment(Enum):
+    LEFT = "left"
+    CENTER = "center"
+    NARROW = "narrow"
+    CENTER_NARROW = "center_narrow"
+
+
+@dataclass
+class SubtitleConfig:
+    """Configuration for subtitle writer."""
+    format: SubtitleFormat
+    mode: SubtitleMode
+    alignment: SubtitleAlignment = SubtitleAlignment.LEFT
+    max_words: int = 50
+    highlight_color: str = "&H00FFFF00"  # ASS highlight color
+
+
+class SubtitleWriter(ABC):
+    """Abstract base class for subtitle writers."""
+    
+    def __init__(self, path: Path, config: SubtitleConfig):
+        self.path = path
+        self.config = config
+        self._file: Optional[TextIO] = None
+        self._index = 0
+        self._opened = False
+    
+    def open(self) -> None:
+        """Open the subtitle file and write header."""
+        if self._opened:
+            return
+        self._file = open(self.path, "w", encoding="utf-8", errors="replace")
+        self._write_header()
+        self._opened = True
+    
+    @abstractmethod
+    def _write_header(self) -> None:
+        pass
+    
+    def write_entry(
+        self,
+        start: float,
+        end: float,
+        text: str,
+        voice: Optional[str] = None,
+    ) -> None:
+        """Write a subtitle entry."""
+        if not self._opened:
+            self.open()
+        
+        text = clean_subtitle_text(text)
+        if not text:
+            return
+        
+        self._index += 1
+        self._write_entry(self._index, start, end, text, voice)
+    
+    @abstractmethod
+    def _write_entry(
+        self,
+        index: int,
+        start: float,
+        end: float,
+        text: str,
+        voice: Optional[str],
+    ) -> None:
+        pass
+    
+    def close(self) -> None:
+        """Close the subtitle file."""
+        if self._file:
+            self._file.close()
+            self._file = None
+            self._opened = False
+    
+    def __enter__(self) -> "SubtitleWriter":
+        self.open()
+        return self
+    
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.close()
+
+
+class SrtWriter(SubtitleWriter):
+    """SRT subtitle writer."""
+    
+    def _write_header(self) -> None:
+        pass  # SRT has no header
+    
+    def _write_entry(
+        self,
+        index: int,
+        start: float,
+        end: float,
+        text: str,
+        voice: Optional[str],
+    ) -> None:
+        start_str = self._format_time(start)
+        end_str = self._format_time(end)
+        
+        if voice:
+            text = f"[{voice}] {text}"
+        
+        self._file.write(f"{index}\n")
+        self._file.write(f"{start_str} --> {end_str}\n")
+        self._file.write(f"{text}\n\n")
+    
+    @staticmethod
+    def _format_time(seconds: float) -> str:
+        hours = int(seconds // 3600)
+        minutes = int((seconds % 3600) // 60)
+        secs = int(seconds % 60)
+        millis = int((seconds - int(seconds)) * 1000)
+        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
+
+
+class VttWriter(SubtitleWriter):
+    """WebVTT subtitle writer."""
+    
+    def _write_header(self) -> None:
+        self._file.write("WEBVTT\n\n")
+    
+    def _write_entry(
+        self,
+        index: int,
+        start: float,
+        end: float,
+        text: str,
+        voice: Optional[str],
+    ) -> None:
+        start_str = self._format_time(start)
+        end_str = self._format_time(end)
+        
+        if voice:
+            text = f"[{voice}] {text}"
+        
+        self._file.write(f"{index}\n")
+        self._file.write(f"{start_str} --> {end_str}\n")
+        self._file.write(f"{text}\n\n")
+    
+    @staticmethod
+    def _format_time(seconds: float) -> str:
+        hours = int(seconds // 3600)
+        minutes = int((seconds % 3600) // 60)
+        secs = seconds % 60
+        return f"{hours:02d}:{minutes:02d}:{secs:06.3f}".replace(".", ".")
+
+
+class AssWriter(SubtitleWriter):
+    """ASS subtitle writer with karaoke highlighting support."""
+    
+    def __init__(self, path: Path, config: SubtitleConfig):
+        super().__init__(path, config)
+        self._is_centered = config.alignment in (SubtitleAlignment.CENTER, SubtitleAlignment.CENTER_NARROW)
+        self._is_narrow = config.alignment in (SubtitleAlignment.NARROW, SubtitleAlignment.CENTER_NARROW)
+    
+    def _write_header(self) -> None:
+        margin = "90" if self._is_narrow else "10"
+        alignment = "5" if self._is_centered else "2"
+        
+        self._file.write("[Script Info]\n")
+        self._file.write("Title: Generated by Abogen\n")
+        self._file.write("ScriptType: v4.00+\n\n")
+        
+        # Styles
+        self._file.write("[V4+ Styles]\n")
+        self._file.write(
+            "Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, "
+            "OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, "
+            "ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, "
+            "Alignment, MarginL, MarginR, MarginV, Encoding\n"
+        )
+        
+        if self.config.mode == SubtitleMode.SENTENCE_HIGHLIGHT:
+            # Karaoke style with highlighting
+            self._file.write(
+                f"Style: Default,Arial,24,&H00FFFFFF,&H00808080,&H00000000,&H00404040,"
+                f"0,0,0,0,100,100,0,0,3,2,0,{alignment},{margin},{margin},10,1\n"
+            )
+            self._file.write(
+                f"Style: Highlight,Arial,24,&H0000FFFF,&H00808080,&H00000000,&H00404040,"
+                f"0,0,0,0,100,100,0,0,3,2,0,{alignment},{margin},{margin},10,1\n\n"
+            )
+        else:
+            self._file.write(
+                f"Style: Default,Arial,24,&H00FFFFFF,&H00808080,&H00000000,&H00404040,"
+                f"0,0,0,0,100,100,0,0,3,2,0,{alignment},{margin},{margin},10,1\n\n"
+            )
+        
+        self._file.write("[Events]\n")
+        self._file.write(
+            "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"
+        )
+    
+    def _write_entry(
+        self,
+        index: int,
+        start: float,
+        end: float,
+        text: str,
+        voice: Optional[str],
+    ) -> None:
+        start_str = self._format_time(start)
+        end_str = self._format_time(end)
+        
+        if voice:
+            text = f"[{voice}] {text}"
+        
+        style = "Default"
+        if self.config.mode == SubtitleMode.SENTENCE_HIGHLIGHT:
+            # Add karaoke tags for highlighting
+            text = self._add_karaoke_tags(text)
+            style = "Highlight"
+        
+        alignment_tag = r"{\an5}" if self._is_centered else ""
+        self._file.write(
+            f"Dialogue: 0,{start_str},{end_str},{style},,0,0,0,,{alignment_tag}{text}\n"
+        )
+    
+    def _add_karaoke_tags(self, text: str) -> str:
+        """Add karaoke highlighting tags to text."""
+        # Simple word-level karaoke timing
+        words = text.split()
+        if not words:
+            return text
+        
+        # This is a simplified version - real karaoke needs per-word timing
+        # For now, just return the text with the highlight color
+        return r"{\k100}" + r"{\k100}".join(words) + r"{\k0}"
+    
+    @staticmethod
+    def _format_time(seconds: float) -> str:
+        hours = int(seconds // 3600)
+        minutes = int((seconds % 3600) // 60)
+        secs = seconds % 60
+        return f"{hours}:{minutes:02d}:{secs:05.2f}"
+
+
+def create_subtitle_writer(
+    path: Path,
+    format: str,
+    mode: str,
+    alignment: str = "left",
+    max_words: int = 50,
+) -> SubtitleWriter:
+    """Factory function to create subtitle writer."""
+    fmt = SubtitleFormat(format.lower())
+    mode = SubtitleMode(mode)
+    align = SubtitleAlignment(alignment.lower())
+    
+    config = SubtitleConfig(
+        format=fmt,
+        mode=mode,
+        alignment=align,
+        max_words=max_words,
+    )
+    
+    if fmt == SubtitleFormat.SRT:
+        return SrtWriter(path, config)
+    elif fmt == SubtitleFormat.VTT:
+        return VttWriter(path, config)
+    elif fmt == SubtitleFormat.ASS:
+        return AssWriter(path, config)
+    else:
+        raise ValueError(f"Unsupported subtitle format: {format}")
+
+
+__all__ = [
+    "SubtitleFormat",
+    "SubtitleMode",
+    "SubtitleAlignment",
+    "SubtitleConfig",
+    "SubtitleWriter",
+    "SrtWriter",
+    "VttWriter",
+    "AssWriter",
+    "create_subtitle_writer",
+]

abogen/integrations/audiobookshelf.py

@@ -2,9 +2,7 @@ from __future__ import annotations
 
 import json
 import logging
-import math
 import mimetypes
-import re
 from contextlib import ExitStack
 from dataclasses import dataclass
 from pathlib import Path
@@ -12,6 +10,8 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
 
 import httpx
 
+from abogen.domain.metadata_helpers import normalize_series_sequence
+
 logger = logging.getLogger(__name__)
 
 
@@ -641,40 +641,7 @@ class AudiobookshelfClient:
         for key in preferred_keys:
             if key not in metadata:
                 continue
-            normalized = AudiobookshelfClient._normalize_series_sequence(metadata.get(key))
+            normalized = normalize_series_sequence(metadata.get(key))
             if normalized:
                 return normalized
         return ""
-
-    @staticmethod
-    def _normalize_series_sequence(raw: Any) -> str:
-        if raw is None:
-            return ""
-
-        if isinstance(raw, (int, float)):
-            if isinstance(raw, float) and (math.isnan(raw) or math.isinf(raw)):
-                return ""
-            text = str(raw)
-        else:
-            text = str(raw).strip()
-
-        if not text:
-            return ""
-
-        candidate = text.replace(",", ".")
-        match = re.search(r"\d+(?:\.\d+)?", candidate)
-        if not match:
-            return ""
-
-        normalized = match.group(0)
-        if "." in normalized:
-            normalized = normalized.rstrip("0").rstrip(".")
-            if not normalized:
-                normalized = "0"
-            return normalized
-
-        try:
-            return str(int(normalized))
-        except ValueError:
-            cleaned = normalized.lstrip("0")
-            return cleaned or "0"

abogen/main.py

@@ -2,13 +2,14 @@
 
 from __future__ import annotations
 
-import atexit
 import os
 import platform
-import signal
-import sys
 
-from abogen.utils import load_config, prevent_sleep_end
+# Initialise global shutdown handling (atexit, signals, Qt) as early as possible.
+from abogen import shutdown  # noqa: F401
+shutdown.register_shutdown()
+
+from abogen.utils import load_config
 from abogen.webui.app import main as _run_web_ui
 
 # Configure Hugging Face Hub behaviour (mirrors legacy GUI defaults).
@@ -27,17 +28,6 @@ os.environ.setdefault("MIOPEN_CONV_PRECISE_ROCM_TUNING", "0")
 if platform.system() == "Darwin" and platform.processor() == "arm":
     os.environ.setdefault("PYTORCH_ENABLE_MPS_FALLBACK", "1")
 
-atexit.register(prevent_sleep_end)
-
-
-def _cleanup_sleep(signum, _frame):
-    prevent_sleep_end()
-    sys.exit(0)
-
-
-signal.signal(signal.SIGINT, _cleanup_sleep)
-signal.signal(signal.SIGTERM, _cleanup_sleep)
-
 
 def main() -> None:
     """Launch the Flask-based web UI."""

abogen/predownload_gui.py

@@ -21,7 +21,8 @@ from PyQt6.QtWidgets import (
 )
 from PyQt6.QtCore import QThread, pyqtSignal
 
-from abogen.constants import COLORS, VOICES_INTERNAL
+from abogen.constants import COLORS
+from abogen.tts_plugin.utils import get_voices
 from abogen.spacy_utils import SPACY_MODELS
 import abogen.hf_tracker
 
@@ -114,7 +115,7 @@ class PreDownloadWorker(QThread):
             self._voices_success = False
             return
 
-        voice_list = VOICES_INTERNAL
+        voice_list = get_voices("kokoro")
         for idx, voice in enumerate(voice_list, start=1):
             if self._cancelled:
                 self._voices_success = False
@@ -462,14 +463,14 @@ class PreDownloadDialog(QDialog):
         try:
             from huggingface_hub import try_to_load_from_cache
 
-            for voice in VOICES_INTERNAL:
+            for voice in get_voices("kokoro"):
                 if not try_to_load_from_cache(
                     repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
                 ):
                     missing.append(voice)
         except Exception:
             # If HF missing, report all as missing
-            return False, list(VOICES_INTERNAL)
+            return False, list(get_voices("kokoro"))
         return (len(missing) == 0), missing
 
     def _check_kokoro_model(self) -> bool:

abogen/pyqt/gui.py

@@ -7,9 +7,9 @@ import base64
 import re
 from abogen.pyqt.queue_manager_gui import QueueManager
 from abogen.pyqt.queued_item import QueuedItem
+from abogen.domain.device import select_device as _select_device
 import abogen.hf_tracker as hf_tracker
 import hashlib  # Added for cache path generation
-from abogen.tts_supertonic import SUPERTONIC_AVAILABLE_LANGS, DEFAULT_SUPERTONIC_VOICES
 from PyQt6.QtWidgets import (
     QApplication,
     QWidget,
@@ -83,13 +83,11 @@ from abogen.constants import (
     GITHUB_URL,
     PROGRAM_DESCRIPTION,
     LANGUAGE_DESCRIPTIONS,
-    VOICES_INTERNAL,
-    KOKORO_LANG_TO_COUNTRY,
-    SUPERTONIC_LANG_TO_COUNTRY,
     SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
     COLORS,
     SUBTITLE_FORMATS,
 )
+from abogen.tts_plugin.utils import get_voices
 import threading
 from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog
 from abogen.voice_profiles import load_profiles
@@ -973,9 +971,6 @@ class abogen(QWidget):
         self.fix_nonstandard_punctuation = self.config.get(
             "fix_nonstandard_punctuation", False
         )
-        self.tts_provider_config = self.config.get("tts_provider", "kokoro")
-        self.supertonic_language_config = self.config.get("supertonic_language", "en")
-        self.supertonic_total_steps_config = self.config.get("supertonic_total_steps", 8)
         self._pending_close_event = None
         self.gpu_ok = False  # Initialize GPU availability status
 
@@ -1024,16 +1019,6 @@ class abogen(QWidget):
                 else:
                     self.mixed_voice_state = entry
                     self.selected_lang = entry[0][0] if entry and entry[0] else None
-        # Restore TTS provider and supertonic settings from config
-        provider_text = "Supertonic" if self.tts_provider_config == "supertonic" else "Kokoro"
-        idx_st = self.provider_combo.findText(provider_text)
-        if idx_st >= 0:
-            self.provider_combo.setCurrentIndex(idx_st)
-        self.st_lang_combo.setCurrentText(self.supertonic_language_config)
-        idx_steps = self.st_steps_combo.findData(self.supertonic_total_steps_config)
-        if idx_steps >= 0:
-            self.st_steps_combo.setCurrentIndex(idx_steps)
-
         if self.save_option == "Choose output folder" and self.selected_output_folder:
             self.save_path_label.setText(self.selected_output_folder)
             self.save_path_row_widget.show()
@@ -1123,53 +1108,6 @@ class abogen(QWidget):
         speed_layout.addWidget(self.speed_label)
         controls_layout.addLayout(speed_layout)
         self.speed_slider.valueChanged.connect(self.update_speed_label)
-
-        # TTS Provider selection
-        provider_layout = QHBoxLayout()
-        provider_layout.setSpacing(7)
-        provider_label = QLabel("TTS Engine:", self)
-        provider_layout.addWidget(provider_label)
-        self.provider_combo = QComboBox(self)
-        self.provider_combo.addItem("Kokoro", "kokoro")
-        self.provider_combo.addItem("Supertonic", "supertonic")
-        self.provider_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
-        self.provider_combo.currentIndexChanged.connect(self.on_provider_changed)
-        provider_layout.addWidget(self.provider_combo)
-        controls_layout.addLayout(provider_layout)
-
-        # Supertonic-specific controls (language + steps), hidden by default
-        self.supertonic_row = QWidget()
-        supertonic_row_layout = QHBoxLayout(self.supertonic_row)
-        supertonic_row_layout.setContentsMargins(0, 0, 0, 0)
-        supertonic_row_layout.setSpacing(7)
-
-        st_lang_label = QLabel("Language:", self)
-        supertonic_row_layout.addWidget(st_lang_label)
-        self.st_lang_combo = QComboBox(self)
-        for code in SUPERTONIC_AVAILABLE_LANGS:
-            country_code = SUPERTONIC_LANG_TO_COUNTRY.get(code, code)
-            flag = get_resource_path("abogen.assets.flags", f"{country_code}.png")
-            icon_st = QIcon(flag) if flag and os.path.exists(flag) else QIcon()
-            self.st_lang_combo.addItem(icon_st, code, code)
-        self.st_lang_combo.setCurrentText("en")
-        self.st_lang_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
-        self.st_lang_combo.currentTextChanged.connect(self._on_st_lang_changed)
-        supertonic_row_layout.addWidget(self.st_lang_combo)
-
-        st_steps_label = QLabel("Steps:", self)
-        supertonic_row_layout.addWidget(st_steps_label)
-        self.st_steps_combo = QComboBox(self)
-        for val in range(2, 16):
-            self.st_steps_combo.addItem(str(val), val)
-        self.st_steps_combo.setCurrentIndex(self.st_steps_combo.findData(8))
-        self.st_steps_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
-        self.st_steps_combo.currentIndexChanged.connect(self._on_st_steps_changed)
-        supertonic_row_layout.addWidget(self.st_steps_combo)
-
-        supertonic_row_layout.addStretch()
-        self.supertonic_row.hide()
-        controls_layout.addWidget(self.supertonic_row)
-
         # Voice selection
         voice_layout = QHBoxLayout()
         voice_layout.setSpacing(7)
@@ -1827,12 +1765,6 @@ class abogen(QWidget):
         Update the enabled state of subtitle options based on the selected language.
         For non-English languages, only sentence-based and line-based modes are supported.
         """
-        provider = self.provider_combo.currentData()
-        if provider == "supertonic":
-            self.subtitle_combo.setEnabled(False)
-            self.subtitle_format_combo.setEnabled(False)
-            return
-
         # Check if current file is a subtitle file
         is_subtitle_input = False
         if self.selected_file and self.selected_file.lower().endswith(
@@ -1892,48 +1824,6 @@ class abogen(QWidget):
         # Enable/disable subtitle options based on language
         self.update_subtitle_options_availability()
 
-    def on_provider_changed(self, index):
-        provider = self.provider_combo.itemData(index)
-        self.config["tts_provider"] = provider
-        save_config(self.config)
-        is_supertonic = provider == "supertonic"
-
-        # Show/hide Supertonic controls
-        self.supertonic_row.setVisible(is_supertonic)
-
-        # Update subtitles availability
-        self.update_subtitle_options_availability()
-
-        # Repopulate voice list
-        self.populate_profiles_in_voice_combo()
-
-        # Clear/reset mixed voice state when switching provider
-        if is_supertonic:
-            self.mixed_voice_state = None
-            self.btn_voice_formula_mixer.setEnabled(False)
-            self.voice_combo.setToolTip(
-                "Supertonic voices:\n"
-                "M1-M5 = Male voices\n"
-                "F1-F5 = Female voices"
-            )
-        else:
-            self.btn_voice_formula_mixer.setEnabled(True)
-            self.voice_combo.setToolTip(
-                "The first character represents the language:\n"
-                '"a" => American English\n"b" => British English\n"e" => Spanish\n"f" => French\n"h" => Hindi\n"i" => Italian\n"j" => Japanese\n"p" => Brazilian Portuguese\n"z" => Mandarin Chinese\nThe second character represents the gender:\n"m" => Male\n"f" => Female'
-            )
-
-    def _on_st_lang_changed(self, lang):
-        self.config["supertonic_language"] = lang
-        save_config(self.config)
-        if self.provider_combo.currentData() == "supertonic":
-            self.selected_lang = lang
-            self.update_subtitle_options_availability()
-
-    def _on_st_steps_changed(self):
-        self.config["supertonic_total_steps"] = self.st_steps_combo.currentData()
-        save_config(self.config)
-
     def on_voice_combo_changed(self, index):
         data = self.voice_combo.itemData(index)
         if isinstance(data, str) and data.startswith("profile:"):
@@ -1942,26 +1832,10 @@ class abogen(QWidget):
             from abogen.voice_profiles import load_profiles
 
             entry = load_profiles().get(pname, {})
+            # set mixed voices and language
             if isinstance(entry, dict):
-                entry_provider = str(entry.get("provider", "")).strip().lower()
+                self.mixed_voice_state = entry.get("voices", [])
-                if entry_provider == "supertonic":
+                self.selected_lang = entry.get("language")
-                    # Switch provider to Supertonic if not already
-                    if self.provider_combo.currentData() != "supertonic":
-                        self.provider_combo.setCurrentIndex(1)
-                    self.mixed_voice_state = None
-                    self.selected_lang = entry.get("language", self.st_lang_combo.currentText())
-                    # Sync supertonic controls from profile
-                    profile_steps = entry.get("total_steps")
-                    if profile_steps is not None:
-                        idx_steps = self.st_steps_combo.findData(int(profile_steps))
-                        if idx_steps >= 0:
-                            self.st_steps_combo.setCurrentIndex(idx_steps)
-                    profile_lang = entry.get("language")
-                    if profile_lang and profile_lang in SUPERTONIC_AVAILABLE_LANGS:
-                        self.st_lang_combo.setCurrentText(profile_lang)
-                else:
-                    self.mixed_voice_state = entry.get("voices", [])
-                    self.selected_lang = entry.get("language")
             else:
                 self.mixed_voice_state = entry
                 self.selected_lang = entry[0][0] if entry and entry[0] else None
@@ -1974,12 +1848,7 @@ class abogen(QWidget):
         else:
             self.mixed_voice_state = None
             self.selected_profile_name = None
-            self.selected_voice = data
+            self.selected_voice, self.selected_lang = data, data[0]
-            provider = self.provider_combo.currentData()
-            if provider == "supertonic":
-                self.selected_lang = self.st_lang_combo.currentText()
-            else:
-                self.selected_lang = data[0] if data else ""
             self.config["selected_voice"] = data
             if "selected_profile_name" in self.config:
                 del self.config["selected_profile_name"]
@@ -1998,40 +1867,19 @@ class abogen(QWidget):
     def populate_profiles_in_voice_combo(self):
         # preserve current voice or profile
         current = self.voice_combo.currentData()
-        provider = self.provider_combo.currentData()
         self.voice_combo.blockSignals(True)
         self.voice_combo.clear()
-        # re-add profiles matching current provider
+        # re-add profiles
         profile_icon = QIcon(get_resource_path("abogen.assets", "profile.png"))
-        for pname, entry in load_profiles().items():
+        for pname in load_profiles().keys():
-            entry_provider = ""
+            self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
-            if isinstance(entry, dict):
-                entry_provider = str(entry.get("provider", "")).strip().lower()
-            if provider == "supertonic":
-                if entry_provider == "supertonic":
-                    self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
-            else:
-                if entry_provider != "supertonic":
-                    self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
         # re-add voices
-        if provider == "supertonic":
+        for v in get_voices("kokoro"):
-            for v in DEFAULT_SUPERTONIC_VOICES:
+            icon = QIcon()
-                icon = QIcon()
+            flag_path = get_resource_path("abogen.assets.flags", f"{v[0]}.png")
-                if v.startswith("F"):
+            if flag_path and os.path.exists(flag_path):
-                    icon_path = get_resource_path("abogen.assets", "female.png")
+                icon = QIcon(flag_path)
-                else:
+            self.voice_combo.addItem(icon, f"{v}", v)
-                    icon_path = get_resource_path("abogen.assets", "male.png")
-                if icon_path and os.path.exists(icon_path):
-                    icon = QIcon(icon_path)
-                self.voice_combo.addItem(icon, f"{v}", v)
-        else:
-            for v in VOICES_INTERNAL:
-                icon = QIcon()
-                country_code = KOKORO_LANG_TO_COUNTRY.get(v[0], v[0])
-                flag_path = get_resource_path("abogen.assets.flags", f"{country_code}.png")
-                if flag_path and os.path.exists(flag_path):
-                    icon = QIcon(flag_path)
-                self.voice_combo.addItem(icon, f"{v}", v)
         # restore selection
         idx = -1
         if self.selected_profile_name:
@@ -2222,9 +2070,6 @@ class abogen(QWidget):
             save_base_path=save_base_path,
             save_chapters_separately=getattr(self, "save_chapters_separately", None),
             merge_chapters_at_end=getattr(self, "merge_chapters_at_end", None),
-            tts_provider=self.provider_combo.currentData(),
-            supertonic_language=self.st_lang_combo.currentText(),
-            supertonic_total_steps=self.st_steps_combo.currentData(),
         )
 
         # Prevent adding duplicate items to the queue
@@ -2368,15 +2213,6 @@ class abogen(QWidget):
                 self.config["replace_numerals"] = self.replace_numerals
                 self.config["fix_nonstandard_punctuation"] = self.fix_nonstandard_punctuation
 
-                # TTS provider settings
-                tts_provider = getattr(queued_item, "tts_provider", "kokoro")
-                self.provider_combo.setCurrentText("Supertonic" if tts_provider == "supertonic" else "Kokoro")
-                self.st_lang_combo.setCurrentText(getattr(queued_item, "supertonic_language", "en"))
-                steps_val = getattr(queued_item, "supertonic_total_steps", 8)
-                idx_steps = self.st_steps_combo.findData(steps_val)
-                if idx_steps >= 0:
-                    self.st_steps_combo.setCurrentIndex(idx_steps)
-
                 # Sync Voice/Profile in config
                 self.config["selected_voice"] = self.selected_voice
                 if "selected_profile_name" in self.config:
@@ -2399,8 +2235,6 @@ class abogen(QWidget):
             self.current_queue_index = 0  # Reset for next time
 
     def get_voice_formula(self) -> str:
-        if self.provider_combo.currentData() == "supertonic":
-            return self._get_supertonic_voice()
         if self.mixed_voice_state:
             formula_components = [
                 f"{name}*{weight}" for name, weight in self.mixed_voice_state
@@ -2410,8 +2244,6 @@ class abogen(QWidget):
             return self.selected_voice
 
     def get_selected_lang(self, voice_formula) -> str:
-        if self.provider_combo.currentData() == "supertonic":
-            return self.st_lang_combo.currentText()
         if self.selected_profile_name:
             from abogen.voice_profiles import load_profiles
 
@@ -2485,9 +2317,9 @@ class abogen(QWidget):
             file_size_str = "Unknown"
 
         # pipeline_loaded_callback remains unchanged
-        def pipeline_loaded_callback(np_module, kpipeline_class, error):
+        def pipeline_loaded_callback(backend, error):
             if error:
-                self.update_log((f"Error loading numpy or KPipeline: {error}", "red"))
+                self.update_log((f"Error loading TTS backend: {error}", "red"))
                 prevent_sleep_end()
                 return
 
@@ -2501,10 +2333,6 @@ class abogen(QWidget):
             # determine selected language: use profile setting if profile selected, else voice code
             selected_lang = self.get_selected_lang(voice_formula)
 
-            tts_provider = self.provider_combo.currentData()
-            supertonic_language = self.st_lang_combo.currentText()
-            supertonic_total_steps = self.st_steps_combo.currentData()
-
             self.conversion_thread = ConversionThread(
                 self.selected_file,
                 selected_lang,
@@ -2514,17 +2342,13 @@ class abogen(QWidget):
                 self.selected_output_folder,
                 subtitle_mode=actual_subtitle_mode,
                 output_format=self.selected_format,
-                np_module=np_module,
+                backend=backend,
-                kpipeline_class=kpipeline_class,
                 start_time=self.start_time,
                 total_char_count=self.char_count,
                 use_gpu=self.gpu_ok,
                 from_queue=from_queue,
-                save_base_path=self.displayed_file_path,
+                save_base_path=self.displayed_file_path,  # Pass the save base path (original file for EPUB)
-                tts_provider=tts_provider,
+            )  # Use gpu_ok status
-                supertonic_language=supertonic_language,
-                supertonic_total_steps=supertonic_total_steps,
-            )
             # Pass the displayed file path to the log_updated signal handler in ConversionThread
             self.conversion_thread.display_path = display_path
             # Pass the file size string
@@ -2601,15 +2425,19 @@ class abogen(QWidget):
             # Store gpu_ok status to use when creating the conversion thread
             self.gpu_ok = gpu_ok
             self.update_log((gpu_msg, gpu_ok))
-            tts_provider = self.provider_combo.currentData()
+            self.update_log("Loading modules...")
-            if tts_provider == "supertonic":
+
-                # Supertonic doesn't need KPipeline, call callback directly
+            # Determine device based on GPU availability
-                import numpy as np
+            if gpu_ok:
-                pipeline_loaded_callback(np, None, None)
+                device = _select_device()
             else:
-                self.update_log("Loading modules...")
+                device = "cpu"
-                load_thread = LoadPipelineThread(pipeline_loaded_callback)
+
-                load_thread.start()
+            lang_code = self.selected_lang or "a"
+            load_thread = LoadPipelineThread(
+                pipeline_loaded_callback, lang_code=lang_code, device=device
+            )
+            load_thread.start()
 
         threading.Thread(target=gpu_and_load, daemon=True).start()
 
@@ -2922,32 +2750,9 @@ class abogen(QWidget):
                 "Open File Error", f"Could not open file:\n{e}"
             )
 
-    def _get_supertonic_voice(self) -> str:
-        """Resolve the effective Supertonic voice from the current combo selection."""
-        if self.selected_profile_name:
-            from abogen.voice_profiles import load_profiles
-            entry = load_profiles().get(self.selected_profile_name, {})
-            if isinstance(entry, dict):
-                return str(entry.get("voice", "M1"))
-            return "M1"
-        current_data = self.voice_combo.currentData()
-        if current_data and isinstance(current_data, str) and not current_data.startswith("profile:"):
-            return current_data
-        return "M1"
-
     def _get_preview_cache_path(self):
         """Generate the expected cache path for the current voice settings."""
         speed = self.speed_slider.value() / 100.0
-        provider = self.provider_combo.currentData()
-
-        if provider == "supertonic":
-            voice_to_cache = self._get_supertonic_voice()
-            lang_to_cache = self.st_lang_combo.currentText()
-            steps = self.st_steps_combo.currentData()
-            cache_dir = get_user_cache_path("preview_cache")
-            filename = f"st_{voice_to_cache}_{lang_to_cache}_steps{steps}_{speed:.2f}.wav"
-            return os.path.join(cache_dir, filename)
-
         voice_to_cache = ""
         lang_to_cache = ""
 
@@ -3052,13 +2857,6 @@ class abogen(QWidget):
         self.btn_voice_formula_mixer.setEnabled(False)  # Disable mixer button
         self.btn_start.setEnabled(False)  # Disable start button during preview
 
-        # For Supertonic, skip KPipeline loading and use SupertonicPipeline directly
-        if self.provider_combo.currentData() == "supertonic":
-            import numpy as np
-            self.loading_movie.start()
-            self._on_pipeline_loaded_for_preview(np, None, None)
-            return
-
         # Start loading animation - ensure signal connection is always active
         if hasattr(self, "loading_movie"):
             # Disconnect previous connections to avoid multiple connections
@@ -3075,18 +2873,24 @@ class abogen(QWidget):
             )
             self.loading_movie.start()
 
-        def pipeline_loaded_callback(np_module, kpipeline_class, error):
+        # Determine device based on GPU availability
-            self._on_pipeline_loaded_for_preview(np_module, kpipeline_class, error)
+        if self.gpu_ok:
+            device = _select_device()
+        else:
+            device = "cpu"
 
-        load_thread = LoadPipelineThread(pipeline_loaded_callback)
+        lang = self.selected_lang or "a"
+        load_thread = LoadPipelineThread(
+            self._on_pipeline_loaded_for_preview, lang_code=lang, device=device
+        )
         load_thread.start()
 
-    def _on_pipeline_loaded_for_preview(self, np_module, kpipeline_class, error):
+    def _on_pipeline_loaded_for_preview(self, backend, error):
         # stop loading animation and restore icon on error
         if error:
             self.loading_movie.stop()
             self._show_error_message_box(
-                "Loading Error", f"Error loading numpy or KPipeline: {error}"
+                "Loading Error", f"Error loading TTS backend: {error}"
             )
             self.btn_preview.setIcon(self.play_icon)
             self.btn_preview.setEnabled(True)
@@ -3117,28 +2921,14 @@ class abogen(QWidget):
                     else None
                 )
         else:
-            if self.provider_combo.currentData() == "supertonic":
+            lang = self.selected_voice[0]
-                voice = self._get_supertonic_voice()
+            voice = self.selected_voice
-            else:
-                voice = self.selected_voice or ""
-
-        tts_provider = self.provider_combo.currentData()
-        supertonic_language = self.st_lang_combo.currentText()
-        supertonic_total_steps = self.st_steps_combo.currentData()
-
-        if tts_provider == "supertonic":
-            lang = supertonic_language
-        else:
-            lang = self.selected_voice[0] if self.selected_voice else ""
 
         # use same gpu/cpu logic as in conversion
         gpu_msg, gpu_ok = get_gpu_acceleration(self.use_gpu)
 
         self.preview_thread = VoicePreviewThread(
-            np_module, kpipeline_class, lang, voice, speed, gpu_ok,
+            backend, lang, voice, speed, gpu_ok
-            tts_provider=tts_provider,
-            supertonic_language=supertonic_language,
-            supertonic_total_steps=supertonic_total_steps,
         )
         self.preview_thread.finished.connect(self._play_preview_audio)
         self.preview_thread.error.connect(self._preview_error)
@@ -3441,12 +3231,16 @@ class abogen(QWidget):
             )
             box.setDefaultButton(QMessageBox.StandardButton.No)
             if box.exec() == QMessageBox.StandardButton.Yes:
+                from abogen import shutdown
+                shutdown.request_shutdown()
                 self.cleanup_conversion_thread()
                 self.cleanup_preview_threads()
                 event.accept()
             else:
                 event.ignore()
         else:
+            from abogen import shutdown
+            shutdown.request_shutdown()
             self.cleanup_conversion_thread()
             self.cleanup_preview_threads()
             event.accept()

abogen/pyqt/main.py

@@ -1,10 +1,10 @@
 import os
 import sys
 import platform
-import atexit
-import signal
-from abogen.utils import get_resource_path, load_config, prevent_sleep_end
 
+# Initialise global shutdown handling (atexit, signals, Qt) as early as possible.
+from abogen import shutdown  # noqa: F401
+shutdown.register_shutdown()
 
 # Fix PyTorch DLL loading issue ([WinError 1114]) on Windows before importing PyQt6
 if platform.system() == "Windows":
@@ -94,6 +94,7 @@ os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"  # Disable Hugging Face telemetry
 os.environ["HF_HUB_ETAG_TIMEOUT"] = "10"  # Metadata request timeout (seconds)
 os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "10"  # File download timeout (seconds)
 os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"  # Disable symlinks warning
+from abogen.utils import load_config
 if load_config().get("disable_kokoro_internet", False):
     print("INFO: Kokoro's internet access is disabled.")
     os.environ["HF_HUB_OFFLINE"] = "1"  # Disable Hugging Face Hub internet access
@@ -105,25 +106,6 @@ from abogen.constants import PROGRAM_NAME, VERSION
 os.environ["MIOPEN_FIND_MODE"] = "FAST"
 os.environ["MIOPEN_CONV_PRECISE_ROCM_TUNING"] = "0"
 
-# Reset sleep states
-atexit.register(prevent_sleep_end)
-
-
-# Also handle signals (Ctrl+C, kill, etc.)
-def _cleanup_sleep(signum, frame):
-    prevent_sleep_end()
-    sys.exit(0)
-
-
-signal.signal(signal.SIGINT, _cleanup_sleep)
-signal.signal(signal.SIGTERM, _cleanup_sleep)
-
-# Ensure sys.stdout and sys.stderr are valid in GUI mode
-if sys.stdout is None:
-    sys.stdout = open(os.devnull, "w")
-if sys.stderr is None:
-    sys.stderr = open(os.devnull, "w")
-
 # Enable MPS GPU acceleration on Mac Apple Silicon
 if platform.system() == "Darwin" and platform.processor() == "arm":
     os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1"
@@ -184,4 +166,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    main()

abogen/pyqt/predownload_gui.py

@@ -21,7 +21,8 @@ from PyQt6.QtWidgets import (
 )
 from PyQt6.QtCore import QThread, pyqtSignal
 
-from abogen.constants import COLORS, VOICES_INTERNAL
+from abogen.constants import COLORS
+from abogen.tts_plugin.utils import get_voices
 from abogen.spacy_utils import SPACY_MODELS
 import abogen.hf_tracker
 
@@ -114,7 +115,7 @@ class PreDownloadWorker(QThread):
             self._voices_success = False
             return
 
-        voice_list = VOICES_INTERNAL
+        voice_list = get_voices("kokoro")
         for idx, voice in enumerate(voice_list, start=1):
             if self._cancelled:
                 self._voices_success = False
@@ -462,14 +463,14 @@ class PreDownloadDialog(QDialog):
         try:
             from huggingface_hub import try_to_load_from_cache
 
-            for voice in VOICES_INTERNAL:
+            for voice in get_voices("kokoro"):
                 if not try_to_load_from_cache(
                     repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
                 ):
                     missing.append(voice)
         except Exception:
             # If HF missing, report all as missing
-            return False, list(VOICES_INTERNAL)
+            return False, list(get_voices("kokoro"))
         return (len(missing) == 0), missing
 
     def _check_kokoro_model(self) -> bool:

abogen/pyqt/queued_item.py

@@ -26,7 +26,3 @@ class QueuedItem:
     replace_all_caps: bool = False
     replace_numerals: bool = False
     fix_nonstandard_punctuation: bool = False
-    # TTS Provider fields
-    tts_provider: str = "kokoro"
-    supertonic_language: str = "en"
-    supertonic_total_steps: int = 8

abogen/pyqt/voice_formula_gui.py

@@ -28,12 +28,11 @@ from PyQt6.QtWidgets import (
 from PyQt6.QtCore import Qt, QTimer, QPoint, QRect, QSize
 from PyQt6.QtGui import QPixmap, QIcon, QAction
 from abogen.constants import (
-    VOICES_INTERNAL,
     SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
     LANGUAGE_DESCRIPTIONS,
-    KOKORO_LANG_TO_COUNTRY,
     COLORS,
 )
+from abogen.tts_plugin.utils import get_voices
 import re
 import platform
 from abogen.utils import get_resource_path
@@ -180,7 +179,7 @@ class VoiceMixer(QWidget):
         layout.addWidget(QLabel(name), alignment=Qt.AlignmentFlag.AlignCenter)
 
         # Voice name label with gender icon
-        is_female = self.voice_name in VOICES_INTERNAL and self.voice_name[1] == "f"
+        is_female = self.voice_name in get_voices("kokoro") and self.voice_name[1] == "f"
 
         # Icons layout (flag and gender)
         icons_layout = QHBoxLayout()
@@ -190,9 +189,8 @@ class VoiceMixer(QWidget):
         )  # Center the icons horizontally
 
         # Flag icon
-        country_code = KOKORO_LANG_TO_COUNTRY.get(language_code, language_code)
         flag_icon_path = get_resource_path(
-            "abogen.assets.flags", f"{country_code}.png"
+            "abogen.assets.flags", f"{language_code}.png"
         )
         gender_icon_path = get_resource_path(
             "abogen.assets", "female.png" if is_female else "male.png"
@@ -514,8 +512,7 @@ class VoiceFormulaDialog(QDialog):
         header_row.addWidget(QLabel("Language:"))
         self.language_combo = QComboBox()
         for code, desc in LANGUAGE_OPTIONS:
-            country_code = KOKORO_LANG_TO_COUNTRY.get(code, code)
+            flag = get_resource_path("abogen.assets.flags", f"{code}.png")
-            flag = get_resource_path("abogen.assets.flags", f"{country_code}.png")
             if flag and os.path.exists(flag):
                 self.language_combo.addItem(QIcon(flag), desc, code)
             else:
@@ -775,7 +772,7 @@ class VoiceFormulaDialog(QDialog):
 
     def add_voices(self, initial_state):
         first_enabled_voice = None
-        for voice in VOICES_INTERNAL:
+        for voice in get_voices("kokoro"):
             language_code = voice[0]  # First character is the language code
             matching_voice = next(
                 (item for item in initial_state if item[0] == voice), None

abogen/shutdown.py

@@ -0,0 +1,160 @@
+"""Graceful shutdown - single module, no over-engineering."""
+from __future__ import annotations
+
+import atexit
+import gc
+import signal
+import sys
+from typing import Callable
+
+_CLEANUP_FUNCS: list[Callable[[], None]] = []
+_EXECUTED = False
+
+
+def register_cleanup(fn: Callable[[], None]) -> None:
+    """Register a cleanup function to run on shutdown."""
+    _CLEANUP_FUNCS.append(fn)
+
+
+def _run_cleanups() -> None:
+    global _EXECUTED
+    if _EXECUTED:
+        return
+    _EXECUTED = True
+    for fn in _CLEANUP_FUNCS:
+        try:
+            fn()
+        except Exception:
+            pass
+
+
+# ---- Register built-in cleanup functions ----
+
+# 1. Restore sleep prevention
+def _restore_sleep() -> None:
+    try:
+        from abogen.utils import prevent_sleep_end
+        prevent_sleep_end()
+    except Exception:
+        pass
+
+register_cleanup(_restore_sleep)
+
+# 2. Shutdown web UI ConversionService
+def _shutdown_conversion_service() -> None:
+    try:
+        from abogen.webui.service import get_service
+        svc = get_service()
+        if svc is not None:
+            svc.shutdown()
+    except Exception:
+        pass
+
+register_cleanup(_shutdown_conversion_service)
+
+# 3. Clear TTS pipelines and GPU memory
+def _cleanup_tts_pipelines() -> None:
+    # Clear web UI pipeline cache
+    try:
+        from abogen.webui.conversion_runner import _PIPELINES
+        _PIPELINES.clear()
+    except Exception:
+        pass
+
+    # Clear PyQt conversion thread voice cache
+    try:
+        from abogen.pyqt.conversion import ConversionThread
+        if hasattr(ConversionThread, "voice_cache"):
+            ConversionThread.voice_cache.clear()
+    except Exception:
+        pass
+
+    gc.collect()
+
+    # Release CUDA cache
+    try:
+        import torch
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+            torch.cuda.ipc_collect()
+    except Exception:
+        pass
+
+register_cleanup(_cleanup_tts_pipelines)
+
+# 4. Clear global voice cache
+def _clear_voice_cache() -> None:
+    try:
+        from abogen.voice_cache import clear_voice_cache
+        clear_voice_cache()
+    except Exception:
+        pass
+
+register_cleanup(_clear_voice_cache)
+
+# 5. Terminate child processes (ffmpeg, etc.)
+def _terminate_subprocesses() -> None:
+    try:
+        import psutil
+    except Exception:
+        return
+
+    try:
+        current = psutil.Process()
+        for child in current.children(recursive=True):
+            try:
+                child.terminate()
+            except Exception:
+                pass
+        gone, alive = psutil.wait_procs(current.children(recursive=True), timeout=3)
+        for proc in alive:
+            try:
+                proc.kill()
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+register_cleanup(_terminate_subprocesses)
+
+
+def register_shutdown() -> None:
+    """Install process-wide shutdown hooks (atexit, signals, Qt)."""
+    if register_shutdown._registered:
+        return
+    register_shutdown._registered = True
+
+    atexit.register(_run_cleanups)
+
+    # POSIX signals
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            signal.signal(sig, _on_signal)
+        except Exception:
+            pass
+
+    # Qt hook
+    try:
+        from PyQt6.QtWidgets import QApplication
+
+        app = QApplication.instance()
+        if app is not None:
+            app.aboutToQuit.connect(_run_cleanups)
+    except Exception:
+        pass
+
+
+register_shutdown._registered = False
+
+
+def _on_signal(signum: int, _frame) -> None:
+    _run_cleanups()
+    sys.exit(0)
+
+
+def request_shutdown() -> None:
+    """Programmatically trigger cleanup (e.g., from GUI closeEvent)."""
+    _run_cleanups()
+
+
+__all__ = ["register_shutdown", "request_shutdown", "register_cleanup"]

abogen/subtitle_utils.py

@@ -466,7 +466,7 @@ def sanitize_name_for_os(name, is_folder=True):
 
 
 def validate_voice_name(voice_name):
-    """Validate voice name against VOICES_INTERNAL list (case-insensitive).
+    """Validate voice name against available voices (case-insensitive).
     Handles both single voices and formulas like 'af_heart*0.5 + am_echo*0.5'.
 
     Args:
@@ -477,10 +477,10 @@ def validate_voice_name(voice_name):
             - is_valid: True if all voices in the name/formula are valid
             - invalid_voice_name: The first invalid voice found, or None if all valid
     """
-    from abogen.constants import VOICES_INTERNAL
+    from abogen.tts_plugin.utils import get_voices
 
     # Create case-insensitive lookup set (done once per call)
-    voice_lookup_lower = {v.lower() for v in VOICES_INTERNAL}
+    voice_lookup_lower = {v.lower() for v in get_voices("kokoro")}
     voice_name = voice_name.strip()
 
     # Check if it's a formula (contains *)
@@ -505,7 +505,7 @@ def split_text_by_voice_markers(text, default_voice):
     """Split text by voice markers, returning list of (voice, text) tuples.
 
     IMPORTANT: Returns the last voice used so it can persist across chapters.
-    Voice names are normalized to lowercase to match VOICES_INTERNAL.
+    Voice names are normalized to lowercase to match canonical voice names.
 
     Args:
         text: Text potentially containing <<VOICE:name>> markers
@@ -518,7 +518,7 @@ def split_text_by_voice_markers(text, default_voice):
             - valid_count: Number of valid voice markers processed
             - invalid_count: Number of invalid voice markers skipped
     """
-    from abogen.constants import VOICES_INTERNAL
+    from abogen.tts_plugin.utils import get_voices
 
     voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text))
 
@@ -560,7 +560,7 @@ def split_text_by_voice_markers(text, default_voice):
                         # Find the canonical (lowercase) voice name
                         voice_part_lower = voice_part.strip().lower()
                         canonical_voice = next(
-                            (v for v in VOICES_INTERNAL if v.lower() == voice_part_lower),
+                            (v for v in get_voices("kokoro") if v.lower() == voice_part_lower),
                             voice_part.strip()
                         )
                         normalized_parts.append(f"{canonical_voice}*{weight.strip()}")
@@ -569,7 +569,7 @@ def split_text_by_voice_markers(text, default_voice):
                 # Find the canonical (lowercase) voice name
                 voice_name_lower = voice_name.lower()
                 current_voice = next(
-                    (v for v in VOICES_INTERNAL if v.lower() == voice_name_lower),
+                    (v for v in get_voices("kokoro") if v.lower() == voice_name_lower),
                     voice_name
                 )
             valid_markers += 1

abogen/tts_plugin/__init__.py

@@ -0,0 +1,170 @@
+"""TTS Plugin Architecture - Public API.
+
+This package defines the frozen Plugin API for the TTS Plugin Architecture.
+All public interfaces are fully defined but contain no business logic.
+
+Public modules:
+- types: Core domain value objects (AudioFormat, Duration, VoiceSelection, etc.)
+- errors: Error hierarchy (EngineError and subtypes)
+- manifest: Plugin manifest types (PluginManifest, EngineManifest, etc.)
+- engine: Engine and EngineSession protocols
+- capabilities: Optional capability interfaces (VoiceLister, PreviewGenerator, etc.)
+- host_context: HostContext dataclass
+- plugin: Plugin contract (create_engine function signature)
+- loader: Plugin discovery and loading
+- plugin_manager: Plugin management and engine creation
+- utils: Direct utility functions (get_voices, create_pipeline, etc.)
+
+Usage:
+    from abogen.tts_plugin import (
+        # Types
+        AudioFormat,
+        Duration,
+        VoiceSelection,
+        ParameterValues,
+        SynthesisRequest,
+        SynthesizedAudio,
+        EngineConfig,
+        # Errors
+        EngineError,
+        ModelNotFoundError,
+        ModelLoadError,
+        NetworkError,
+        InvalidInputError,
+        ConfigurationError,
+        CancelledError,
+        InternalError,
+        # Manifest
+        PluginManifest,
+        EngineManifest,
+        VoiceSourceManifest,
+        VoiceManifest,
+        ParameterManifest,
+        AudioFormatManifest,
+        EnumOption,
+        RequirementManifest,
+        GpuRequirement,
+        ModelManifest,
+        # Engine
+        Engine,
+        EngineSession,
+        # Capabilities
+        VoiceLister,
+        PreviewGenerator,
+        StreamingSynthesizer,
+        CancelableSession,
+        # Host Context
+        HostContext,
+        HttpClient,
+        # Plugin Manager
+        get_plugin_manager,
+        reset_plugin_manager,
+        # Utils
+        get_voices,
+        get_default_voice,
+        is_plugin_registered,
+        resolve_voice_to_plugin,
+        create_pipeline,
+    )
+"""
+
+from abogen.tts_plugin.capabilities import (
+    CancelableSession,
+    PreviewGenerator,
+    StreamingSynthesizer,
+    VoiceLister,
+)
+from abogen.tts_plugin.engine import Engine, EngineSession
+from abogen.tts_plugin.errors import (
+    CancelledError,
+    ConfigurationError,
+    EngineError,
+    InternalError,
+    InvalidInputError,
+    ModelLoadError,
+    ModelNotFoundError,
+    NetworkError,
+)
+from abogen.tts_plugin.host_context import HttpClient, HostContext
+from abogen.tts_plugin.manifest import (
+    AudioFormatManifest,
+    EngineManifest,
+    EnumOption,
+    GpuRequirement,
+    ModelManifest,
+    ParameterManifest,
+    PluginManifest,
+    RequirementManifest,
+    VoiceManifest,
+    VoiceSourceManifest,
+)
+from abogen.tts_plugin.types import (
+    AudioFormat,
+    Duration,
+    EngineConfig,
+    ParameterValues,
+    SynthesisRequest,
+    SynthesizedAudio,
+    VoiceSelection,
+)
+
+# Plugin Manager and Utils
+from abogen.tts_plugin.plugin_manager import get_plugin_manager, reset_plugin_manager
+from abogen.tts_plugin.utils import (
+    create_pipeline,
+    get_default_voice,
+    get_voices,
+    is_plugin_registered,
+    resolve_voice_to_plugin,
+)
+
+__all__ = [
+    # Types
+    "AudioFormat",
+    "Duration",
+    "VoiceSelection",
+    "ParameterValues",
+    "SynthesisRequest",
+    "SynthesizedAudio",
+    "EngineConfig",
+    # Errors
+    "EngineError",
+    "ModelNotFoundError",
+    "ModelLoadError",
+    "NetworkError",
+    "InvalidInputError",
+    "ConfigurationError",
+    "CancelledError",
+    "InternalError",
+    # Manifest
+    "PluginManifest",
+    "EngineManifest",
+    "VoiceSourceManifest",
+    "VoiceManifest",
+    "ParameterManifest",
+    "AudioFormatManifest",
+    "EnumOption",
+    "RequirementManifest",
+    "GpuRequirement",
+    "ModelManifest",
+    # Engine
+    "Engine",
+    "EngineSession",
+    # Capabilities
+    "VoiceLister",
+    "PreviewGenerator",
+    "StreamingSynthesizer",
+    "CancelableSession",
+    # Host Context
+    "HostContext",
+    "HttpClient",
+    # Plugin Manager
+    "get_plugin_manager",
+    "reset_plugin_manager",
+    # Utils
+    "get_voices",
+    "get_default_voice",
+    "is_plugin_registered",
+    "resolve_voice_to_plugin",
+    "create_pipeline",
+]

abogen/tts_plugin/capabilities.py

@@ -0,0 +1,103 @@
+"""Capability interfaces for the TTS Plugin Architecture.
+
+This module defines optional capability interfaces that engines can implement.
+Capabilities are additive; implementing new capabilities doesn't break old plugins.
+"""
+
+from __future__ import annotations
+
+from typing import Iterator, Protocol, runtime_checkable
+
+from abogen.tts_plugin.manifest import VoiceManifest
+from abogen.tts_plugin.types import SynthesisRequest, SynthesizedAudio, VoiceSelection
+
+
+@runtime_checkable
+class VoiceLister(Protocol):
+    """Protocol for listing available voices.
+
+    Engines that support voice listing should implement this interface.
+    """
+
+    def listVoices(self, sourceId: str) -> list[VoiceManifest]:
+        """List available voices for a given source.
+
+        Args:
+            sourceId: The voice source identifier.
+
+        Returns:
+            List of VoiceManifest describing available voices.
+
+        Raises:
+            EngineError: On failure.
+        """
+        ...
+
+
+@runtime_checkable
+class PreviewGenerator(Protocol):
+    """Protocol for generating voice previews.
+
+    Engines that support voice preview should implement this interface.
+    """
+
+    def generatePreview(self, voice: VoiceSelection, text: str) -> SynthesizedAudio:
+        """Generate a preview audio for a voice.
+
+        Args:
+            voice: Voice selection for the preview.
+            text: Text to use for the preview.
+
+        Returns:
+            SynthesizedAudio with the preview audio data.
+
+        Raises:
+            EngineError: On failure.
+        """
+        ...
+
+
+@runtime_checkable
+class StreamingSynthesizer(Protocol):
+    """Protocol for streaming synthesis.
+
+    Optional capability of EngineSession, not Engine.
+    Engines that support streaming synthesis should implement this interface.
+    """
+
+    def synthesizeStream(self, request: SynthesisRequest) -> Iterator[bytes]:
+        """Synthesize audio in streaming mode.
+
+        Args:
+            request: The synthesis request.
+
+        Yields:
+            Audio chunks as they become available.
+
+        Raises:
+            CancelledError: If cancel() is called during iteration.
+            EngineError: On synthesis failure.
+        """
+        ...
+        # This is a generator function; implementation will use yield
+        yield b""  # pragma: no cover
+
+
+@runtime_checkable
+class CancelableSession(Protocol):
+    """Protocol for cancellation support.
+
+    Optional capability for engines that support cancellation.
+    cancel() causes synthesize() to raise CancelledError.
+    """
+
+    def cancel(self) -> None:
+        """Cancel in-progress synthesis.
+
+        After cancellation, synthesize() raises CancelledError.
+        The session remains usable after cancellation.
+
+        Raises:
+            EngineError: If called after dispose().
+        """
+        ...

abogen/tts_plugin/engine.py

@@ -0,0 +1,95 @@
+"""Engine interfaces for the TTS Plugin Architecture.
+
+This module defines the core Engine and EngineSession protocols.
+These are the primary interfaces that plugin implementations must satisfy.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from abogen.tts_plugin.types import SynthesisRequest, SynthesizedAudio
+
+
+@runtime_checkable
+class EngineSession(Protocol):
+    """Protocol for a session that owns mutable execution state.
+
+    An EngineSession is created by Engine.createSession() and owns
+    mutable execution state isolated from other concurrent work.
+    It is NOT thread-safe.
+
+    Lifecycle:
+        1. Created by Engine.createSession()
+        2. Used for synthesis via synthesize()
+        3. Disposed via dispose()
+
+    After dispose(), all methods except dispose() raise EngineError.
+    """
+
+    def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio:
+        """Synthesize audio from text.
+
+        Args:
+            request: The synthesis request containing text, voice, parameters, and format.
+
+        Returns:
+            SynthesizedAudio with the synthesized audio data.
+
+        Raises:
+            EngineError: On synthesis failure. Session remains usable after error.
+            EngineError: If called after dispose().
+        """
+        ...
+
+    def dispose(self) -> None:
+        """Release session resources.
+
+        This method is idempotent and safe to call multiple times.
+        It never raises exceptions (catches and logs internally).
+        After dispose(), all methods except dispose() raise EngineError.
+        """
+        ...
+
+
+@runtime_checkable
+class Engine(Protocol):
+    """Protocol for a TTS engine that creates sessions.
+
+    An Engine is a factory for EngineSession instances. It is stateless
+    and thread-safe for createSession().
+
+    Lifecycle:
+        1. Created via create_engine() (plugin contract)
+        2. Sessions created via createSession()
+        3. Disposed via dispose()
+
+    Thread Safety:
+        - createSession() is thread-safe and can be called from any thread.
+        - dispose() must be called after all sessions are disposed.
+        - Disposing engine while sessions are alive violates API contract.
+    """
+
+    def createSession(self) -> EngineSession:
+        """Create a new session for synthesis.
+
+        Returns:
+            A new EngineSession instance. Ownership transfers to caller.
+
+        Raises:
+            EngineError: On failure. No partially initialized session is returned.
+        """
+        ...
+
+    def dispose(self) -> None:
+        """Release engine resources.
+
+        Caller must ensure all sessions created by this engine are disposed
+        before calling dispose(). Disposing an engine while any session is
+        still alive violates the API contract; behavior is undefined.
+
+        This method is idempotent and safe to call multiple times.
+        It never raises exceptions (catches and logs internally).
+        After dispose(), all methods except dispose() raise EngineError.
+        """
+        ...

abogen/tts_plugin/errors.py

@@ -0,0 +1,62 @@
+"""Error hierarchy for the TTS Plugin Architecture.
+
+This module defines typed exceptions that engines raise.
+Engines should never raise raw exceptions; they must use EngineError or its subtypes.
+"""
+
+from __future__ import annotations
+
+
+class EngineError(Exception):
+    """Base exception for all engine errors.
+
+    All engine operations that can fail should raise EngineError or one of its subtypes.
+    After dispose(), all methods except dispose() raise EngineError.
+    """
+
+    pass
+
+
+class ModelNotFoundError(EngineError):
+    """Raised when a required model is not found."""
+
+    pass
+
+
+class ModelLoadError(EngineError):
+    """Raised when a model fails to load."""
+
+    pass
+
+
+class NetworkError(EngineError):
+    """Raised when a network operation fails."""
+
+    pass
+
+
+class InvalidInputError(EngineError):
+    """Raised when invalid input is provided to the engine."""
+
+    pass
+
+
+class ConfigurationError(EngineError):
+    """Raised when there is a configuration error."""
+
+    pass
+
+
+class CancelledError(EngineError):
+    """Raised when an operation is cancelled.
+
+    This is raised by synthesize() when cancel() is called during synthesis.
+    """
+
+    pass
+
+
+class InternalError(EngineError):
+    """Raised when an internal engine error occurs."""
+
+    pass

abogen/tts_plugin/host_context.py

@@ -0,0 +1,46 @@
+"""Host context for the TTS Plugin Architecture.
+
+This module defines the HostContext dataclass that provides minimal
+host services to plugins. It is the only interface through which
+plugins can access host functionality.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class HttpClient(Protocol):
+    """Protocol for HTTP client provided by host.
+
+    Plugins can use this for network requests (e.g., API-based engines).
+    """
+
+    def get(self, url: str, **kwargs: object) -> object:
+        """Perform an HTTP GET request."""
+        ...
+
+    def post(self, url: str, **kwargs: object) -> object:
+        """Perform an HTTP POST request."""
+        ...
+
+
+@dataclass(frozen=True)
+class HostContext:
+    """Minimal host context provided to plugins.
+
+    Contains only essential host services. No business logic.
+
+    Attributes:
+        config_dir: Directory for API keys, preferences, and configuration.
+        logger: Logger for plugin logging.
+        http_client: HTTP client for network requests.
+    """
+
+    config_dir: Path
+    logger: logging.Logger
+    http_client: HttpClient

abogen/tts_plugin/loader.py

@@ -0,0 +1,365 @@
+"""Plugin loader infrastructure for the TTS Plugin Architecture.
+
+This module provides functionality to discover, import, validate, and load
+TTS plugins. It handles both valid and invalid plugins, providing diagnostic
+messages for errors.
+
+The loader does NOT:
+- Create Engine instances (that's the plugin's create_engine() responsibility)
+- Manage plugin lifecycle (that's the Plugin Manager's responsibility)
+- Implement any TTS engine functionality
+"""
+
+from __future__ import annotations
+
+import importlib
+import re
+import sys
+import types
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable
+
+from abogen.tts_plugin.manifest import ModelManifest, PluginManifest
+
+
+# Host API version for compatibility checking
+HOST_API_VERSION = "1.0"
+
+
+@dataclass(frozen=True)
+class PluginLoadError:
+    """Diagnostic information for a failed plugin load.
+
+    Attributes:
+        plugin_id: Plugin identifier if available, otherwise directory name.
+        path: Path to the plugin directory.
+        errors: List of error messages describing what went wrong.
+    """
+
+    plugin_id: str
+    path: Path
+    errors: tuple[str, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class PluginLoadResult:
+    """Result of loading a plugin.
+
+    Attributes:
+        success: Whether the plugin loaded successfully.
+        manifest: The plugin manifest if successful.
+        model_requirements: Model requirements if successful.
+        create_engine: The create_engine function if successful.
+        module: The plugin module if successful.
+        error: Error information if failed.
+    """
+
+    success: bool
+    manifest: PluginManifest | None = None
+    model_requirements: tuple[ModelManifest, ...] | None = None
+    create_engine: Callable[..., Any] | None = None
+    module: types.ModuleType | None = None
+    error: PluginLoadError | None = None
+
+
+def _parse_api_version(version: str) -> tuple[int, int] | None:
+    """Parse an api_version string into (major, minor) tuple.
+
+    Args:
+        version: Version string in format "MAJOR.MINOR".
+
+    Returns:
+        Tuple of (major, minor) or None if invalid format.
+    """
+    match = re.match(r"^(\d+)\.(\d+)$", version)
+    if match:
+        return int(match.group(1)), int(match.group(2))
+    return None
+
+
+def _check_api_version_compatibility(plugin_version: str) -> str | None:
+    """Check if plugin api_version is compatible with host.
+
+    Architecture spec:
+    - Format: semver (MAJOR.MINOR)
+    - Compatibility: Host rejects plugin if major version differs
+    - Minor version: backward compatible, Host accepts higher minor
+
+    Args:
+        plugin_version: Plugin's api_version string.
+
+    Returns:
+        Error message if incompatible, None if compatible.
+    """
+    plugin_ver = _parse_api_version(plugin_version)
+    if plugin_ver is None:
+        return f"Invalid api_version format: '{plugin_version}'. Expected format: MAJOR.MINOR"
+
+    host_ver = _parse_api_version(HOST_API_VERSION)
+    if host_ver is None:
+        return f"Invalid host api_version format: '{HOST_API_VERSION}'"
+
+    if plugin_ver[0] != host_ver[0]:
+        return (
+            f"api_version major mismatch: plugin={plugin_ver[0]}, host={host_ver[0]}. "
+            f"Major version must match for compatibility."
+        )
+
+    return None
+
+
+def _validate_manifest(module: types.ModuleType, plugin_dir: Path) -> list[str]:
+    """Validate that a plugin module has required exports.
+
+    Args:
+        module: The imported plugin module.
+        plugin_dir: Path to the plugin directory.
+
+    Returns:
+        List of error messages (empty if valid).
+    """
+    errors: list[str] = []
+
+    # Check PLUGIN_MANIFEST
+    manifest = getattr(module, "PLUGIN_MANIFEST", None)
+    if manifest is None:
+        errors.append("Missing PLUGIN_MANIFEST export")
+    elif not isinstance(manifest, PluginManifest):
+        errors.append(
+            f"PLUGIN_MANIFEST must be a PluginManifest instance, "
+            f"got {type(manifest).__name__}"
+        )
+
+    # Check MODEL_REQUIREMENTS
+    model_reqs = getattr(module, "MODEL_REQUIREMENTS", None)
+    if model_reqs is None:
+        errors.append("Missing MODEL_REQUIREMENTS export")
+    elif not isinstance(model_reqs, list):
+        errors.append(
+            f"MODEL_REQUIREMENTS must be a list, got {type(model_reqs).__name__}"
+        )
+    else:
+        for i, req in enumerate(model_reqs):
+            if not isinstance(req, ModelManifest):
+                errors.append(
+                    f"MODEL_REQUIREMENTS[{i}] must be a ModelManifest instance, "
+                    f"got {type(req).__name__}"
+                )
+
+    # Check create_engine
+    create_engine = getattr(module, "create_engine", None)
+    if create_engine is None:
+        errors.append("Missing create_engine export")
+    elif not callable(create_engine):
+        errors.append(
+            f"create_engine must be callable, got {type(create_engine).__name__}"
+        )
+
+    return errors
+
+
+def _validate_capabilities(manifest: PluginManifest) -> list[str]:
+    """Validate plugin capabilities.
+
+    Args:
+        manifest: The plugin manifest to validate.
+
+    Returns:
+        List of error messages (empty if valid).
+    """
+    errors: list[str] = []
+
+    # Known capabilities (can be extended)
+    known_capabilities = frozenset({
+        "voice_list",
+        "preview",
+        "voice_clone",
+        "voice_blend",
+        "streaming",
+        "cancel",
+    })
+
+    for cap in manifest.capabilities:
+        if cap not in known_capabilities:
+            errors.append(f"Unknown capability: '{cap}'")
+
+    return errors
+
+
+def _validate_api_version(manifest: PluginManifest) -> list[str]:
+    """Validate api_version compatibility.
+
+    Args:
+        manifest: The plugin manifest to validate.
+
+    Returns:
+        List of error messages (empty if valid).
+    """
+    errors: list[str] = []
+    error = _check_api_version_compatibility(manifest.api_version)
+    if error:
+        errors.append(error)
+    return errors
+
+
+def load_plugin_from_dir(plugin_dir: Path) -> PluginLoadResult:
+    """Load and validate a plugin from a directory.
+
+    The plugin directory must contain an __init__.py that exports:
+    - PLUGIN_MANIFEST: PluginManifest
+    - MODEL_REQUIREMENTS: list[ModelManifest]
+    - create_engine: Callable
+
+    Args:
+        plugin_dir: Path to the plugin directory.
+
+    Returns:
+        PluginLoadResult with success status and either plugin data or error info.
+    """
+    plugin_id = plugin_dir.name
+    errors: list[str] = []
+
+    # Check if directory exists
+    if not plugin_dir.exists():
+        return PluginLoadResult(
+            success=False,
+            error=PluginLoadError(
+                plugin_id=plugin_id,
+                path=plugin_dir,
+                errors=(f"Plugin directory does not exist: {plugin_dir}",),
+            ),
+        )
+
+    # Check for __init__.py
+    init_file = plugin_dir / "__init__.py"
+    if not init_file.exists():
+        return PluginLoadResult(
+            success=False,
+            error=PluginLoadError(
+                plugin_id=plugin_id,
+                path=plugin_dir,
+                errors=("Missing __init__.py in plugin directory",),
+            ),
+        )
+
+    # Import the module
+    module_name = f"abogen.tts_plugin._loaded.{plugin_id}"
+    try:
+        # Remove from cache if already imported (for testing)
+        if module_name in sys.modules:
+            del sys.modules[module_name]
+
+        spec = importlib.util.spec_from_file_location(
+            module_name, init_file, submodule_search_locations=[]
+        )
+        if spec is None or spec.loader is None:
+            return PluginLoadResult(
+                success=False,
+                error=PluginLoadError(
+                    plugin_id=plugin_id,
+                    path=plugin_dir,
+                    errors=(f"Failed to create module spec for {init_file}",),
+                ),
+            )
+
+        module = importlib.util.module_from_spec(spec)
+        sys.modules[module_name] = module
+        spec.loader.exec_module(module)
+    except Exception as e:
+        # Clean up module from sys.modules on import failure
+        if module_name in sys.modules:
+            del sys.modules[module_name]
+        return PluginLoadResult(
+            success=False,
+            error=PluginLoadError(
+                plugin_id=plugin_id,
+                path=plugin_dir,
+                errors=(f"Failed to import plugin module: {e}",),
+            ),
+        )
+
+    # Validate manifest
+    manifest_errors = _validate_manifest(module, plugin_dir)
+    errors.extend(manifest_errors)
+
+    # If manifest is valid, perform additional validation
+    manifest = getattr(module, "PLUGIN_MANIFEST", None)
+    if isinstance(manifest, PluginManifest):
+        # Validate api_version
+        api_errors = _validate_api_version(manifest)
+        errors.extend(api_errors)
+
+        # Validate capabilities
+        cap_errors = _validate_capabilities(manifest)
+        errors.extend(cap_errors)
+
+        # Use manifest id if available
+        plugin_id = manifest.id
+
+    # Check if any errors occurred
+    if errors:
+        # Clean up module from sys.modules
+        if module_name in sys.modules:
+            del sys.modules[module_name]
+
+        return PluginLoadResult(
+            success=False,
+            error=PluginLoadError(
+                plugin_id=plugin_id,
+                path=plugin_dir,
+                errors=tuple(errors),
+            ),
+        )
+
+    # Get MODEL_REQUIREMENTS
+    model_requirements = tuple(getattr(module, "MODEL_REQUIREMENTS", []))
+    create_engine = getattr(module, "create_engine", None)
+
+    return PluginLoadResult(
+        success=True,
+        manifest=manifest,
+        model_requirements=model_requirements,
+        create_engine=create_engine,
+        module=module,
+    )
+
+
+def discover_plugins(plugin_dirs: list[Path]) -> list[PluginLoadResult]:
+    """Discover and load plugins from multiple directories.
+
+    Args:
+        plugin_dirs: List of directories to scan for plugins.
+
+    Returns:
+        List of PluginLoadResult, one per plugin directory found.
+    """
+    results: list[PluginLoadResult] = []
+
+    for plugin_dir in plugin_dirs:
+        if not plugin_dir.exists():
+            continue
+
+        # Scan for subdirectories (each is a potential plugin)
+        for item in sorted(plugin_dir.iterdir()):
+            if item.is_dir() and not item.name.startswith("."):
+                result = load_plugin_from_dir(item)
+                results.append(result)
+
+    return results
+
+
+def load_plugin(
+    plugin_dir: Path,
+) -> PluginLoadResult:
+    """Load a single plugin from a directory.
+
+    This is the main entry point for loading a plugin.
+
+    Args:
+        plugin_dir: Path to the plugin directory.
+
+    Returns:
+        PluginLoadResult with success status and either plugin data or error info.
+    """
+    return load_plugin_from_dir(plugin_dir)

abogen/tts_plugin/manifest.py

@@ -0,0 +1,189 @@
+"""Plugin manifest types for the TTS Plugin Architecture.
+
+This module contains static metadata types that describe plugins.
+These types have no dependencies and are immutable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class AudioFormatManifest:
+    """Manifest describing an audio format.
+
+    Attributes:
+        mime: MIME type of the audio.
+        extension: File extension.
+    """
+
+    mime: str
+    extension: str
+
+
+@dataclass(frozen=True)
+class EnumOption:
+    """Manifest describing an enum option for a parameter.
+
+    Attributes:
+        value: The enum value.
+        label: Human-readable label.
+    """
+
+    value: str
+    label: str
+
+
+@dataclass(frozen=True)
+class ParameterManifest:
+    """Manifest describing a synthesis parameter.
+
+    Attributes:
+        id: Parameter identifier.
+        name: Human-readable name.
+        description: Parameter description.
+        type: Parameter type ("float", "int", "string", "boolean", "enum").
+        default: Default value.
+        min: Minimum value (optional, for numeric types).
+        max: Maximum value (optional, for numeric types).
+        step: Step size (optional, for numeric types).
+        options: Available options (optional, for enum type).
+        unit: Unit of measurement (optional).
+        group: Parameter group (optional).
+    """
+
+    id: str
+    name: str
+    description: str
+    type: str
+    default: Any
+    min: float | None = None
+    max: float | None = None
+    step: float | None = None
+    options: tuple[EnumOption, ...] = field(default_factory=tuple)
+    unit: str | None = None
+    group: str | None = None
+
+
+@dataclass(frozen=True)
+class VoiceManifest:
+    """Manifest describing a voice.
+
+    Attributes:
+        id: Voice identifier.
+        name: Human-readable name.
+        tags: Voice tags (e.g., language, style).
+    """
+
+    id: str
+    name: str
+    tags: tuple[str, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class VoiceSourceManifest:
+    """Manifest describing a voice source.
+
+    Attributes:
+        id: Voice source identifier.
+        name: Human-readable name.
+        type: Source type ("list", "speaker_id", "clone", "blend", "generate", "none").
+        config: Source-specific configuration.
+    """
+
+    id: str
+    name: str
+    type: str
+    config: Any = None
+
+
+@dataclass(frozen=True)
+class EngineManifest:
+    """Manifest describing engine capabilities.
+
+    Attributes:
+        voiceSources: Available voice sources.
+        parameters: Available synthesis parameters.
+        audioFormats: Supported audio formats.
+    """
+
+    voiceSources: tuple[VoiceSourceManifest, ...] = field(default_factory=tuple)
+    parameters: tuple[ParameterManifest, ...] = field(default_factory=tuple)
+    audioFormats: tuple[AudioFormatManifest, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class GpuRequirement:
+    """Manifest describing GPU requirements.
+
+    Attributes:
+        required: Whether GPU is required.
+        type: GPU type (e.g., "cuda", "rocm").
+        memory: Required GPU memory in GB.
+    """
+
+    required: bool = False
+    type: str | None = None
+    memory: float | None = None
+
+
+@dataclass(frozen=True)
+class RequirementManifest:
+    """Manifest describing plugin requirements.
+
+    Attributes:
+        gpu: GPU requirements (optional).
+        memory: Required RAM in GB (optional).
+        internet: Whether internet is required (optional).
+    """
+
+    gpu: GpuRequirement | None = None
+    memory: float | None = None
+    internet: bool | None = None
+
+
+@dataclass(frozen=True)
+class ModelManifest:
+    """Manifest describing a model requirement.
+
+    Attributes:
+        id: Model identifier.
+        name: Human-readable name.
+        size: Model size as string (e.g., "100MB", "2GB").
+    """
+
+    id: str
+    name: str
+    size: str
+
+
+@dataclass(frozen=True)
+class PluginManifest:
+    """Main manifest for a TTS plugin.
+
+    Attributes:
+        id: Plugin identifier (unique).
+        name: Human-readable name.
+        version: Plugin version.
+        api_version: API version (semver format: MAJOR.MINOR).
+        description: Plugin description.
+        author: Plugin author.
+        capabilities: List of capability identifiers.
+        requires: Plugin requirements.
+        engine: Engine manifest.
+        voices: Optional static voice catalog. None = not declared (use VoiceLister),
+            empty tuple = explicitly no static voices, non-empty = static catalog.
+    """
+
+    id: str
+    name: str
+    version: str
+    api_version: str
+    description: str
+    author: str
+    capabilities: tuple[str, ...] = field(default_factory=tuple)
+    requires: RequirementManifest = field(default_factory=RequirementManifest)
+    engine: EngineManifest = field(default_factory=EngineManifest)
+    voices: tuple[VoiceManifest, ...] | None = None

abogen/tts_plugin/plugin.py

@@ -0,0 +1,55 @@
+"""Plugin contract for the TTS Plugin Architecture.
+
+This module defines the plugin contract that all TTS plugins must implement.
+Each plugin must export:
+- PLUGIN_MANIFEST: PluginManifest instance
+- MODEL_REQUIREMENTS: list of ModelManifest instances
+- create_engine(): Factory function that creates an Engine
+
+The create_engine() function is the entry point for plugin activation.
+It must be atomic: succeed fully or raise and clean up.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+from abogen.tts_plugin.engine import Engine
+from abogen.tts_plugin.host_context import HostContext
+from abogen.tts_plugin.types import EngineConfig
+
+
+@runtime_checkable
+class Plugin(Protocol):
+    """Protocol defining the plugin contract.
+
+    Every TTS plugin must implement this protocol by exporting:
+    - PLUGIN_MANIFEST: PluginManifest
+    - MODEL_REQUIREMENTS: list[ModelManifest]
+    - create_engine: Callable[[HostContext, Path | None, EngineConfig], Engine]
+    """
+
+    def create_engine(
+        self,
+        context: HostContext,
+        model_path: Path | None,
+        config: EngineConfig,
+    ) -> Engine:
+        """Create an engine instance.
+
+        This is the factory function that creates an Engine from a plugin.
+        It must be atomic: succeed fully or raise EngineError and clean up.
+
+        Args:
+            context: Host services (config dir, logger, http client).
+            model_path: Resolved model path, or None for cloud/no-model engines.
+            config: Engine initialization settings.
+
+        Returns:
+            A fully initialized Engine instance.
+
+        Raises:
+            EngineError: On failure. Cleans up partially created resources.
+        """
+        ...

abogen/tts_plugin/plugin_manager.py

@@ -0,0 +1,153 @@
+"""Plugin Manager
+
+Provides a simple interface for consumers to access TTS engines via the
+new Plugin Architecture. Discovers, loads, and manages plugins from the
+plugins directory.
+
+Usage:
+    from abogen.tts_plugin.plugin_manager import get_plugin_manager
+
+    manager = get_plugin_manager()
+    engine = manager.create_engine("kokoro", lang_code="a", device="cpu")
+    session = engine.create_session()
+    try:
+        result = session.synthesize("Hello world")
+    finally:
+        session.dispose()
+"""
+
+from typing import Any, Dict, List, Optional, Type
+
+from abogen.tts_plugin.engine import Engine, EngineSession
+from abogen.tts_plugin.manifest import PluginManifest
+from abogen.tts_plugin.types import AudioFormat
+
+
+class PluginManager:
+    """Manages TTS plugins and provides a simple interface for consumers."""
+
+    def __init__(self) -> None:
+        self._plugins: Dict[str, dict] = {}
+        self._engines: Dict[str, Engine] = {}
+        self._loaded = False
+
+    def discover(self, plugins_dir: str = "plugins") -> None:
+        """Discover and load all plugins from the given directory."""
+        import os
+        from pathlib import Path
+        from abogen.tts_plugin.loader import load_plugin_from_dir
+
+        self._plugins.clear()
+        self._engines.clear()
+
+        plugins_path = Path(plugins_dir)
+        if not plugins_path.exists():
+            self._loaded = True
+            return
+
+        for entry in plugins_path.iterdir():
+            if entry.is_dir() and (entry / "__init__.py").exists():
+                try:
+                    result = load_plugin_from_dir(entry)
+                    if result.success and result.manifest is not None:
+                        self._plugins[result.manifest.id] = {
+                            "manifest": result.manifest,
+                            "create_engine": result.create_engine,
+                            "module": result.module,
+                        }
+                except Exception as e:
+                    # Log error but continue with other plugins
+                    print(f"Warning: Failed to load plugin from {entry}: {e}")
+
+        self._loaded = True
+
+    def _ensure_loaded(self) -> None:
+        """Ensure plugins have been discovered."""
+        if not self._loaded:
+            self.discover()
+
+    def list_plugins(self) -> List[PluginManifest]:
+        """Return manifests for all loaded plugins."""
+        self._ensure_loaded()
+        return [info["manifest"] for info in self._plugins.values()]
+
+    def get_plugin(self, plugin_id: str) -> Optional[dict]:
+        """Get plugin info by ID."""
+        self._ensure_loaded()
+        return self._plugins.get(plugin_id)
+
+    def has_plugin(self, plugin_id: str) -> bool:
+        """Check if a plugin is loaded."""
+        self._ensure_loaded()
+        return plugin_id in self._plugins
+
+    def create_engine(self, plugin_id: str, **kwargs: Any) -> Engine:
+        """Create an engine instance for the given plugin.
+
+        Args:
+            plugin_id: The plugin identifier (e.g., "kokoro")
+            **kwargs: Arguments passed to the engine constructor
+
+        Returns:
+            An Engine instance
+
+        Raises:
+            KeyError: If plugin_id is not found
+            Exception: If engine creation fails
+        """
+        self._ensure_loaded()
+
+        if plugin_id not in self._plugins:
+            raise KeyError(f"Plugin not found: {plugin_id}")
+
+        plugin_info = self._plugins[plugin_id]
+        create_engine_func = plugin_info["create_engine"]
+
+        # Create engine using the plugin's factory
+        engine = create_engine_func(**kwargs)
+        return engine
+
+    def get_or_create_engine(self, plugin_id: str, **kwargs: Any) -> Engine:
+        """Get an existing engine or create a new one.
+
+        Engines are cached by plugin_id. If you need multiple instances
+        with different parameters, use create_engine() directly.
+        """
+        self._ensure_loaded()
+
+        cache_key = plugin_id
+        if cache_key in self._engines:
+            return self._engines[cache_key]
+
+        engine = self.create_engine(plugin_id, **kwargs)
+        self._engines[cache_key] = engine
+        return engine
+
+    def dispose_all(self) -> None:
+        """Dispose all cached engines."""
+        for engine in self._engines.values():
+            try:
+                engine.dispose()
+            except Exception:
+                pass  # dispose() should never raise
+        self._engines.clear()
+
+
+# Global singleton
+_manager: Optional[PluginManager] = None
+
+
+def get_plugin_manager() -> PluginManager:
+    """Get the global PluginManager instance."""
+    global _manager
+    if _manager is None:
+        _manager = PluginManager()
+    return _manager
+
+
+def reset_plugin_manager() -> None:
+    """Reset the global PluginManager (for testing)."""
+    global _manager
+    if _manager is not None:
+        _manager.dispose_all()
+    _manager = None

abogen/tts_plugin/types.py

@@ -0,0 +1,111 @@
+"""Core domain types for the TTS Plugin Architecture.
+
+This module contains immutable value objects that form the core domain.
+These types have zero dependencies and are used across the plugin system.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Mapping
+
+
+@dataclass(frozen=True)
+class AudioFormat:
+    """Immutable value object representing an audio format.
+
+    Attributes:
+        mime: MIME type of the audio (e.g., "audio/wav", "audio/mpeg").
+        extension: File extension (e.g., "wav", "mp3").
+    """
+
+    mime: str
+    extension: str
+
+
+@dataclass(frozen=True)
+class Duration:
+    """Immutable value object representing a time duration.
+
+    Attributes:
+        seconds: Duration in seconds.
+    """
+
+    seconds: float
+
+
+@dataclass(frozen=True)
+class VoiceSelection:
+    """Immutable value object for voice selection. Opaque to engine.
+
+    Attributes:
+        source: Voice source identifier (e.g., "builtin", "clone").
+        key: Voice key within the source.
+        payload: Optional payload for clone/blend sources.
+    """
+
+    source: str
+    key: str
+    payload: Any = None
+
+
+@dataclass(frozen=True)
+class ParameterValues:
+    """Immutable value object for synthesis parameters. Behaves like Mapping[str, Any].
+
+    Attributes:
+        values: Mapping of parameter names to their values.
+    """
+
+    values: Mapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SynthesisRequest:
+    """Immutable value object for a synthesis request.
+
+    Attributes:
+        text: Text to synthesize.
+        voice: Voice selection.
+        parameters: Synthesis parameters.
+        format: Desired audio output format.
+    """
+
+    text: str
+    voice: VoiceSelection
+    parameters: ParameterValues
+    format: AudioFormat
+
+
+@dataclass(frozen=True)
+class SynthesizedAudio:
+    """Immutable value object for synthesized audio result.
+
+    Attributes:
+        data: Raw audio bytes.
+        format: Audio format of the result.
+        duration: Duration of the audio.
+    """
+
+    data: bytes
+    format: AudioFormat
+    duration: Duration
+
+
+@dataclass(frozen=True)
+class EngineConfig:
+    """Immutable configuration of an Engine instance.
+
+    Contains parameters that define how a particular Engine instance is
+    created and that remain constant throughout the lifetime of that Engine.
+
+    Plugin implementations may ignore fields that are not applicable to them.
+
+    Attributes:
+        device: Device to use (e.g., "cpu", "cuda:0").
+        lang_code: Language code for the engine (e.g., "a" for Kokoro English).
+            Plugins that do not require a language code ignore this field.
+    """
+
+    device: str = "cpu"
+    lang_code: str = "a"

abogen/tts_plugin/utils.py

@@ -0,0 +1,235 @@
+"""TTS Plugin Architecture — direct utility functions.
+
+Provides helpers that replace the former compatibility adapter by
+calling the Plugin Manager directly.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterator
+
+import numpy as np
+
+from abogen.tts_plugin.plugin_manager import get_plugin_manager
+
+
+def get_voices(plugin_id: str) -> tuple[str, ...]:
+    """Return the voice-id tuple for *plugin_id*.
+
+    Uses the official Plugin Architecture: PluginManager → Engine → VoiceLister.
+    First checks plugin manifest for static voice catalog.
+    """
+    import logging
+    import tempfile
+    from pathlib import Path
+
+    from abogen.tts_plugin.host_context import HostContext
+    from abogen.tts_plugin.types import EngineConfig
+
+    manager = get_plugin_manager()
+    if not manager.has_plugin(plugin_id):
+        return ()
+
+    # Check manifest for static voice catalog
+    plugin_info = manager.get_plugin(plugin_id)
+    if plugin_info is not None:
+        manifest = plugin_info.get("manifest")
+        if manifest is not None and manifest.voices is not None:
+            return tuple(v.id for v in manifest.voices)
+
+    ctx = HostContext(
+        config_dir=Path(tempfile.gettempdir()),
+        logger=logging.getLogger(f"abogen.utils.{plugin_id}"),
+        http_client=type("_StubHttpClient", (), {
+            "get": staticmethod(lambda url, **kw: None),
+            "post": staticmethod(lambda url, **kw: None),
+        })(),
+    )
+
+    try:
+        engine = manager.create_engine(
+            plugin_id,
+            context=ctx,
+            model_path=None,
+            config=EngineConfig(device="cpu"),
+        )
+    except Exception:
+        return ()
+
+    try:
+        from abogen.tts_plugin.capabilities import VoiceLister
+
+        if isinstance(engine, VoiceLister):
+            manifests = engine.listVoices("builtin")
+            return tuple(v.id for v in manifests)
+        return ()
+    except Exception:
+        return ()
+    finally:
+        engine.dispose()
+
+
+def get_default_voice(plugin_id: str, fallback: str = "") -> str:
+    """Return the first voice of *plugin_id*, or *fallback*."""
+    voices = get_voices(plugin_id)
+    return voices[0] if voices else fallback
+
+
+def is_plugin_registered(plugin_id: str) -> bool:
+    """Check whether *plugin_id* is loaded by the Plugin Manager."""
+    return get_plugin_manager().has_plugin(plugin_id)
+
+
+def resolve_voice_to_plugin(spec: str, fallback: str = "kokoro") -> str:
+    """Determine which plugin owns the given voice specification.
+
+    Resolution rules:
+    1. Empty spec -> fallback
+    2. Kokoro formula (contains '*' or '+') -> "kokoro"
+    3. Exact voice-id match against loaded plugins -> plugin id
+    4. Unknown voice -> fallback
+    """
+    raw = str(spec or "").strip()
+    if not raw:
+        return fallback
+
+    if "*" in raw or "+" in raw:
+        return "kokoro"
+
+    upper = raw.upper()
+    manager = get_plugin_manager()
+
+    for manifest in manager.list_plugins():
+        for voice_source in manifest.engine.voiceSources:
+            if voice_source.type == "list" and isinstance(voice_source.config, dict):
+                try:
+                    engine = manager.create_engine(manifest.id)
+                    try:
+                        if hasattr(engine, "listVoices"):
+                            voice_manifests = engine.listVoices(voice_source.id)
+                            voice_ids = [v.id.upper() for v in voice_manifests]
+                            if upper in voice_ids:
+                                return manifest.id
+                    finally:
+                        engine.dispose()
+                except Exception:
+                    continue
+
+    return fallback
+
+
+class Pipeline:
+    """Callable wrapper around Engine / EngineSession.
+
+    Presents the same interface that old callers expect::
+
+        pipeline = create_pipeline("kokoro", lang_code="a", device="cpu")
+        for segment in pipeline(text, voice="af_nova", speed=1.0):
+            audio = segment.audio
+    """
+
+    def __init__(self, engine: Any, **engine_kwargs: Any) -> None:
+        self._engine = engine
+        self._engine_kwargs = engine_kwargs
+        self._session: Any = None
+
+    def _ensure_session(self) -> Any:
+        if self._session is None:
+            self._session = self._engine.createSession()
+        return self._session
+
+    def __call__(
+        self,
+        text: str,
+        voice: str = "default",
+        speed: float = 1.0,
+        split_pattern: str | None = None,
+        **kwargs: Any,
+    ) -> Iterator[Any]:
+        from abogen.tts_plugin.types import (
+            AudioFormat,
+            ParameterValues,
+            SynthesisRequest,
+            VoiceSelection,
+        )
+
+        session = self._ensure_session()
+
+        params: dict[str, Any] = {"speed": speed}
+        if split_pattern is not None:
+            params["split_pattern"] = split_pattern
+        params.update(kwargs)
+
+        request = SynthesisRequest(
+            text=text,
+            voice=VoiceSelection(source="builtin", key=voice),
+            parameters=ParameterValues(values=params),
+            format=AudioFormat(mime="audio/wav", extension="wav"),
+        )
+
+        result = session.synthesize(request)
+        audio_array = np.frombuffer(result.data, dtype=np.float32)
+
+        from dataclasses import dataclass
+
+        @dataclass
+        class Segment:
+            graphemes: str
+            audio: np.ndarray
+
+        yield Segment(graphemes=text, audio=audio_array)
+
+    def dispose(self) -> None:
+        if self._session is not None:
+            try:
+                self._session.dispose()
+            except Exception:
+                pass
+            self._session = None
+
+    def __del__(self) -> None:
+        self.dispose()
+
+
+def create_pipeline(
+    plugin_id: str,
+    *,
+    lang_code: str = "a",
+    device: str = "cpu",
+) -> Pipeline:
+    """Create a callable TTS pipeline via the Plugin Architecture.
+
+    Builds a proper HostContext and EngineConfig, then delegates to the
+    PluginManager to create the engine. Returns a :class:`Pipeline` whose
+    ``__call__`` interface matches the callable protocol used by consumers.
+
+    Args:
+        plugin_id: Plugin identifier (e.g., "kokoro", "supertonic").
+        lang_code: Language code for the engine.
+        device: Device to use (e.g., "cpu", "cuda:0").
+
+    Returns:
+        A callable Pipeline instance.
+    """
+    import logging
+    import tempfile
+    from pathlib import Path
+
+    from abogen.tts_plugin.host_context import HostContext
+    from abogen.tts_plugin.types import EngineConfig
+
+    manager = get_plugin_manager()
+
+    ctx = HostContext(
+        config_dir=Path(tempfile.gettempdir()),
+        logger=logging.getLogger(f"abogen.pipeline.{plugin_id}"),
+        http_client=type("_StubHttpClient", (), {
+            "get": staticmethod(lambda url, **kw: None),
+            "post": staticmethod(lambda url, **kw: None),
+        })(),
+    )
+
+    config = EngineConfig(device=device, lang_code=lang_code)
+
+    engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config)
+    return Pipeline(engine)

abogen/utils.py

@@ -529,21 +529,20 @@ def prevent_sleep_end():
                 _sleep_procs[system] = None
 
 
-def load_numpy_kpipeline():
-    import numpy as np
-    from kokoro import KPipeline  # type: ignore[import-not-found]
-
-    return np, KPipeline
-
-
 class LoadPipelineThread(Thread):
-    def __init__(self, callback):
+    def __init__(self, callback, lang_code="a", device="cpu"):
         super().__init__()
         self.callback = callback
+        self.lang_code = lang_code
+        self.device = device
 
     def run(self):
         try:
-            np_module, kpipeline_class = load_numpy_kpipeline()
+            from abogen.tts_plugin.utils import create_pipeline
-            self.callback(np_module, kpipeline_class, None)
+
+            backend = create_pipeline(
+                "kokoro", lang_code=self.lang_code, device=self.device
+            )
+            self.callback(backend, None)
         except Exception as e:
-            self.callback(None, None, str(e))
+            self.callback(None, str(e))

abogen/voice_cache.py

@@ -17,7 +17,7 @@ if LocalEntryNotFoundError is None:  # pragma: no cover - fallback for tests
         pass
 
 
-from abogen.constants import VOICES_INTERNAL
+from abogen.tts_plugin.utils import get_voices
 
 _CACHE_LOCK = threading.Lock()
 _CACHED_VOICES: Set[str] = set()
@@ -26,8 +26,9 @@ _BOOTSTRAPPED = False
 
 
 def _normalize_targets(voices: Optional[Iterable[str]]) -> Set[str]:
+    kokoro_voices = get_voices("kokoro")
     if not voices:
-        return set(VOICES_INTERNAL)
+        return set(kokoro_voices)
     normalized: Set[str] = set()
     for voice in voices:
         if not voice:
@@ -35,7 +36,7 @@ def _normalize_targets(voices: Optional[Iterable[str]]) -> Set[str]:
         voice_id = str(voice).strip()
         if not voice_id:
             continue
-        if voice_id in VOICES_INTERNAL:
+        if voice_id in kokoro_voices:
             normalized.add(voice_id)
     return normalized
 
@@ -143,3 +144,11 @@ def _ensure_single_voice_asset(
 
     hf_hub_download(resume_download=True, **common_kwargs)
     return True
+
+
+def clear_voice_cache() -> None:
+    """Clear the in‑process voice cache (used during shutdown)."""
+    with _CACHE_LOCK:
+        _CACHED_VOICES.clear()
+    global _BOOTSTRAPPED
+    _BOOTSTRAPPED = False