Compare commits

..
5 Commits
Author SHA1 Message Date
Deniz Şafak aaa6ac112b fix(audio_helpers): update opus bitrate to 128kbps in ffmpeg command 2026-08-20 21:32:07 +03:00
Deniz Şafak 11274ad6bf fix(pyqt): migrate GUI to Language enums, fix subtitle mode gating
Complete the Language enum unification (commit 0dc491e) for the PyQt
GUI, which was left resolving languages from kokoro letter codes
(voice[0]) while domain and WebUI already used Language enums.

- selected_lang is now always a Language enum: voices resolve via
  language_for_voice_id(), profiles via resolve_profile_language()
- kokoro letter->Language mapping stays in the engine (new public
  language_for_code()); the Language.from_code() domain shim is removed
- legacy profile files with letter codes ("a", "e", ...) are tolerated
  only at the profile read boundary; new profiles save ISO codes
- conversion.py letter comparisons replaced with enum comparisons, and
  SynthParams lang_code= -> language= (would TypeError at runtime)
- subtitle dropdown enabled for all languages; word-count and
  highlighting modes restricted to English with auto-switch to Sentence
- tests: engine mapping + profile language resolution added; 1566 pass
2026-08-20 21:28:30 +03:00
Deniz Şafak f340b976db feat: added colorful logging using rich, improved startup times by lazy-loading spacy, some other fixes 2026-08-20 20:52:08 +03:00
Deniz Şafak 9da15aefa4 Merge pull request #192 from Sanjays2402/fix/preserve-user-metadata
Preserve user-entered book metadata after extraction
2026-08-20 09:37:34 -07:00
Sanjay Santhanam 7340d52ebb fix(webui): preserve user-entered book metadata
Apply the book form after extraction-backed pending job creation so explicit title and author values take precedence over fallback metadata. Add a focused regression test for the upload path.
2026-07-18 07:32:41 -07:00
27 changed files with 585 additions and 217 deletions
+3
View File
@@ -40,3 +40,6 @@ test_assets/
dev_notes/ dev_notes/
.claude/ .claude/
.coverage .coverage
# CodeGraph index (local, machine-specific)
.codegraph/
+6 -5
View File
@@ -70,11 +70,12 @@ SUPPORTED_INPUT_FORMATS = [
"vtt", "vtt",
] ]
# Supported languages for subtitle generation # Supported languages for subtitle generation.
# Currently, only English (EN_US, EN_GB) are supported for subtitle generation. # All languages are supported: only English emits per-word timestamped tokens
# This is because tokens that contain timestamps are not generated for other languages in the Kokoro pipeline. # in the Kokoro pipeline, but other languages fall back to segment-level fake
# Please refer to: https://github.com/hexgrad/kokoro/blob/6d87f4ae7abc2d14dbc4b3ef2e5f19852e861ac2/kokoro/pipeline.py # tokens (see abogen.domain.tokens.FakeToken), so subtitles are still
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = [Language.EN_US, Language.EN_GB] # generated at segment granularity.
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = list(Language)
# Voice and sample text mapping # Voice and sample text mapping
SAMPLE_VOICE_TEXTS = { SAMPLE_VOICE_TEXTS = {
+1 -1
View File
@@ -33,7 +33,7 @@ def build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str]
if fmt == "mp3": if fmt == "mp3":
base += ["-c:a", "libmp3lame", "-qscale:a", "2"] base += ["-c:a", "libmp3lame", "-qscale:a", "2"]
elif fmt == "opus": elif fmt == "opus":
base += ["-c:a", "libopus", "-b:a", "24000"] base += ["-c:a", "libopus", "-b:a", "128000"]
elif fmt == "m4b": elif fmt == "m4b":
base += ["-c:a", "aac", "-q:a", "2", "-movflags", "+faststart+use_metadata_tags"] base += ["-c:a", "aac", "-q:a", "2", "-movflags", "+faststart+use_metadata_tags"]
else: else:
+2 -4
View File
@@ -112,12 +112,10 @@ def spacy_pre_tts_segmentation(
def _to_language_enum(lang_code: Any) -> Language: def _to_language_enum(lang_code: Any) -> Language:
"""Convert lang_code to Language enum.""" """Convert lang_code to Language enum (ISO code or Language enum)."""
if isinstance(lang_code, Language):
return lang_code
try: try:
return Language.from_str(str(lang_code)) return Language.from_str(str(lang_code))
except (ValueError, AttributeError): except ValueError:
return Language.EN_US return Language.EN_US
+6 -2
View File
@@ -212,8 +212,12 @@ class Language(str, Enum):
@property @property
def supports_subtitle_tokens(self) -> bool: def supports_subtitle_tokens(self) -> bool:
"""True if this language generates timestamped tokens for subtitles.""" """True if this language supports subtitle generation.
return self in (self.EN_US, self.EN_GB)
All languages are supported: languages without per-word timestamped
tokens fall back to segment-level fake tokens in the pipeline.
"""
return True
@classmethod @classmethod
def from_str(cls, value: str) -> Language: def from_str(cls, value: str) -> Language:
+1 -1
View File
@@ -16,7 +16,7 @@ def get_split_pattern(language: Language, subtitle_mode: str) -> str:
"""Get the appropriate split pattern based on language and subtitle mode. """Get the appropriate split pattern based on language and subtitle mode.
Args: Args:
language: Language enum value. language: Language enum value, ISO code, or kokoro letter code.
subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.) subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.)
Returns: Returns:
+17 -5
View File
@@ -9,15 +9,26 @@ from collections import Counter
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple
try: # pragma: no cover - fallback when spaCy not available during tests
import spacy # type: ignore[import-not-found]
except Exception: # pragma: no cover - spaCy optional during runtime bootstrap
spacy = None
_Language = Any # type: ignore[misc,assignment] _Language = Any # type: ignore[misc,assignment]
Doc = Any # type: ignore[misc,assignment] Doc = Any # type: ignore[misc,assignment]
Span = Any # type: ignore[misc,assignment] Span = Any # type: ignore[misc,assignment]
_SPACY: Any = None
_SPACY_LOADED = False
def _get_spacy() -> Any:
"""Import spaCy lazily (it pulls in torch/thinc, ~2s at startup)."""
global _SPACY, _SPACY_LOADED
if not _SPACY_LOADED:
_SPACY_LOADED = True
try: # pragma: no cover - fallback when spaCy not available during tests
import spacy # type: ignore[import-not-found]
except Exception: # pragma: no cover - spaCy optional during runtime bootstrap
spacy = None
_SPACY = spacy
return _SPACY
_TITLE_PREFIXES = ( _TITLE_PREFIXES = (
"mr", "mr",
@@ -167,6 +178,7 @@ def _resolve_model_name(language: str) -> str:
def _load_model(language: str) -> Any: def _load_model(language: str) -> Any:
spacy = _get_spacy()
if spacy is None: if spacy is None:
raise EntityModelError( raise EntityModelError(
"spaCy is not available. Install spaCy to enable entity extraction." "spaCy is not available. Install spaCy to enable entity extraction."
+17 -5
View File
@@ -5,10 +5,21 @@ import re
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
try: # pragma: no cover - optional dependency _SPACY: Any = None
import spacy # type: ignore _SPACY_LOADED = False
except Exception: # pragma: no cover - spaCy may be unavailable in minimal environments
spacy = None
def _get_spacy() -> Any:
"""Import spaCy lazily (it pulls in torch/thinc, ~2s at startup)."""
global _SPACY, _SPACY_LOADED
if not _SPACY_LOADED:
_SPACY_LOADED = True
try: # pragma: no cover - optional dependency
import spacy # type: ignore
except Exception: # pragma: no cover - spaCy may be unavailable in minimal environments
spacy = None
_SPACY = spacy
return _SPACY
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -184,6 +195,7 @@ def _build_replacement_sentence(
def _load_spacy(language: str) -> Any: def _load_spacy(language: str) -> Any:
spacy = _get_spacy()
if spacy is None: if spacy is None:
return None return None
@@ -221,7 +233,7 @@ def extract_heteronym_overrides(
if not lang.startswith("en"): if not lang.startswith("en"):
return [] return []
if spacy is None: if _get_spacy() is None:
return [] return []
nlp = _load_spacy(lang) nlp = _load_spacy(lang)
+3 -3
View File
@@ -45,9 +45,9 @@ import urllib.parse
import textwrap import textwrap
# Setup logging # Setup logging
logging.basicConfig( from abogen.utils import setup_console_logging
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
) setup_console_logging()
_HTML_TAG_PATTERN = re.compile(r"<[^>]+>") _HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
_LEADING_DASH_PATTERN = re.compile(r"^\s*[-–—]\s*") _LEADING_DASH_PATTERN = re.compile(r"^\s*[-–—]\s*")
+5 -4
View File
@@ -22,6 +22,7 @@ from abogen.constants import (
) )
from abogen.infrastructure.subtitle_writer import make_subtitle_writer, resolve_subtitle_format from abogen.infrastructure.subtitle_writer import make_subtitle_writer, resolve_subtitle_format
from abogen.domain.split_pattern import get_split_pattern from abogen.domain.split_pattern import get_split_pattern
from abogen.domain.enums import Language
from abogen.domain.subtitle_processor import ( from abogen.domain.subtitle_processor import (
parse_subtitle_file, parse_subtitle_file,
process_subtitle_entries, process_subtitle_entries,
@@ -872,12 +873,12 @@ class ConversionThread(QThread):
) )
spacy_sentences = None spacy_sentences = None
active_split_pattern = self.split_pattern active_split_pattern = self.split_pattern
spacing_pattern = r"\s*" if self.lang_code in ["z", "j"] else r"\s+" spacing_pattern = r"\s*" if self.lang_code in (Language.JA, Language.ZH) else r"\s+"
# Pre-load spaCy model for English if it will be needed for subtitle generation # Pre-load spaCy model for English if it will be needed for subtitle generation
if ( if (
use_spacy use_spacy
and self.lang_code in ["a", "b"] and self.lang_code in (Language.EN_US, Language.EN_GB)
and self.subtitle_mode in ["Sentence", "Sentence + Comma"] and self.subtitle_mode in ["Sentence", "Sentence + Comma"]
): ):
from abogen.spacy_utils import get_spacy_model from abogen.spacy_utils import get_spacy_model
@@ -894,7 +895,7 @@ class ConversionThread(QThread):
) )
) )
if use_spacy and self.lang_code not in ["a", "b"]: if use_spacy and self.lang_code not in (Language.EN_US, Language.EN_GB):
# Non-English: use spaCy for pre-TTS segmentation # Non-English: use spaCy for pre-TTS segmentation
self.log_updated.emit( self.log_updated.emit(
("\nUsing spaCy for sentence segmentation (pre-TTS)...", "grey") ("\nUsing spaCy for sentence segmentation (pre-TTS)...", "grey")
@@ -1011,7 +1012,7 @@ class ConversionThread(QThread):
audio_sink=merged_sink if merge_chapters_at_end else None, audio_sink=merged_sink if merge_chapters_at_end else None,
subtitle_mode=self.subtitle_mode, subtitle_mode=self.subtitle_mode,
max_subtitle_words=self.max_subtitle_words, max_subtitle_words=self.max_subtitle_words,
lang_code=self.lang_code, language=self.lang_code,
use_spacy_segmentation=getattr(self, "use_spacy_segmentation", False), use_spacy_segmentation=getattr(self, "use_spacy_segmentation", False),
) )
+49 -41
View File
@@ -75,6 +75,7 @@ from abogen.domain.text_utils import calculate_text_length
from abogen.pyqt.conversion import ConversionThread, VoicePreviewThread, PlayAudioThread, ChapterOptionsDialog, TimestampDetectionDialog from abogen.pyqt.conversion import ConversionThread, VoicePreviewThread, PlayAudioThread, ChapterOptionsDialog, TimestampDetectionDialog
from abogen.pyqt.book_handler import HandlerDialog from abogen.pyqt.book_handler import HandlerDialog
from abogen.domain.enums import Language
from abogen.constants import ( from abogen.constants import (
PROGRAM_NAME, PROGRAM_NAME,
VERSION, VERSION,
@@ -88,8 +89,9 @@ from abogen.constants import (
from abogen.tts_plugin.utils import get_voices from abogen.tts_plugin.utils import get_voices
import threading import threading
from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles, resolve_profile_language
from abogen.domain.settings_core import all_settings_defaults from abogen.domain.settings_core import all_settings_defaults
from plugins.kokoro.engine import language_for_code, language_for_voice_id
# Module-level default cache for use outside __init__ # Module-level default cache for use outside __init__
_DEFAULTS = all_settings_defaults() _DEFAULTS = all_settings_defaults()
@@ -397,11 +399,7 @@ class InputBox(QLabel):
# Re-enable subtitle and replace newlines controls when cleared # Re-enable subtitle and replace newlines controls when cleared
window = self.window() window = self.window()
if hasattr(window, "subtitle_combo"): if hasattr(window, "subtitle_combo"):
# Only enable if language supports it window.subtitle_combo.setEnabled(True)
current_lang = getattr(window, "selected_lang", "a")
window.subtitle_combo.setEnabled(
current_lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
)
if hasattr(window, "replace_newlines_combo"): if hasattr(window, "replace_newlines_combo"):
window.replace_newlines_combo.setEnabled(True) window.replace_newlines_combo.setEnabled(True)
@@ -941,7 +939,7 @@ class abogen(QWidget):
self.selected_lang = None self.selected_lang = None
else: else:
self.selected_voice = self.config.get("selected_voice", _d["selected_voice"]) self.selected_voice = self.config.get("selected_voice", _d["selected_voice"])
self.selected_lang = self.selected_voice[0] if self.selected_voice else None self.selected_lang = language_for_voice_id(self.selected_voice)
self.is_converting = False self.is_converting = False
self.subtitle_mode = self.config.get("subtitle_mode", _d["subtitle_mode"]) self.subtitle_mode = self.config.get("subtitle_mode", _d["subtitle_mode"])
self.max_subtitle_words = self.config.get("max_subtitle_words", _d["max_subtitle_words"]) self.max_subtitle_words = self.config.get("max_subtitle_words", _d["max_subtitle_words"])
@@ -988,7 +986,12 @@ class abogen(QWidget):
self.queued_items = [] self.queued_items = []
self.current_queue_index = 0 self.current_queue_index = 0
self.initUI() from abogen.utils import timed_log
import logging
_startup_log = logging.getLogger("abogen.startup")
with timed_log("GUI initUI (widget building)", logger=_startup_log):
self.initUI()
self.speed_slider.setValue(int(self.config.get("speed", _d["speed"]) * 100)) self.speed_slider.setValue(int(self.config.get("speed", _d["speed"]) * 100))
self.update_speed_label() self.update_speed_label()
# Set initial selection: prefer profile, else voice # Set initial selection: prefer profile, else voice
@@ -1003,13 +1006,16 @@ class abogen(QWidget):
if self.selected_profile_name: if self.selected_profile_name:
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {}) with timed_log("voice profile load", logger=_startup_log):
entry = load_profiles().get(self.selected_profile_name, {})
if isinstance(entry, dict): if isinstance(entry, dict):
self.mixed_voice_state = entry.get("voices", []) self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language") self.selected_lang = resolve_profile_language(entry)
else: else:
self.mixed_voice_state = entry self.mixed_voice_state = entry
self.selected_lang = entry[0][0] if entry and entry[0] else None self.selected_lang = (
language_for_voice_id(entry[0]) if entry and entry[0] else Language.EN_US
)
if self.save_option == "Choose output folder" and self.selected_output_folder: if self.save_option == "Choose output folder" and self.selected_output_folder:
self.save_path_label.setText(self.selected_output_folder) self.save_path_label.setText(self.selected_output_folder)
self.save_path_row_widget.show() self.save_path_row_widget.show()
@@ -1175,6 +1181,7 @@ class abogen(QWidget):
"Sentence + Comma: Subtitles will be generated for each sentence and comma.\n" "Sentence + Comma: Subtitles will be generated for each sentence and comma.\n"
"Sentence + Highlighting: Subtitles with word-by-word karaoke highlighting.\n" "Sentence + Highlighting: Subtitles with word-by-word karaoke highlighting.\n"
"1+ word: Subtitles will be generated for each word(s).\n\n" "1+ word: Subtitles will be generated for each word(s).\n\n"
"Word-count and highlighting modes are only available for English.\n"
"Supported languages for subtitle generation:\n" "Supported languages for subtitle generation:\n"
+ "\n".join( + "\n".join(
f'"{lang}" => {LANGUAGE_DESCRIPTIONS.get(lang, lang)}' f'"{lang}" => {LANGUAGE_DESCRIPTIONS.get(lang, lang)}'
@@ -1753,8 +1760,9 @@ class abogen(QWidget):
def update_subtitle_options_availability(self): def update_subtitle_options_availability(self):
""" """
Update the enabled state of subtitle options based on the selected language. Update the enabled state of subtitle options based on the selected
For non-English languages, only sentence-based and line-based modes are supported. language and input type. Subtitle generation works for every language,
but word-count and highlighting modes are only available for English.
""" """
# Check if current file is a subtitle file # Check if current file is a subtitle file
is_subtitle_input = False is_subtitle_input = False
@@ -1763,16 +1771,14 @@ class abogen(QWidget):
): ):
is_subtitle_input = True is_subtitle_input = True
if self.selected_lang not in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION:
self.subtitle_combo.setEnabled(False)
self.subtitle_format_combo.setEnabled(False)
return
# Only enable subtitle_combo if it's NOT a subtitle input # Only enable subtitle_combo if it's NOT a subtitle input
self.subtitle_combo.setEnabled(not is_subtitle_input) self.subtitle_combo.setEnabled(not is_subtitle_input)
self.subtitle_format_combo.setEnabled(True) self.subtitle_format_combo.setEnabled(True)
is_english = self.selected_lang in ["a", "b"] is_english = self.selected_lang in (
Language.EN_US,
Language.EN_GB,
)
# Items to keep enabled for non-English # Items to keep enabled for non-English
allowed_modes = ["Disabled", "Line", "Sentence", "Sentence + Comma"] allowed_modes = ["Disabled", "Line", "Sentence", "Sentence + Comma"]
@@ -1787,10 +1793,7 @@ class abogen(QWidget):
if is_english: if is_english:
item.setEnabled(True) item.setEnabled(True)
else: else:
if text in allowed_modes: item.setEnabled(text in allowed_modes)
item.setEnabled(True)
else:
item.setEnabled(False)
# If current selection is disabled, switch to a valid one # If current selection is disabled, switch to a valid one
current_text = self.subtitle_combo.currentText() current_text = self.subtitle_combo.currentText()
@@ -1809,7 +1812,7 @@ class abogen(QWidget):
def on_voice_changed(self, index): def on_voice_changed(self, index):
voice = self.voice_combo.itemData(index) voice = self.voice_combo.itemData(index)
self.selected_voice, self.selected_lang = voice, voice[0] self.selected_voice, self.selected_lang = voice, language_for_voice_id(voice)
self.config["selected_voice"] = voice self.config["selected_voice"] = voice
save_config(self.config) save_config(self.config)
# Enable/disable subtitle options based on language # Enable/disable subtitle options based on language
@@ -1826,10 +1829,12 @@ class abogen(QWidget):
# set mixed voices and language # set mixed voices and language
if isinstance(entry, dict): if isinstance(entry, dict):
self.mixed_voice_state = entry.get("voices", []) self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language") self.selected_lang = resolve_profile_language(entry)
else: else:
self.mixed_voice_state = entry self.mixed_voice_state = entry
self.selected_lang = entry[0][0] if entry and entry[0] else None self.selected_lang = (
language_for_voice_id(entry[0]) if entry and entry[0] else Language.EN_US
)
self.selected_voice = None self.selected_voice = None
self.config["selected_profile_name"] = pname self.config["selected_profile_name"] = pname
self.config.pop("selected_voice", None) self.config.pop("selected_voice", None)
@@ -1839,7 +1844,7 @@ class abogen(QWidget):
else: else:
self.mixed_voice_state = None self.mixed_voice_state = None
self.selected_profile_name = None self.selected_profile_name = None
self.selected_voice, self.selected_lang = data, data[0] self.selected_voice, self.selected_lang = data, language_for_voice_id(data)
self.config["selected_voice"] = data self.config["selected_voice"] = data
if "selected_profile_name" in self.config: if "selected_profile_name" in self.config:
del self.config["selected_profile_name"] del self.config["selected_profile_name"]
@@ -1850,8 +1855,9 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles
entry = load_profiles().get(profile_name, {}) entry = load_profiles().get(profile_name, {})
lang = entry.get("language") if isinstance(entry, dict) else None enable = (
enable = lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION resolve_profile_language(entry) in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
)
self.subtitle_combo.setEnabled(enable) self.subtitle_combo.setEnabled(enable)
self.subtitle_format_combo.setEnabled(enable) self.subtitle_format_combo.setEnabled(enable)
@@ -2233,18 +2239,18 @@ class abogen(QWidget):
else: else:
return self.selected_voice return self.selected_voice
def get_selected_lang(self, voice_formula) -> str: def get_selected_lang(self, voice_formula) -> Language:
if self.selected_profile_name: if self.selected_profile_name:
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {}) entry = load_profiles().get(self.selected_profile_name, {})
selected_lang = entry.get("language") selected_lang = resolve_profile_language(entry)
else: else:
selected_lang = self.selected_voice[0] if self.selected_voice else None selected_lang = language_for_voice_id(self.selected_voice)
# fallback: extract from formula if missing # fallback: extract from formula if missing
if not selected_lang: if not selected_lang:
m = re.search(r"\b([a-z])", voice_formula) m = re.search(r"\b([a-z])", voice_formula)
selected_lang = m.group(1) if m else None selected_lang = language_for_code(m.group(1)) if m else Language.EN_US
return selected_lang return selected_lang
def get_actual_subtitle_mode(self) -> str: def get_actual_subtitle_mode(self) -> str:
@@ -2420,7 +2426,7 @@ class abogen(QWidget):
self.update_log((gpu_msg, gpu_ok)) self.update_log((gpu_msg, gpu_ok))
self.update_log("Loading modules...") self.update_log("Loading modules...")
lang_code = self.selected_lang or "a" lang_code = self.selected_lang or Language.EN_US
load_thread = LoadPipelineThread( load_thread = LoadPipelineThread(
pipeline_loaded_callback, lang_code=lang_code, use_gpu=gpu_ok pipeline_loaded_callback, lang_code=lang_code, use_gpu=gpu_ok
) )
@@ -2751,12 +2757,12 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {}) entry = load_profiles().get(self.selected_profile_name, {})
lang_to_cache = entry.get("language") lang_to_cache = resolve_profile_language(entry)
else: else:
lang_to_cache = self.selected_lang lang_to_cache = self.selected_lang
if not lang_to_cache and self.mixed_voice_state: if not lang_to_cache and self.mixed_voice_state:
lang_to_cache = ( lang_to_cache = (
self.mixed_voice_state[0][0][0] language_for_voice_id(self.mixed_voice_state[0][0])
if self.mixed_voice_state and self.mixed_voice_state[0][0] if self.mixed_voice_state and self.mixed_voice_state[0][0]
else None else None
) )
@@ -2860,7 +2866,7 @@ class abogen(QWidget):
) )
self.loading_movie.start() self.loading_movie.start()
lang = self.selected_lang or "a" lang = self.selected_lang or Language.EN_US
load_thread = LoadPipelineThread( load_thread = LoadPipelineThread(
self._on_pipeline_loaded_for_preview, lang_code=lang, use_gpu=self.gpu_ok self._on_pipeline_loaded_for_preview, lang_code=lang, use_gpu=self.gpu_ok
) )
@@ -2892,17 +2898,17 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {}) entry = load_profiles().get(self.selected_profile_name, {})
lang = entry.get("language") lang = resolve_profile_language(entry)
else: else:
lang = self.selected_lang lang = self.selected_lang
if not lang and self.mixed_voice_state: if not lang and self.mixed_voice_state:
lang = ( lang = (
self.mixed_voice_state[0][0][0] language_for_voice_id(self.mixed_voice_state[0][0])
if self.mixed_voice_state and self.mixed_voice_state[0][0] if self.mixed_voice_state and self.mixed_voice_state[0][0]
else None else None
) )
else: else:
lang = self.selected_voice[0] lang = language_for_voice_id(self.selected_voice)
voice = self.selected_voice voice = self.selected_voice
# use same gpu/cpu logic as in conversion # use same gpu/cpu logic as in conversion
@@ -3948,7 +3954,9 @@ Categories=AudioVideo;Audio;Utility;
initial_state = entry.get("voices", []) initial_state = entry.get("voices", [])
else: else:
initial_state = entry initial_state = entry
self.selected_lang = entry[0][0] if entry and entry[0] else None self.selected_lang = (
language_for_voice_id(entry[0]) if entry and entry[0] else Language.EN_US
)
dialog = VoiceFormulaDialog( dialog = VoiceFormulaDialog(
self, initial_state=initial_state, selected_profile=selected_profile self, initial_state=initial_state, selected_profile=selected_profile
) )
+89 -74
View File
@@ -1,3 +1,4 @@
import logging
import os import os
import sys import sys
import platform import platform
@@ -6,103 +7,113 @@ import platform
from abogen import shutdown # noqa: F401 from abogen import shutdown # noqa: F401
shutdown.register_shutdown() shutdown.register_shutdown()
from abogen.utils import get_resource_path, setup_console_logging, timed_log # noqa: E402
_log = logging.getLogger("abogen.startup")
setup_console_logging()
# Fix PyTorch DLL loading issue ([WinError 1114]) on Windows before importing PyQt6 # Fix PyTorch DLL loading issue ([WinError 1114]) on Windows before importing PyQt6
if platform.system() == "Windows": if platform.system() == "Windows":
import ctypes with timed_log("PyTorch DLLs (Windows)", logger=_log):
from importlib.util import find_spec import ctypes
from importlib.util import find_spec
try: try:
if ( if (
(spec := find_spec("torch")) (spec := find_spec("torch"))
and spec.origin and spec.origin
and os.path.exists( and os.path.exists(
dll_path := os.path.join(os.path.dirname(spec.origin), "lib", "c10.dll") dll_path := os.path.join(os.path.dirname(spec.origin), "lib", "c10.dll")
) )
): ):
ctypes.CDLL(os.path.normpath(dll_path)) ctypes.CDLL(os.path.normpath(dll_path))
except Exception: except Exception:
pass pass
# Qt platform plugin detection (fixes #59) # Qt platform plugin detection (fixes #59)
try: with timed_log("Qt platform plugin detection", logger=_log):
from PyQt6.QtCore import QLibraryInfo try:
from PyQt6.QtCore import QLibraryInfo
# Get the path to the plugins directory # Get the path to the plugins directory
plugins = QLibraryInfo.path(QLibraryInfo.LibraryPath.PluginsPath) plugins = QLibraryInfo.path(QLibraryInfo.LibraryPath.PluginsPath)
# Normalize path to use the OS-native separators and absolute path # Normalize path to use the OS-native separators and absolute path
platform_dir = os.path.normpath(os.path.join(plugins, "platforms")) platform_dir = os.path.normpath(os.path.join(plugins, "platforms"))
# Ensure we work with an absolute path for clarity # Ensure we work with an absolute path for clarity
platform_dir = os.path.abspath(platform_dir) platform_dir = os.path.abspath(platform_dir)
if os.path.isdir(platform_dir): if os.path.isdir(platform_dir):
os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = platform_dir os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = platform_dir
print("QT_QPA_PLATFORM_PLUGIN_PATH set to:", platform_dir) _log.info("QT_QPA_PLATFORM_PLUGIN_PATH set to: %s", platform_dir)
else: else:
print("PyQt6 platform plugins not found at", platform_dir) _log.warning("PyQt6 platform plugins not found at %s", platform_dir)
except ImportError: except ImportError:
print("PyQt6 not installed.") _log.warning("PyQt6 not installed.")
from abogen.utils import get_resource_path
# Pre-load "libxcb-cursor" on Linux (fixes #101) # Pre-load "libxcb-cursor" on Linux (fixes #101)
if platform.system() == "Linux": if platform.system() == "Linux":
arch = platform.machine().lower() with timed_log("libxcb-cursor preload (Linux)", logger=_log):
lib_filename = {"x86_64": "libxcb-cursor-amd64.so.0", "amd64": "libxcb-cursor-amd64.so.0", "aarch64": "libxcb-cursor-arm64.so.0", "arm64": "libxcb-cursor-arm64.so.0"}.get(arch) arch = platform.machine().lower()
if lib_filename: lib_filename = {"x86_64": "libxcb-cursor-amd64.so.0", "amd64": "libxcb-cursor-amd64.so.0", "aarch64": "libxcb-cursor-arm64.so.0", "arm64": "libxcb-cursor-arm64.so.0"}.get(arch)
import ctypes if lib_filename:
try: import ctypes
# Try to load the system libxcb-cursor.so.0 first try:
ctypes.CDLL('libxcb-cursor.so.0', mode=ctypes.RTLD_GLOBAL) # Try to load the system libxcb-cursor.so.0 first
except OSError: ctypes.CDLL('libxcb-cursor.so.0', mode=ctypes.RTLD_GLOBAL)
# System lib not available, load the bundled version except OSError:
lib_path = get_resource_path('abogen.libs', lib_filename) # System lib not available, load the bundled version
if lib_path: lib_path = get_resource_path('abogen.libs', lib_filename)
try: if lib_path:
ctypes.CDLL(lib_path, mode=ctypes.RTLD_GLOBAL) try:
except OSError: ctypes.CDLL(lib_path, mode=ctypes.RTLD_GLOBAL)
# If it fails (e.g. wrong glibc version on very old systems), except OSError:
# we simply ignore it and hope the system has the library. # If it fails (e.g. wrong glibc version on very old systems),
pass # we simply ignore it and hope the system has the library.
pass
# Set application ID for Windows taskbar icon # Set application ID for Windows taskbar icon
if platform.system() == "Windows": if platform.system() == "Windows":
try: with timed_log("Windows AppUserModelID", logger=_log):
from abogen.constants import PROGRAM_NAME, VERSION try:
import ctypes from abogen.constants import PROGRAM_NAME, VERSION
import ctypes
app_id = f"{PROGRAM_NAME}.{VERSION}" app_id = f"{PROGRAM_NAME}.{VERSION}"
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id) ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
except Exception as e: except Exception as e:
print("Warning: failed to set AppUserModelID:", e) _log.warning("Failed to set AppUserModelID: %s", e)
from PyQt6.QtWidgets import QApplication with timed_log("PyQt6 imports", logger=_log):
from PyQt6.QtGui import QIcon from PyQt6.QtWidgets import QApplication
from PyQt6.QtCore import ( from PyQt6.QtGui import QIcon
QLibraryInfo, from PyQt6.QtCore import (
qInstallMessageHandler, QLibraryInfo,
QtMsgType, qInstallMessageHandler,
) QtMsgType,
)
# Add the directory to Python path # Add the directory to Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__))) sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
# Set Hugging Face Hub environment variables # Set Hugging Face Hub environment variables
os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" # Disable Hugging Face telemetry with timed_log("config load + HF env setup", logger=_log):
os.environ["HF_HUB_ETAG_TIMEOUT"] = "10" # Metadata request timeout (seconds) os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" # Disable Hugging Face telemetry
os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "10" # File download timeout (seconds) os.environ["HF_HUB_ETAG_TIMEOUT"] = "10" # Metadata request timeout (seconds)
os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1" # Disable symlinks warning os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "10" # File download timeout (seconds)
from abogen.utils import load_config os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1" # Disable symlinks warning
if load_config().get("disable_kokoro_internet", False): from abogen.utils import load_config
print("INFO: Kokoro's internet access is disabled.") if load_config().get("disable_kokoro_internet", False):
os.environ["HF_HUB_OFFLINE"] = "1" # Disable Hugging Face Hub internet access _log.info("Kokoro's internet access is disabled.")
os.environ["HF_HUB_OFFLINE"] = "1" # Disable Hugging Face Hub internet access
from abogen.pyqt.gui import abogen with timed_log("GUI module import (abogen.pyqt.gui)", logger=_log):
from abogen.constants import PROGRAM_NAME, VERSION from abogen.pyqt.gui import abogen
from abogen.constants import PROGRAM_NAME, VERSION
# Set environment variables for AMD ROCm # Set environment variables for AMD ROCm
os.environ["MIOPEN_FIND_MODE"] = "FAST" os.environ["MIOPEN_FIND_MODE"] = "FAST"
@@ -150,7 +161,8 @@ if platform.system() == "Linux":
def main(): def main():
"""Main entry point for console usage.""" """Main entry point for console usage."""
app = QApplication(sys.argv) with timed_log("QApplication creation", logger=_log):
app = QApplication(sys.argv)
# Set application icon using get_resource_path from utils # Set application icon using get_resource_path from utils
icon_path = get_resource_path("abogen.assets", "icon.ico") icon_path = get_resource_path("abogen.assets", "icon.ico")
@@ -164,8 +176,11 @@ def main():
except AttributeError: except AttributeError:
pass pass
ex = abogen() with timed_log("main window construction", logger=_log):
ex.show() ex = abogen()
with timed_log("window show", logger=_log):
ex.show()
_log.info("App startup complete. Showing window.")
sys.exit(app.exec()) sys.exit(app.exec())
+7 -5
View File
@@ -28,7 +28,6 @@ from PyQt6.QtWidgets import (
from PyQt6.QtCore import Qt, QTimer, QPoint, QRect, QSize from PyQt6.QtCore import Qt, QTimer, QPoint, QRect, QSize
from PyQt6.QtGui import QPixmap, QIcon, QAction from PyQt6.QtGui import QPixmap, QIcon, QAction
from abogen.constants import ( from abogen.constants import (
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
LANGUAGE_DESCRIPTIONS, LANGUAGE_DESCRIPTIONS,
COLORS, COLORS,
) )
@@ -949,7 +948,9 @@ class VoiceFormulaDialog(QDialog):
lang = state.get("language") if isinstance(state, dict) else None lang = state.get("language") if isinstance(state, dict) else None
# apply language selection # apply language selection
if lang: if lang:
i = self.language_combo.findData(lang) from abogen.voice_profiles import resolve_profile_language
i = self.language_combo.findData(resolve_profile_language(state))
if i >= 0: if i >= 0:
self.language_combo.blockSignals(True) self.language_combo.blockSignals(True)
self.language_combo.setCurrentIndex(i) self.language_combo.setCurrentIndex(i)
@@ -1571,9 +1572,10 @@ class VoiceFormulaDialog(QDialog):
parent.selected_profile_name = None parent.selected_profile_name = None
lang = self.language_combo.currentData() lang = self.language_combo.currentData()
parent.selected_lang = lang parent.selected_lang = lang
parent.subtitle_combo.setEnabled( if hasattr(parent, "update_subtitle_options_availability"):
lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION parent.update_subtitle_options_availability()
) else:
parent.subtitle_combo.setEnabled(True)
# Reset start flag and trigger preview # Reset start flag and trigger preview
self._started = False self._started = False
parent.preview_voice() parent.preview_voice()
+6 -5
View File
@@ -6,10 +6,9 @@ from dataclasses import dataclass
from functools import lru_cache from functools import lru_cache
from typing import Any, Dict, Optional, Tuple from typing import Any, Dict, Optional, Tuple
try: # pragma: no cover - optional dependency # spaCy is intentionally NOT imported at module level: importing it pulls in
import spacy # thinc -> torch, which costs seconds of startup time. It is imported lazily
except Exception: # pragma: no cover - spaCy unavailable at runtime # inside _load_spacy_model below.
spacy = None
# Lazy spaCy type hints to avoid a hard dependency at import time. # Lazy spaCy type hints to avoid a hard dependency at import time.
Language = Any # type: ignore[assignment] Language = Any # type: ignore[assignment]
@@ -37,7 +36,9 @@ _DEFAULT_MODEL = os.environ.get("ABOGEN_SPACY_MODEL", "en_core_web_sm")
@lru_cache(maxsize=1) @lru_cache(maxsize=1)
def _load_spacy_model(model: str = _DEFAULT_MODEL) -> Optional[Language]: def _load_spacy_model(model: str = _DEFAULT_MODEL) -> Optional[Language]:
if spacy is None: try: # pragma: no cover - optional dependency
import spacy
except Exception: # pragma: no cover - spaCy unavailable at runtime
logger.debug("spaCy is not installed; skipping contraction disambiguation") logger.debug("spaCy is not installed; skipping contraction disambiguation")
return None return None
+3 -1
View File
@@ -57,7 +57,9 @@ def get_spacy_model(language: Language, log_callback=None):
print(msg) print(msg)
if not isinstance(language, Language): if not isinstance(language, Language):
raise TypeError(f"language must be Language enum, got {type(language).__name__}: {language!r}") raise TypeError(
f"language must be Language enum, got {type(language).__name__}: {language!r}"
)
if language in _nlp_cache: if language in _nlp_cache:
return _nlp_cache[language] return _nlp_cache[language]
+128 -3
View File
@@ -6,7 +6,9 @@ import re
import shutil import shutil
import subprocess import subprocess
import sys import sys
import time
import warnings import warnings
from contextlib import contextmanager
from threading import Thread from threading import Thread
from typing import Dict, Optional from typing import Dict, Optional
@@ -29,6 +31,125 @@ _load_environment()
warnings.filterwarnings("ignore") warnings.filterwarnings("ignore")
# --- Console log colorization via rich (mirrors AutoSubSync's approach) ---
try: # rich is a declared dependency, but degrade gracefully if unavailable
from rich.console import Console
from rich.highlighter import NullHighlighter
from rich.logging import RichHandler
_RICH_AVAILABLE = True
except Exception: # pragma: no cover - fallback to plain logging
Console = None
NullHighlighter = None
RichHandler = None
_RICH_AVAILABLE = False
def _console_supports_color() -> bool:
if os.environ.get("NO_COLOR"):
return False
try:
return bool(sys.stderr.isatty())
except Exception:
return False
_RICH_CONSOLE = None
if Console is not None:
try:
_RICH_CONSOLE = Console(stderr=True, no_color=not _console_supports_color())
except Exception: # pragma: no cover - defensive
_RICH_CONSOLE = None
_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
if RichHandler is not None:
class RichConsoleHandler(RichHandler):
"""RichHandler with default settings, except raw ANSI escapes are
stripped from messages first (werkzeug colorizes its own log lines
when attached to a TTY; without this they render as literal "[36m"
fragments)."""
def emit(self, record):
# Werkzeug logs its dev-server banner at INFO but hardcodes a
# "WARNING: " prefix into the message text. Promote the record so
# the level tag matches the content.
try:
message = _ANSI_ESCAPE_RE.sub("", record.getMessage())
except Exception: # pragma: no cover - defensive
message = ""
if record.levelno < logging.WARNING and message.startswith("WARNING: "):
record.levelno = logging.WARNING
record.levelname = "WARNING"
super().emit(record)
def render_message(self, record, message):
message = _ANSI_ESCAPE_RE.sub("", message)
if message.startswith("WARNING: "):
message = message[len("WARNING: ") :]
return super().render_message(record, message)
else: # pragma: no cover - rich unavailable fallback
RichConsoleHandler = None # type: ignore[assignment, misc]
def console_handler(show_level=True):
"""Build a colored console handler. Rich's RichHandler when available
(no timestamps, colored level tags), plain StreamHandler otherwise."""
if _RICH_CONSOLE is not None and RichConsoleHandler is not None:
return RichConsoleHandler(
console=_RICH_CONSOLE,
show_path=False,
show_time=False,
rich_tracebacks=True,
)
handler = logging.StreamHandler(sys.stderr)
prefix = "%(levelname)s - " if show_level else ""
handler.setFormatter(logging.Formatter(f"{prefix}%(message)s"))
return handler
def setup_console_logging(level=logging.INFO):
"""Configure the root logger once with a colored console handler."""
root = logging.getLogger()
if not root.handlers:
root.addHandler(console_handler())
root.setLevel(level)
@contextmanager
def timed_log(label, logger=None, level=logging.INFO):
"""Context manager that logs the wall-clock time a block of code takes.
Used to surface which load/startup steps are slow. The elapsed time is
colorized: green < 1s, yellow 1-5s, red > 5s.
"""
log = logger or logging.getLogger(__name__)
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
if _RICH_AVAILABLE and _RICH_CONSOLE is not None and not _RICH_CONSOLE.no_color:
if elapsed >= 5.0:
color = "red"
elif elapsed >= 1.0:
color = "yellow"
else:
color = "green"
log.log(
level,
"Loaded %s in %s",
f"[cyan]{label}[/cyan]",
f"[{color}]{elapsed:.2f}s[/{color}]",
extra={"markup": True, "highlighter": NullHighlighter()},
)
else:
log.log(level, "Loaded %s in %.2fs", label, elapsed)
def detect_encoding(file_path): def detect_encoding(file_path):
try: try:
@@ -527,9 +648,13 @@ class LoadPipelineThread(Thread):
try: try:
from abogen.domain.pipeline_factory import create_pipeline_for_job from abogen.domain.pipeline_factory import create_pipeline_for_job
backend = create_pipeline_for_job( with timed_log(
"kokoro", language=self.lang_code, use_gpu=self.use_gpu f"TTS pipeline (lang={self.lang_code}, gpu={self.use_gpu})",
) logger=logging.getLogger("abogen.startup"),
):
backend = create_pipeline_for_job(
"kokoro", language=self.lang_code, use_gpu=self.use_gpu
)
self.callback(backend, None) self.callback(backend, None)
except Exception as e: except Exception as e:
self.callback(None, str(e)) self.callback(None, str(e))
+24 -1
View File
@@ -2,6 +2,7 @@ import json
import os import os
from typing import Any, Dict, Iterable, List, Tuple from typing import Any, Dict, Iterable, List, Tuple
from abogen.domain.enums import Language
from abogen.tts_plugin.utils import get_voices, is_plugin_registered from abogen.tts_plugin.utils import get_voices, is_plugin_registered
from abogen.utils import get_user_config_path from abogen.utils import get_user_config_path
@@ -176,13 +177,35 @@ def save_profile(name: str, *, language: str, voices: Iterable) -> None:
raise ValueError("At least one voice with a weight above zero is required") raise ValueError("At least one voice with a weight above zero is required")
if not language: if not language:
language = "a" language = Language.EN_US
profiles = load_profiles() profiles = load_profiles()
profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized} profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized}
save_profiles(profiles) save_profiles(profiles)
def resolve_profile_language(entry: Any) -> Language:
"""Resolve a profile's stored language to a Language enum.
New profiles store ISO codes (Language enum values); legacy profiles may
store kokoro letter codes ("a", "b", ...). Unparseable values fall back
to EN_US.
"""
raw = entry.get("language") if isinstance(entry, dict) else None
if isinstance(raw, Language):
return raw
text = str(raw or "").strip()
if not text:
return Language.EN_US
try:
return Language.from_str(text)
except ValueError:
from plugins.kokoro.engine import language_for_code
return language_for_code(text)
def remove_profile(name: str) -> None: def remove_profile(name: str) -> None:
delete_profile(name) delete_profile(name)
+65 -43
View File
@@ -9,11 +9,19 @@ from flask import Flask
from abogen import shutdown # noqa: F401 from abogen import shutdown # noqa: F401
shutdown.register_shutdown() shutdown.register_shutdown()
from abogen.utils import get_user_cache_path, get_user_output_path, get_user_settings_dir from abogen.utils import (
get_user_cache_path,
get_user_output_path,
get_user_settings_dir,
setup_console_logging,
timed_log,
)
from .conversion_runner import run_conversion_job from .conversion_runner import run_conversion_job
from .service import build_service from .service import build_service
_logger = logging.getLogger("abogen.startup")
class _SuppressSuccessfulAccessFilter(logging.Filter): class _SuppressSuccessfulAccessFilter(logging.Filter):
"""Filter out successful (HTTP 200) werkzeug access logs.""" """Filter out successful (HTTP 200) werkzeug access logs."""
@@ -79,53 +87,57 @@ def _get_secret_key() -> str:
def create_app(config: Optional[dict[str, Any]] = None) -> Flask: def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
uploads_dir, outputs_dir = _default_dirs() with timed_log("default directories", logger=_logger):
uploads_dir, outputs_dir = _default_dirs()
app = Flask( with timed_log("Flask app creation + config", logger=_logger):
__name__, app = Flask(
static_folder="static", __name__,
template_folder="templates", static_folder="static",
) template_folder="templates",
base_config = { )
"SECRET_KEY": _get_secret_key(), base_config = {
"UPLOAD_FOLDER": str(uploads_dir), "SECRET_KEY": _get_secret_key(),
"OUTPUT_FOLDER": str(outputs_dir), "UPLOAD_FOLDER": str(uploads_dir),
"MAX_CONTENT_LENGTH": 1024 * 1024 * 400, # 400 MB uploads "OUTPUT_FOLDER": str(outputs_dir),
# Large books can submit four form fields per chapter. Werkzeug's "MAX_CONTENT_LENGTH": 1024 * 1024 * 400, # 400 MB uploads
# defaults reject those requests before the wizard route can process # Large books can submit four form fields per chapter. Werkzeug's
# them, even though the encoded payload is much smaller than the upload # defaults reject those requests before the wizard route can process
# limit above. # them, even though the encoded payload is much smaller than the upload
"MAX_FORM_MEMORY_SIZE": 10 * 1024 * 1024, # limit above.
"MAX_FORM_PARTS": 10_000, "MAX_FORM_MEMORY_SIZE": 10 * 1024 * 1024,
} "MAX_FORM_PARTS": 10_000,
if config: }
base_config.update(config) if config:
app.config.update(base_config) base_config.update(config)
app.config.update(base_config)
service = build_service( with timed_log("conversion service (incl. queue state load)", logger=_logger):
runner=run_conversion_job, service = build_service(
output_root=Path(app.config["OUTPUT_FOLDER"]), runner=run_conversion_job,
uploads_root=Path(app.config["UPLOAD_FOLDER"]), output_root=Path(app.config["OUTPUT_FOLDER"]),
) uploads_root=Path(app.config["UPLOAD_FOLDER"]),
)
app.extensions["conversion_service"] = service app.extensions["conversion_service"] = service
from abogen.webui.routes import ( with timed_log("blueprint registration", logger=_logger):
main_bp, from abogen.webui.routes import (
jobs_bp, main_bp,
settings_bp, jobs_bp,
voices_bp, settings_bp,
entities_bp, voices_bp,
books_bp, entities_bp,
api_bp, books_bp,
) api_bp,
)
app.register_blueprint(main_bp) app.register_blueprint(main_bp)
app.register_blueprint(jobs_bp, url_prefix="/jobs") app.register_blueprint(jobs_bp, url_prefix="/jobs")
app.register_blueprint(settings_bp, url_prefix="/settings") app.register_blueprint(settings_bp, url_prefix="/settings")
app.register_blueprint(voices_bp, url_prefix="/voices") app.register_blueprint(voices_bp, url_prefix="/voices")
app.register_blueprint(entities_bp, url_prefix="/overrides") app.register_blueprint(entities_bp, url_prefix="/overrides")
app.register_blueprint(books_bp, url_prefix="/find-books") app.register_blueprint(books_bp, url_prefix="/find-books")
app.register_blueprint(api_bp, url_prefix="/api") app.register_blueprint(api_bp, url_prefix="/api")
global _access_log_filter_attached global _access_log_filter_attached
if not _access_log_filter_attached: if not _access_log_filter_attached:
@@ -137,6 +149,16 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
def main() -> None: def main() -> None:
setup_console_logging()
# Route Flask's dev-server banner through our logger instead of click.echo.
import flask.cli as flask_cli
def _show_server_banner(debug, app_import_path):
_logger.info(" * Serving Flask app %r", app_import_path)
_logger.info(" * Debug mode: %s", "on" if debug else "off")
flask_cli.show_server_banner = _show_server_banner
app = create_app() app = create_app()
host = os.environ.get("ABOGEN_HOST", "0.0.0.0") host = os.environ.get("ABOGEN_HOST", "0.0.0.0")
port = int(os.environ.get("ABOGEN_PORT", "8808")) port = int(os.environ.get("ABOGEN_PORT", "8808"))
+11 -4
View File
@@ -47,10 +47,17 @@ def _resolve_voice_setting(value: str) -> tuple[str, Optional[str], Optional[str
def _load_pipeline(language: Language, use_gpu: bool) -> Any: def _load_pipeline(language: Language, use_gpu: bool) -> Any:
device = "cpu" import logging
if use_gpu: from abogen.utils import timed_log
device = _select_device()
return create_pipeline("kokoro", language=language, device=device) with timed_log(
f"TTS pipeline (lang={language}, gpu={use_gpu})",
logger=logging.getLogger("abogen.startup"),
):
device = "cpu"
if use_gpu:
device = _select_device()
return create_pipeline("kokoro", language=language, device=device)
def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]: def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]:
+2
View File
@@ -809,6 +809,8 @@ def build_pending_job_from_extraction(
analysis_requested=initial_analysis, analysis_requested=initial_analysis,
) )
apply_book_step_form(pending, form, settings=settings, profiles=profiles_map)
return PendingBuildResult( return PendingBuildResult(
pending=pending, pending=pending,
selected_speaker_config=selected_speaker_config or None, selected_speaker_config=selected_speaker_config or None,
+2 -4
View File
@@ -17,7 +17,7 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping
from abogen.domain.metadata_helpers import normalize_metadata_map from abogen.domain.metadata_helpers import normalize_metadata_map
from abogen.domain.enums import Language from abogen.domain.enums import Language
from abogen.utils import get_internal_cache_path, get_user_settings_dir from abogen.utils import console_handler, get_internal_cache_path, get_user_settings_dir
@@ -32,9 +32,7 @@ STATE_VERSION = 8
_JOB_LOGGER = logging.getLogger("abogen.jobs") _JOB_LOGGER = logging.getLogger("abogen.jobs")
if not _JOB_LOGGER.handlers: if not _JOB_LOGGER.handlers:
handler = logging.StreamHandler(sys.stdout) _JOB_LOGGER.addHandler(console_handler())
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S"))
_JOB_LOGGER.addHandler(handler)
_JOB_LOGGER.propagate = False _JOB_LOGGER.propagate = False
_JOB_LOGGER.setLevel(logging.DEBUG) _JOB_LOGGER.setLevel(logging.DEBUG)
+15 -5
View File
@@ -73,17 +73,27 @@ def engine_language(lang: Language) -> str:
return _KOKORO_LANG_MAP.get(lang, "a") return _KOKORO_LANG_MAP.get(lang, "a")
def language_for_voice_id(voice_id: str) -> Language: def language_for_code(code: str | None) -> Language:
"""Map a kokoro engine language code (single letter) to a Language enum.
Used to resolve legacy data such as old profile files that stored
kokoro letter codes. This is kokoro-specific knowledge that stays
inside the engine. Unparseable values fall back to EN_US.
"""
letter = str(code or "").strip()[:1].lower()
if letter in _CODE_TO_LANGUAGE:
return _CODE_TO_LANGUAGE[letter]
return Language.EN_US
def language_for_voice_id(voice_id: str | None) -> Language:
"""Determine which Language a voice belongs to from its voice ID. """Determine which Language a voice belongs to from its voice ID.
Kokoro voice IDs encode language as a prefix (e.g. "af_heart" "a" EN_US). Kokoro voice IDs encode language as a prefix (e.g. "af_heart" "a" EN_US).
This is kokoro-specific knowledge that stays inside the engine. This is kokoro-specific knowledge that stays inside the engine.
Callers pass a voice ID string; the engine returns a Language enum. Callers pass a voice ID string; the engine returns a Language enum.
""" """
prefix = str(voice_id or "").strip()[:1].lower() return language_for_code(voice_id)
if prefix in _CODE_TO_LANGUAGE:
return _CODE_TO_LANGUAGE[prefix]
return Language.EN_US
class KokoroSession: class KokoroSession:
+1
View File
@@ -50,6 +50,7 @@ dependencies = [
"num2words>=0.5.13", "num2words>=0.5.13",
"httpx>=0.27.0", "httpx>=0.27.0",
"PyQt6>=6.5.0", "PyQt6>=6.5.0",
"rich>=13.0.0",
] ]
classifiers = [ classifiers = [
+6 -1
View File
@@ -119,7 +119,7 @@ class TestLanguage:
def test_supports_subtitle_tokens(self): def test_supports_subtitle_tokens(self):
assert Language.EN_US.supports_subtitle_tokens is True assert Language.EN_US.supports_subtitle_tokens is True
assert Language.EN_GB.supports_subtitle_tokens is True assert Language.EN_GB.supports_subtitle_tokens is True
assert Language.ZH.supports_subtitle_tokens is False assert Language.ZH.supports_subtitle_tokens is True
def test_from_str_case_insensitive(self): def test_from_str_case_insensitive(self):
assert Language.from_str("EN-US") == Language.EN_US assert Language.from_str("EN-US") == Language.EN_US
@@ -129,3 +129,8 @@ class TestLanguage:
def test_from_str_invalid(self): def test_from_str_invalid(self):
with pytest.raises(ValueError, match="Invalid Language"): with pytest.raises(ValueError, match="Invalid Language"):
Language.from_str("en") Language.from_str("en")
def test_kokoro_letter_codes_not_accepted_by_domain(self):
# Letter codes are kokoro-engine internals, not domain API.
with pytest.raises(ValueError):
Language.from_str("a")
+39
View File
@@ -196,3 +196,42 @@ class TestKokoroVoiceLister:
assert isinstance(voice.tags, tuple) assert isinstance(voice.tags, tuple)
assert len(voice.tags) > 0 assert len(voice.tags) > 0
engine.dispose() engine.dispose()
# ──────────────────────────────────────────────────────────────
# Language mapping helpers
# ──────────────────────────────────────────────────────────────
class TestKokoroLanguageMapping:
"""Language resolution helpers: the letter codes live in the engine only."""
def test_language_for_code(self) -> None:
from abogen.domain.enums import Language
from plugins.kokoro.engine import language_for_code
assert language_for_code("a") == Language.EN_US
assert language_for_code("b") == Language.EN_GB
assert language_for_code("e") == Language.ES
assert language_for_code("f") == Language.FR
assert language_for_code("h") == Language.HI
assert language_for_code("i") == Language.IT
assert language_for_code("j") == Language.JA
assert language_for_code("p") == Language.PT_BR
assert language_for_code("z") == Language.ZH
def test_language_for_code_fallbacks(self) -> None:
from abogen.domain.enums import Language
from plugins.kokoro.engine import language_for_code
assert language_for_code("x") == Language.EN_US
assert language_for_code("") == Language.EN_US
def test_language_for_voice_id(self) -> None:
from abogen.domain.enums import Language
from plugins.kokoro.engine import language_for_voice_id
assert language_for_voice_id("af_heart") == Language.EN_US
assert language_for_voice_id("bf_emma") == Language.EN_GB
assert language_for_voice_id("ef_dora") == Language.ES
assert language_for_voice_id("zf_xiaobei") == Language.ZH
assert language_for_voice_id("") == Language.EN_US
+49
View File
@@ -0,0 +1,49 @@
import sys
import types
from pathlib import Path
from types import SimpleNamespace
_real_routes = sys.modules.get("abogen.webui.routes")
# Import routes.utils.form without executing routes/__init__.py (which imports
# every blueprint). Use a temporary namespace package, then restore the real
# module so later tests can still `from abogen.webui.routes import ...`.
routes_package = types.ModuleType("abogen.webui.routes")
routes_package.__path__ = [
str(Path(__file__).parents[1] / "abogen" / "webui" / "routes")
]
sys.modules["abogen.webui.routes"] = routes_package
from abogen.webui.routes.utils.form import ( # noqa: E402
build_pending_job_from_extraction,
load_settings,
)
if _real_routes is not None:
sys.modules["abogen.webui.routes"] = _real_routes
else:
del sys.modules["abogen.webui.routes"]
def test_user_metadata_overrides_extraction_fallback(tmp_path: Path) -> None:
extraction = SimpleNamespace(
chapters=[SimpleNamespace(title="Chapter 1", text="Text")],
metadata={"title": "423d828962c34d2b8a53bbe91176305a"},
cover_image=None,
cover_mime=None,
total_characters=4,
combined_text="Text",
)
result = build_pending_job_from_extraction(
stored_path=tmp_path / "book.txt",
original_name="book.txt",
extraction=extraction,
form={"meta_title": "My Book", "meta_author": "Ada Author"},
settings=load_settings(),
profiles={},
)
assert result.pending.metadata_tags["title"] == "My Book"
assert result.pending.metadata_tags["author"] == "Ada Author"
assert result.pending.metadata_tags["authors"] == "Ada Author"
+28
View File
@@ -0,0 +1,28 @@
"""Tests for voice profile language resolution."""
from __future__ import annotations
from abogen.domain.enums import Language
from abogen.voice_profiles import resolve_profile_language
class TestResolveProfileLanguage:
def test_iso_code(self) -> None:
assert resolve_profile_language({"language": "en-US"}) == Language.EN_US
assert resolve_profile_language({"language": "es"}) == Language.ES
def test_enum_value(self) -> None:
assert resolve_profile_language({"language": Language.ZH}) == Language.ZH
def test_legacy_kokoro_letter(self) -> None:
assert resolve_profile_language({"language": "a"}) == Language.EN_US
assert resolve_profile_language({"language": "e"}) == Language.ES
assert resolve_profile_language({"language": "z"}) == Language.ZH
def test_missing_or_unparseable_falls_back(self) -> None:
assert resolve_profile_language({}) == Language.EN_US
assert resolve_profile_language({"language": ""}) == Language.EN_US
assert resolve_profile_language({"language": "xx"}) == Language.EN_US
assert resolve_profile_language(None) == Language.EN_US
assert resolve_profile_language([]) == Language.EN_US