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/
.claude/
.coverage
# CodeGraph index (local, machine-specific)
.codegraph/
+6 -5
View File
@@ -70,11 +70,12 @@ SUPPORTED_INPUT_FORMATS = [
"vtt",
]
# Supported languages for subtitle generation
# Currently, only English (EN_US, EN_GB) are supported for subtitle generation.
# This is because tokens that contain timestamps are not generated for other languages in the Kokoro pipeline.
# Please refer to: https://github.com/hexgrad/kokoro/blob/6d87f4ae7abc2d14dbc4b3ef2e5f19852e861ac2/kokoro/pipeline.py
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = [Language.EN_US, Language.EN_GB]
# Supported languages for subtitle generation.
# All languages are supported: only English emits per-word timestamped tokens
# in the Kokoro pipeline, but other languages fall back to segment-level fake
# tokens (see abogen.domain.tokens.FakeToken), so subtitles are still
# generated at segment granularity.
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = list(Language)
# Voice and sample text mapping
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":
base += ["-c:a", "libmp3lame", "-qscale:a", "2"]
elif fmt == "opus":
base += ["-c:a", "libopus", "-b:a", "24000"]
base += ["-c:a", "libopus", "-b:a", "128000"]
elif fmt == "m4b":
base += ["-c:a", "aac", "-q:a", "2", "-movflags", "+faststart+use_metadata_tags"]
else:
+2 -4
View File
@@ -112,12 +112,10 @@ def spacy_pre_tts_segmentation(
def _to_language_enum(lang_code: Any) -> Language:
"""Convert lang_code to Language enum."""
if isinstance(lang_code, Language):
return lang_code
"""Convert lang_code to Language enum (ISO code or Language enum)."""
try:
return Language.from_str(str(lang_code))
except (ValueError, AttributeError):
except ValueError:
return Language.EN_US
+6 -2
View File
@@ -212,8 +212,12 @@ class Language(str, Enum):
@property
def supports_subtitle_tokens(self) -> bool:
"""True if this language generates timestamped tokens for subtitles."""
return self in (self.EN_US, self.EN_GB)
"""True if this language supports subtitle generation.
All languages are supported: languages without per-word timestamped
tokens fall back to segment-level fake tokens in the pipeline.
"""
return True
@classmethod
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.
Args:
language: Language enum value.
language: Language enum value, ISO code, or kokoro letter code.
subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.)
Returns:
+17 -5
View File
@@ -9,15 +9,26 @@ from collections import Counter
from dataclasses import dataclass, field
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]
Doc = 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 = (
"mr",
@@ -167,6 +178,7 @@ def _resolve_model_name(language: str) -> str:
def _load_model(language: str) -> Any:
spacy = _get_spacy()
if spacy is None:
raise EntityModelError(
"spaCy is not available. Install spaCy to enable entity extraction."
+15 -3
View File
@@ -5,10 +5,21 @@ import re
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
try: # pragma: no cover - optional dependency
_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 - optional dependency
import spacy # type: ignore
except Exception: # pragma: no cover - spaCy may be unavailable in minimal environments
except Exception: # pragma: no cover - spaCy may be unavailable in minimal environments
spacy = None
_SPACY = spacy
return _SPACY
@dataclass(frozen=True)
@@ -184,6 +195,7 @@ def _build_replacement_sentence(
def _load_spacy(language: str) -> Any:
spacy = _get_spacy()
if spacy is None:
return None
@@ -221,7 +233,7 @@ def extract_heteronym_overrides(
if not lang.startswith("en"):
return []
if spacy is None:
if _get_spacy() is None:
return []
nlp = _load_spacy(lang)
+3 -3
View File
@@ -45,9 +45,9 @@ import urllib.parse
import textwrap
# Setup logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
from abogen.utils import setup_console_logging
setup_console_logging()
_HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
_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.domain.split_pattern import get_split_pattern
from abogen.domain.enums import Language
from abogen.domain.subtitle_processor import (
parse_subtitle_file,
process_subtitle_entries,
@@ -872,12 +873,12 @@ class ConversionThread(QThread):
)
spacy_sentences = None
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
if (
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"]
):
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
self.log_updated.emit(
("\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,
subtitle_mode=self.subtitle_mode,
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),
)
+47 -39
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.book_handler import HandlerDialog
from abogen.domain.enums import Language
from abogen.constants import (
PROGRAM_NAME,
VERSION,
@@ -88,8 +89,9 @@ from abogen.constants import (
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
from abogen.voice_profiles import load_profiles, resolve_profile_language
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__
_DEFAULTS = all_settings_defaults()
@@ -397,11 +399,7 @@ class InputBox(QLabel):
# Re-enable subtitle and replace newlines controls when cleared
window = self.window()
if hasattr(window, "subtitle_combo"):
# Only enable if language supports it
current_lang = getattr(window, "selected_lang", "a")
window.subtitle_combo.setEnabled(
current_lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
)
window.subtitle_combo.setEnabled(True)
if hasattr(window, "replace_newlines_combo"):
window.replace_newlines_combo.setEnabled(True)
@@ -941,7 +939,7 @@ class abogen(QWidget):
self.selected_lang = None
else:
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.subtitle_mode = self.config.get("subtitle_mode", _d["subtitle_mode"])
self.max_subtitle_words = self.config.get("max_subtitle_words", _d["max_subtitle_words"])
@@ -988,6 +986,11 @@ class abogen(QWidget):
self.queued_items = []
self.current_queue_index = 0
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.update_speed_label()
@@ -1003,13 +1006,16 @@ class abogen(QWidget):
if self.selected_profile_name:
from abogen.voice_profiles import load_profiles
with timed_log("voice profile load", logger=_startup_log):
entry = load_profiles().get(self.selected_profile_name, {})
if isinstance(entry, dict):
self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language")
self.selected_lang = resolve_profile_language(entry)
else:
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:
self.save_path_label.setText(self.selected_output_folder)
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 + Highlighting: Subtitles with word-by-word karaoke highlighting.\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"
+ "\n".join(
f'"{lang}" => {LANGUAGE_DESCRIPTIONS.get(lang, lang)}'
@@ -1753,8 +1760,9 @@ class abogen(QWidget):
def update_subtitle_options_availability(self):
"""
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.
Update the enabled state of subtitle options based on the selected
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
is_subtitle_input = False
@@ -1763,16 +1771,14 @@ class abogen(QWidget):
):
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
self.subtitle_combo.setEnabled(not is_subtitle_input)
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
allowed_modes = ["Disabled", "Line", "Sentence", "Sentence + Comma"]
@@ -1787,10 +1793,7 @@ class abogen(QWidget):
if is_english:
item.setEnabled(True)
else:
if text in allowed_modes:
item.setEnabled(True)
else:
item.setEnabled(False)
item.setEnabled(text in allowed_modes)
# If current selection is disabled, switch to a valid one
current_text = self.subtitle_combo.currentText()
@@ -1809,7 +1812,7 @@ class abogen(QWidget):
def on_voice_changed(self, 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
save_config(self.config)
# Enable/disable subtitle options based on language
@@ -1826,10 +1829,12 @@ class abogen(QWidget):
# set mixed voices and language
if isinstance(entry, dict):
self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language")
self.selected_lang = resolve_profile_language(entry)
else:
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.config["selected_profile_name"] = pname
self.config.pop("selected_voice", None)
@@ -1839,7 +1844,7 @@ class abogen(QWidget):
else:
self.mixed_voice_state = 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
if "selected_profile_name" in self.config:
del self.config["selected_profile_name"]
@@ -1850,8 +1855,9 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles
entry = load_profiles().get(profile_name, {})
lang = entry.get("language") if isinstance(entry, dict) else None
enable = lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
enable = (
resolve_profile_language(entry) in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
)
self.subtitle_combo.setEnabled(enable)
self.subtitle_format_combo.setEnabled(enable)
@@ -2233,18 +2239,18 @@ class abogen(QWidget):
else:
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:
from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {})
selected_lang = entry.get("language")
selected_lang = resolve_profile_language(entry)
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
if not selected_lang:
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
def get_actual_subtitle_mode(self) -> str:
@@ -2420,7 +2426,7 @@ class abogen(QWidget):
self.update_log((gpu_msg, gpu_ok))
self.update_log("Loading modules...")
lang_code = self.selected_lang or "a"
lang_code = self.selected_lang or Language.EN_US
load_thread = LoadPipelineThread(
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
entry = load_profiles().get(self.selected_profile_name, {})
lang_to_cache = entry.get("language")
lang_to_cache = resolve_profile_language(entry)
else:
lang_to_cache = self.selected_lang
if not lang_to_cache and self.mixed_voice_state:
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]
else None
)
@@ -2860,7 +2866,7 @@ class abogen(QWidget):
)
self.loading_movie.start()
lang = self.selected_lang or "a"
lang = self.selected_lang or Language.EN_US
load_thread = LoadPipelineThread(
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
entry = load_profiles().get(self.selected_profile_name, {})
lang = entry.get("language")
lang = resolve_profile_language(entry)
else:
lang = self.selected_lang
if not lang and self.mixed_voice_state:
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]
else None
)
else:
lang = self.selected_voice[0]
lang = language_for_voice_id(self.selected_voice)
voice = self.selected_voice
# use same gpu/cpu logic as in conversion
@@ -3948,7 +3954,9 @@ Categories=AudioVideo;Audio;Utility;
initial_state = entry.get("voices", [])
else:
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(
self, initial_state=initial_state, selected_profile=selected_profile
)
+36 -21
View File
@@ -1,3 +1,4 @@
import logging
import os
import sys
import platform
@@ -6,8 +7,14 @@ import platform
from abogen import shutdown # noqa: F401
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
if platform.system() == "Windows":
with timed_log("PyTorch DLLs (Windows)", logger=_log):
import ctypes
from importlib.util import find_spec
@@ -25,7 +32,8 @@ if platform.system() == "Windows":
# Qt platform plugin detection (fixes #59)
try:
with timed_log("Qt platform plugin detection", logger=_log):
try:
from PyQt6.QtCore import QLibraryInfo
# Get the path to the plugins directory
@@ -39,17 +47,16 @@ try:
if os.path.isdir(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:
print("PyQt6 platform plugins not found at", platform_dir)
except ImportError:
print("PyQt6 not installed.")
_log.warning("PyQt6 platform plugins not found at %s", platform_dir)
except ImportError:
_log.warning("PyQt6 not installed.")
from abogen.utils import get_resource_path
# Pre-load "libxcb-cursor" on Linux (fixes #101)
if platform.system() == "Linux":
with timed_log("libxcb-cursor preload (Linux)", logger=_log):
arch = platform.machine().lower()
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)
if lib_filename:
@@ -71,6 +78,7 @@ if platform.system() == "Linux":
# Set application ID for Windows taskbar icon
if platform.system() == "Windows":
with timed_log("Windows AppUserModelID", logger=_log):
try:
from abogen.constants import PROGRAM_NAME, VERSION
import ctypes
@@ -78,31 +86,34 @@ if platform.system() == "Windows":
app_id = f"{PROGRAM_NAME}.{VERSION}"
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
except Exception as e:
print("Warning: failed to set AppUserModelID:", e)
_log.warning("Failed to set AppUserModelID: %s", e)
from PyQt6.QtWidgets import QApplication
from PyQt6.QtGui import QIcon
from PyQt6.QtCore import (
with timed_log("PyQt6 imports", logger=_log):
from PyQt6.QtWidgets import QApplication
from PyQt6.QtGui import QIcon
from PyQt6.QtCore import (
QLibraryInfo,
qInstallMessageHandler,
QtMsgType,
)
)
# Add the directory to Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
# Set Hugging Face Hub environment variables
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.")
with timed_log("config load + HF env setup", logger=_log):
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):
_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
from abogen.constants import PROGRAM_NAME, VERSION
with timed_log("GUI module import (abogen.pyqt.gui)", logger=_log):
from abogen.pyqt.gui import abogen
from abogen.constants import PROGRAM_NAME, VERSION
# Set environment variables for AMD ROCm
os.environ["MIOPEN_FIND_MODE"] = "FAST"
@@ -150,6 +161,7 @@ if platform.system() == "Linux":
def main():
"""Main entry point for console usage."""
with timed_log("QApplication creation", logger=_log):
app = QApplication(sys.argv)
# Set application icon using get_resource_path from utils
@@ -164,8 +176,11 @@ def main():
except AttributeError:
pass
with timed_log("main window construction", logger=_log):
ex = abogen()
with timed_log("window show", logger=_log):
ex.show()
_log.info("App startup complete. Showing window.")
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.QtGui import QPixmap, QIcon, QAction
from abogen.constants import (
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
LANGUAGE_DESCRIPTIONS,
COLORS,
)
@@ -949,7 +948,9 @@ class VoiceFormulaDialog(QDialog):
lang = state.get("language") if isinstance(state, dict) else None
# apply language selection
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:
self.language_combo.blockSignals(True)
self.language_combo.setCurrentIndex(i)
@@ -1571,9 +1572,10 @@ class VoiceFormulaDialog(QDialog):
parent.selected_profile_name = None
lang = self.language_combo.currentData()
parent.selected_lang = lang
parent.subtitle_combo.setEnabled(
lang in SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION
)
if hasattr(parent, "update_subtitle_options_availability"):
parent.update_subtitle_options_availability()
else:
parent.subtitle_combo.setEnabled(True)
# Reset start flag and trigger preview
self._started = False
parent.preview_voice()
+6 -5
View File
@@ -6,10 +6,9 @@ from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Dict, Optional, Tuple
try: # pragma: no cover - optional dependency
import spacy
except Exception: # pragma: no cover - spaCy unavailable at runtime
spacy = None
# spaCy is intentionally NOT imported at module level: importing it pulls in
# thinc -> torch, which costs seconds of startup time. It is imported lazily
# inside _load_spacy_model below.
# Lazy spaCy type hints to avoid a hard dependency at import time.
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)
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")
return None
+3 -1
View File
@@ -57,7 +57,9 @@ def get_spacy_model(language: Language, log_callback=None):
print(msg)
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:
return _nlp_cache[language]
+125
View File
@@ -6,7 +6,9 @@ import re
import shutil
import subprocess
import sys
import time
import warnings
from contextlib import contextmanager
from threading import Thread
from typing import Dict, Optional
@@ -29,6 +31,125 @@ _load_environment()
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):
try:
@@ -527,6 +648,10 @@ class LoadPipelineThread(Thread):
try:
from abogen.domain.pipeline_factory import create_pipeline_for_job
with timed_log(
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
)
+24 -1
View File
@@ -2,6 +2,7 @@ import json
import os
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.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")
if not language:
language = "a"
language = Language.EN_US
profiles = load_profiles()
profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized}
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:
delete_profile(name)
+23 -1
View File
@@ -9,11 +9,19 @@ from flask import Flask
from abogen import shutdown # noqa: F401
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 .service import build_service
_logger = logging.getLogger("abogen.startup")
class _SuppressSuccessfulAccessFilter(logging.Filter):
"""Filter out successful (HTTP 200) werkzeug access logs."""
@@ -79,8 +87,10 @@ def _get_secret_key() -> str:
def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
with timed_log("default directories", logger=_logger):
uploads_dir, outputs_dir = _default_dirs()
with timed_log("Flask app creation + config", logger=_logger):
app = Flask(
__name__,
static_folder="static",
@@ -102,6 +112,7 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
base_config.update(config)
app.config.update(base_config)
with timed_log("conversion service (incl. queue state load)", logger=_logger):
service = build_service(
runner=run_conversion_job,
output_root=Path(app.config["OUTPUT_FOLDER"]),
@@ -109,6 +120,7 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
)
app.extensions["conversion_service"] = service
with timed_log("blueprint registration", logger=_logger):
from abogen.webui.routes import (
main_bp,
jobs_bp,
@@ -137,6 +149,16 @@ def create_app(config: Optional[dict[str, Any]] = None) -> Flask:
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()
host = os.environ.get("ABOGEN_HOST", "0.0.0.0")
port = int(os.environ.get("ABOGEN_PORT", "8808"))
+7
View File
@@ -47,6 +47,13 @@ def _resolve_voice_setting(value: str) -> tuple[str, Optional[str], Optional[str
def _load_pipeline(language: Language, use_gpu: bool) -> Any:
import logging
from abogen.utils import timed_log
with timed_log(
f"TTS pipeline (lang={language}, gpu={use_gpu})",
logger=logging.getLogger("abogen.startup"),
):
device = "cpu"
if use_gpu:
device = _select_device()
+2
View File
@@ -809,6 +809,8 @@ def build_pending_job_from_extraction(
analysis_requested=initial_analysis,
)
apply_book_step_form(pending, form, settings=settings, profiles=profiles_map)
return PendingBuildResult(
pending=pending,
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.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")
if not _JOB_LOGGER.handlers:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S"))
_JOB_LOGGER.addHandler(handler)
_JOB_LOGGER.addHandler(console_handler())
_JOB_LOGGER.propagate = False
_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")
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.
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.
Callers pass a voice ID string; the engine returns a Language enum.
"""
prefix = str(voice_id or "").strip()[:1].lower()
if prefix in _CODE_TO_LANGUAGE:
return _CODE_TO_LANGUAGE[prefix]
return Language.EN_US
return language_for_code(voice_id)
class KokoroSession:
+1
View File
@@ -50,6 +50,7 @@ dependencies = [
"num2words>=0.5.13",
"httpx>=0.27.0",
"PyQt6>=6.5.0",
"rich>=13.0.0",
]
classifiers = [
+6 -1
View File
@@ -119,7 +119,7 @@ class TestLanguage:
def test_supports_subtitle_tokens(self):
assert Language.EN_US.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):
assert Language.from_str("EN-US") == Language.EN_US
@@ -129,3 +129,8 @@ class TestLanguage:
def test_from_str_invalid(self):
with pytest.raises(ValueError, match="Invalid Language"):
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 len(voice.tags) > 0
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