mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 11:40:57 +02:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aaa6ac112b | ||
|
|
11274ad6bf | ||
|
|
f340b976db | ||
|
|
9da15aefa4 | ||
|
|
7340d52ebb |
@@ -40,3 +40,6 @@ test_assets/
|
||||
dev_notes/
|
||||
.claude/
|
||||
.coverage
|
||||
|
||||
# CodeGraph index (local, machine-specific)
|
||||
.codegraph/
|
||||
|
||||
+6
-5
@@ -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 = {
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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
|
||||
import spacy # type: ignore
|
||||
except Exception: # pragma: no cover - spaCy may be unavailable in minimal environments
|
||||
spacy = None
|
||||
_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
|
||||
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)
|
||||
|
||||
@@ -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*")
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
|
||||
+49
-41
@@ -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,7 +986,12 @@ class abogen(QWidget):
|
||||
self.queued_items = []
|
||||
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.update_speed_label()
|
||||
# Set initial selection: prefer profile, else voice
|
||||
@@ -1003,13 +1006,16 @@ class abogen(QWidget):
|
||||
if self.selected_profile_name:
|
||||
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):
|
||||
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
|
||||
)
|
||||
|
||||
+89
-74
@@ -1,3 +1,4 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import platform
|
||||
@@ -6,103 +7,113 @@ 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":
|
||||
import ctypes
|
||||
from importlib.util import find_spec
|
||||
with timed_log("PyTorch DLLs (Windows)", logger=_log):
|
||||
import ctypes
|
||||
from importlib.util import find_spec
|
||||
|
||||
try:
|
||||
if (
|
||||
(spec := find_spec("torch"))
|
||||
and spec.origin
|
||||
and os.path.exists(
|
||||
dll_path := os.path.join(os.path.dirname(spec.origin), "lib", "c10.dll")
|
||||
)
|
||||
):
|
||||
ctypes.CDLL(os.path.normpath(dll_path))
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
if (
|
||||
(spec := find_spec("torch"))
|
||||
and spec.origin
|
||||
and os.path.exists(
|
||||
dll_path := os.path.join(os.path.dirname(spec.origin), "lib", "c10.dll")
|
||||
)
|
||||
):
|
||||
ctypes.CDLL(os.path.normpath(dll_path))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# Qt platform plugin detection (fixes #59)
|
||||
try:
|
||||
from PyQt6.QtCore import QLibraryInfo
|
||||
with timed_log("Qt platform plugin detection", logger=_log):
|
||||
try:
|
||||
from PyQt6.QtCore import QLibraryInfo
|
||||
|
||||
# Get the path to the plugins directory
|
||||
plugins = QLibraryInfo.path(QLibraryInfo.LibraryPath.PluginsPath)
|
||||
# Get the path to the plugins directory
|
||||
plugins = QLibraryInfo.path(QLibraryInfo.LibraryPath.PluginsPath)
|
||||
|
||||
# Normalize path to use the OS-native separators and absolute path
|
||||
platform_dir = os.path.normpath(os.path.join(plugins, "platforms"))
|
||||
# Normalize path to use the OS-native separators and absolute path
|
||||
platform_dir = os.path.normpath(os.path.join(plugins, "platforms"))
|
||||
|
||||
# Ensure we work with an absolute path for clarity
|
||||
platform_dir = os.path.abspath(platform_dir)
|
||||
# Ensure we work with an absolute path for clarity
|
||||
platform_dir = os.path.abspath(platform_dir)
|
||||
|
||||
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)
|
||||
else:
|
||||
print("PyQt6 platform plugins not found at", platform_dir)
|
||||
except ImportError:
|
||||
print("PyQt6 not installed.")
|
||||
if os.path.isdir(platform_dir):
|
||||
os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = platform_dir
|
||||
_log.info("QT_QPA_PLATFORM_PLUGIN_PATH set to: %s", platform_dir)
|
||||
else:
|
||||
_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":
|
||||
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:
|
||||
import ctypes
|
||||
try:
|
||||
# Try to load the system libxcb-cursor.so.0 first
|
||||
ctypes.CDLL('libxcb-cursor.so.0', mode=ctypes.RTLD_GLOBAL)
|
||||
except OSError:
|
||||
# System lib not available, load the bundled version
|
||||
lib_path = get_resource_path('abogen.libs', lib_filename)
|
||||
if lib_path:
|
||||
try:
|
||||
ctypes.CDLL(lib_path, mode=ctypes.RTLD_GLOBAL)
|
||||
except OSError:
|
||||
# If it fails (e.g. wrong glibc version on very old systems),
|
||||
# we simply ignore it and hope the system has the library.
|
||||
pass
|
||||
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:
|
||||
import ctypes
|
||||
try:
|
||||
# Try to load the system libxcb-cursor.so.0 first
|
||||
ctypes.CDLL('libxcb-cursor.so.0', mode=ctypes.RTLD_GLOBAL)
|
||||
except OSError:
|
||||
# System lib not available, load the bundled version
|
||||
lib_path = get_resource_path('abogen.libs', lib_filename)
|
||||
if lib_path:
|
||||
try:
|
||||
ctypes.CDLL(lib_path, mode=ctypes.RTLD_GLOBAL)
|
||||
except OSError:
|
||||
# If it fails (e.g. wrong glibc version on very old systems),
|
||||
# we simply ignore it and hope the system has the library.
|
||||
pass
|
||||
|
||||
|
||||
# Set application ID for Windows taskbar icon
|
||||
if platform.system() == "Windows":
|
||||
try:
|
||||
from abogen.constants import PROGRAM_NAME, VERSION
|
||||
import ctypes
|
||||
with timed_log("Windows AppUserModelID", logger=_log):
|
||||
try:
|
||||
from abogen.constants import PROGRAM_NAME, VERSION
|
||||
import ctypes
|
||||
|
||||
app_id = f"{PROGRAM_NAME}.{VERSION}"
|
||||
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
|
||||
except Exception as e:
|
||||
print("Warning: failed to set AppUserModelID:", e)
|
||||
app_id = f"{PROGRAM_NAME}.{VERSION}"
|
||||
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
|
||||
except Exception as e:
|
||||
_log.warning("Failed to set AppUserModelID: %s", e)
|
||||
|
||||
from PyQt6.QtWidgets import QApplication
|
||||
from PyQt6.QtGui import QIcon
|
||||
from PyQt6.QtCore import (
|
||||
QLibraryInfo,
|
||||
qInstallMessageHandler,
|
||||
QtMsgType,
|
||||
)
|
||||
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.")
|
||||
os.environ["HF_HUB_OFFLINE"] = "1" # Disable Hugging Face Hub internet access
|
||||
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,7 +161,8 @@ if platform.system() == "Linux":
|
||||
|
||||
def main():
|
||||
"""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
|
||||
icon_path = get_resource_path("abogen.assets", "icon.ico")
|
||||
@@ -164,8 +176,11 @@ def main():
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
ex = abogen()
|
||||
ex.show()
|
||||
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())
|
||||
|
||||
|
||||
|
||||
@@ -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,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
|
||||
|
||||
|
||||
@@ -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]
|
||||
|
||||
+128
-3
@@ -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,9 +648,13 @@ class LoadPipelineThread(Thread):
|
||||
try:
|
||||
from abogen.domain.pipeline_factory import create_pipeline_for_job
|
||||
|
||||
backend = create_pipeline_for_job(
|
||||
"kokoro", language=self.lang_code, use_gpu=self.use_gpu
|
||||
)
|
||||
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
|
||||
)
|
||||
self.callback(backend, None)
|
||||
except Exception as e:
|
||||
self.callback(None, str(e))
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
+65
-43
@@ -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,53 +87,57 @@ def _get_secret_key() -> str:
|
||||
|
||||
|
||||
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(
|
||||
__name__,
|
||||
static_folder="static",
|
||||
template_folder="templates",
|
||||
)
|
||||
base_config = {
|
||||
"SECRET_KEY": _get_secret_key(),
|
||||
"UPLOAD_FOLDER": str(uploads_dir),
|
||||
"OUTPUT_FOLDER": str(outputs_dir),
|
||||
"MAX_CONTENT_LENGTH": 1024 * 1024 * 400, # 400 MB uploads
|
||||
# Large books can submit four form fields per chapter. Werkzeug's
|
||||
# defaults reject those requests before the wizard route can process
|
||||
# them, even though the encoded payload is much smaller than the upload
|
||||
# limit above.
|
||||
"MAX_FORM_MEMORY_SIZE": 10 * 1024 * 1024,
|
||||
"MAX_FORM_PARTS": 10_000,
|
||||
}
|
||||
if config:
|
||||
base_config.update(config)
|
||||
app.config.update(base_config)
|
||||
with timed_log("Flask app creation + config", logger=_logger):
|
||||
app = Flask(
|
||||
__name__,
|
||||
static_folder="static",
|
||||
template_folder="templates",
|
||||
)
|
||||
base_config = {
|
||||
"SECRET_KEY": _get_secret_key(),
|
||||
"UPLOAD_FOLDER": str(uploads_dir),
|
||||
"OUTPUT_FOLDER": str(outputs_dir),
|
||||
"MAX_CONTENT_LENGTH": 1024 * 1024 * 400, # 400 MB uploads
|
||||
# Large books can submit four form fields per chapter. Werkzeug's
|
||||
# defaults reject those requests before the wizard route can process
|
||||
# them, even though the encoded payload is much smaller than the upload
|
||||
# limit above.
|
||||
"MAX_FORM_MEMORY_SIZE": 10 * 1024 * 1024,
|
||||
"MAX_FORM_PARTS": 10_000,
|
||||
}
|
||||
if config:
|
||||
base_config.update(config)
|
||||
app.config.update(base_config)
|
||||
|
||||
service = build_service(
|
||||
runner=run_conversion_job,
|
||||
output_root=Path(app.config["OUTPUT_FOLDER"]),
|
||||
uploads_root=Path(app.config["UPLOAD_FOLDER"]),
|
||||
)
|
||||
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"]),
|
||||
uploads_root=Path(app.config["UPLOAD_FOLDER"]),
|
||||
)
|
||||
app.extensions["conversion_service"] = service
|
||||
|
||||
from abogen.webui.routes import (
|
||||
main_bp,
|
||||
jobs_bp,
|
||||
settings_bp,
|
||||
voices_bp,
|
||||
entities_bp,
|
||||
books_bp,
|
||||
api_bp,
|
||||
)
|
||||
with timed_log("blueprint registration", logger=_logger):
|
||||
from abogen.webui.routes import (
|
||||
main_bp,
|
||||
jobs_bp,
|
||||
settings_bp,
|
||||
voices_bp,
|
||||
entities_bp,
|
||||
books_bp,
|
||||
api_bp,
|
||||
)
|
||||
|
||||
app.register_blueprint(main_bp)
|
||||
app.register_blueprint(jobs_bp, url_prefix="/jobs")
|
||||
app.register_blueprint(settings_bp, url_prefix="/settings")
|
||||
app.register_blueprint(voices_bp, url_prefix="/voices")
|
||||
app.register_blueprint(entities_bp, url_prefix="/overrides")
|
||||
app.register_blueprint(books_bp, url_prefix="/find-books")
|
||||
app.register_blueprint(api_bp, url_prefix="/api")
|
||||
app.register_blueprint(main_bp)
|
||||
app.register_blueprint(jobs_bp, url_prefix="/jobs")
|
||||
app.register_blueprint(settings_bp, url_prefix="/settings")
|
||||
app.register_blueprint(voices_bp, url_prefix="/voices")
|
||||
app.register_blueprint(entities_bp, url_prefix="/overrides")
|
||||
app.register_blueprint(books_bp, url_prefix="/find-books")
|
||||
app.register_blueprint(api_bp, url_prefix="/api")
|
||||
|
||||
global _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:
|
||||
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"))
|
||||
|
||||
@@ -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:
|
||||
device = "cpu"
|
||||
if use_gpu:
|
||||
device = _select_device()
|
||||
return create_pipeline("kokoro", language=language, device=device)
|
||||
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()
|
||||
return create_pipeline("kokoro", language=language, device=device)
|
||||
|
||||
|
||||
def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -50,6 +50,7 @@ dependencies = [
|
||||
"num2words>=0.5.13",
|
||||
"httpx>=0.27.0",
|
||||
"PyQt6>=6.5.0",
|
||||
"rich>=13.0.0",
|
||||
]
|
||||
|
||||
classifiers = [
|
||||
|
||||
+6
-1
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user