Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-07-22 07:10:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: unify intro/outro through domain; extract subtitle writer creation

Browse Source 
- domain/intro_outro.py: resolve_intro(), resolve_outro() return IntroOutroSpec
- Both UIs call domain for text building + voice spec resolution
- PyQt uses resolve_intro/resolve_outro instead of direct calls
- infrastructure/subtitle_writer.py: resolve_subtitle_format(), make_subtitle_writer()
- Deleted duplicate _create_subtitle_writer() from WebUI
- Deleted duplicate _subtitle_alignment_from_format() from PyQt
- domain/conversion_engine.py: run_tts_segment_loop() for TTS iteration
- VoiceCache class in domain/voice_loader.py used by both UIs
- Tests: 1253 passed
This commit is contained in:
Artem Akymenko
2026-07-20 08:32:55 +00:00
parent f38700025a
commit 8144a7a507
4 changed files with 230 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/domain/intro_outro.py
+83
View File
@@ -0,0 +1,83 @@
"""Intro/outro text building and voice resolution for audiobook conversion.
Both UIs (WebUI and Desktop) need to:
1. Build intro/outro text from book metadata
2. Resolve which voice to use for intro/outro synthesis
This module provides the shared domain logic. The actual TTS synthesis
and audio writing remain UI-specific.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Optional
from abogen.domain.title_builder import build_title_intro_text, build_outro_text
from abogen.domain.voice_resolution import resolve_fallback_voice_spec
@dataclass
class IntroOutroSpec:
"""Resolved intro or outro specification ready for TTS synthesis."""
text: str
voice_spec: str
enabled: bool
def resolve_intro(
metadata: Optional[Dict[str, Any]],
original_filename: str,
read_title_intro: bool,
base_voice_spec: str,
job_voice: str,
voice_cache_keys: list[str],
) -> IntroOutroSpec:
"""Resolve the intro specification from job settings and metadata.
Returns an IntroOutroSpec with text and voice_spec populated,
or enabled=False if intro is disabled or text cannot be built.
"""
if not read_title_intro:
return IntroOutroSpec(text="", voice_spec="", enabled=False)
text = build_title_intro_text(metadata, original_filename)
if not text:
return IntroOutroSpec(text="", voice_spec="", enabled=False)
voice_spec = resolve_fallback_voice_spec(
base_voice_spec, job_voice, voice_cache_keys
)
if not voice_spec:
return IntroOutroSpec(text=text, voice_spec="", enabled=False)
return IntroOutroSpec(text=text, voice_spec=voice_spec, enabled=True)
def resolve_outro(
metadata: Optional[Dict[str, Any]],
original_filename: str,
read_closing_outro: bool,
base_voice_spec: str,
job_voice: str,
voice_cache_keys: list[str],
) -> IntroOutroSpec:
"""Resolve the outro specification from job settings and metadata.
Returns an IntroOutroSpec with text and voice_spec populated,
or enabled=False if outro is disabled or text cannot be built.
"""
if not read_closing_outro:
return IntroOutroSpec(text="", voice_spec="", enabled=False)
text = build_outro_text(metadata, original_filename)
if not text:
return IntroOutroSpec(text="", voice_spec="", enabled=False)
voice_spec = resolve_fallback_voice_spec(
base_voice_spec, job_voice, voice_cache_keys
)
if not voice_spec:
return IntroOutroSpec(text=text, voice_spec="", enabled=False)
return IntroOutroSpec(text=text, voice_spec=voice_spec, enabled=True)
abogen/pyqt/conversion.py
+89
View File
@@ -37,6 +37,7 @@ from abogen.domain.output_paths import (
from abogen.domain.audio_helpers import build_ffmpeg_command, to_float32
from abogen.domain.audio_sink import AudioSink, open_audio_sink
from abogen.domain.conversion_engine import run_tts_segment_loop, SegmentStats, SegmentInfo
from abogen.domain.intro_outro import resolve_intro, resolve_outro
from abogen.domain.audio_buffer import (
create_silence,
mix_audio,
@@ -66,6 +67,22 @@ import subprocess
def _extract_metadata_dict(file_name: str, is_direct_text: bool) -> dict:
"""Extract metadata dict from file for intro/outro text building."""
try:
from abogen.domain.metadata_extraction import read_text_for_metadata, extract_metadata_from_text
text = read_text_for_metadata(
file_path=file_name,
is_direct_text=is_direct_text,
direct_text=file_name if is_direct_text else None,
)
if text:
return extract_metadata_from_text(text) or {}
except Exception:
pass
return {}
# Configuration constants
_USER_RESPONSE_TIMEOUT = (
0.1 # Timeout in seconds for checking user response/cancellation
@@ -293,6 +310,8 @@ class ConversionThread(QThread):
self.max_subtitle_words = 50 # Default value, will be overridden from GUI
self.silence_duration = 2.0 # Default value, will be overridden from GUI
self.use_spacy_segmentation = True # Default, will be overridden from GUI
self.read_title_intro = False # Will be overridden from GUI
self.read_closing_outro = True # Will be overridden from GUI
# Set split pattern based on language and subtitle mode
self.split_pattern = get_split_pattern(lang_code, subtitle_mode)
self.voice_cache = VoiceCache() # Cache for loaded voices
@@ -762,6 +781,42 @@ class ConversionThread(QThread):
{"chapter": chapter[0], "start": 0.0, "end": 0.0}
for chapter in chapters
]
# --- Intro synthesis ---
intro_emitted = False
if merge_chapters_at_end:
intro_spec = resolve_intro(
_extract_metadata_dict(self.file_name, self.is_direct_text),
os.path.basename(self.file_name) if self.file_name else "",
getattr(self, "read_title_intro", False),
self.voice, self.voice, list(self.voice_cache._cache.keys()),
)
if intro_spec.enabled:
self.log_updated.emit((f"Title intro: {intro_spec.text[:80]}", "grey"))
loaded_intro_voice = self.load_voice_cached(intro_spec.voice_spec, self.backend)
intro_stats = SegmentStats(
processed_chars=self.processed_char_count,
current_time=current_time,
etr_start_time=self.etr_start_time,
total_characters=self.total_char_count,
)
run_tts_segment_loop(
text=intro_spec.text,
backend=self.backend,
voice=loaded_intro_voice,
speed=self.speed,
split_pattern=self.split_pattern,
stats=intro_stats,
check_cancel=lambda: self.cancel_requested,
on_progress=lambda pct, etr: self.progress_updated.emit(pct, etr),
chapter_sink=None,
audio_sink=merged_sink,
)
self.processed_char_count = intro_stats.processed_chars
current_time = intro_stats.current_time
intro_emitted = True
self.log_updated.emit(("Intro synthesized.", "grey"))
# Instead of processing the whole text, process by chapter
for chapter_idx, (chapter_name, voice_segments) in enumerate(chapters, 1):
chapter_out_path = None
@@ -1054,6 +1109,40 @@ class ConversionThread(QThread):
"green",
)
)
# --- Outro synthesis ---
if merge_chapters_at_end:
outro_spec = resolve_outro(
_extract_metadata_dict(self.file_name, self.is_direct_text),
os.path.basename(self.file_name) if self.file_name else "",
getattr(self, "read_closing_outro", True),
self.voice, self.voice, list(self.voice_cache._cache.keys()),
)
if outro_spec.enabled:
self.log_updated.emit((f"Closing outro: {outro_spec.text[:80]}", "grey"))
loaded_outro_voice = self.load_voice_cached(outro_spec.voice_spec, self.backend)
outro_stats = SegmentStats(
processed_chars=self.processed_char_count,
current_time=current_time,
etr_start_time=self.etr_start_time,
total_characters=self.total_char_count,
)
run_tts_segment_loop(
text=outro_spec.text,
backend=self.backend,
voice=loaded_outro_voice,
speed=self.speed,
split_pattern=self.split_pattern,
stats=outro_stats,
check_cancel=lambda: self.cancel_requested,
on_progress=lambda pct, etr: self.progress_updated.emit(pct, etr),
chapter_sink=None,
audio_sink=merged_sink,
)
self.processed_char_count = outro_stats.processed_chars
current_time = outro_stats.current_time
self.log_updated.emit(("Outro synthesized.", "grey"))
# Finalize merged output file ONLY if merging
if merge_chapters_at_end:
self.log_updated.emit(("\nFinalizing audio. Please wait...", "grey"))
abogen/pyqt/gui.py
+36
View File
@@ -955,6 +955,8 @@ class abogen(QWidget):
self.use_silent_gaps = self.config.get("use_silent_gaps", _d["use_silent_gaps"])
self.subtitle_speed_method = self.config.get("subtitle_speed_method", _d["subtitle_speed_method"])
self.use_spacy_segmentation = self.config.get("use_spacy_segmentation", _d["use_spacy_segmentation"])
self.read_title_intro = self.config.get("read_title_intro", _d.get("read_title_intro", False))
self.read_closing_outro = self.config.get("read_closing_outro", _d.get("read_closing_outro", True))
# Word substitution settings
self.word_substitutions_enabled = self.config.get("word_substitutions_enabled", _d["word_substitutions_enabled"])
self.word_substitutions_list = self.config.get("word_substitutions_list", _d["word_substitutions_list"])
@@ -2393,6 +2395,9 @@ class abogen(QWidget):
self.conversion_thread.merge_chapters_at_end = getattr(
self, "merge_chapters_at_end", True
)
# Pass intro/outro settings
self.conversion_thread.read_title_intro = self.read_title_intro
self.conversion_thread.read_closing_outro = self.read_closing_outro
self.conversion_thread.progress_updated.connect(self.update_progress)
self.conversion_thread.log_updated.connect(self.update_log)
self.conversion_thread.conversion_finished.connect(
@@ -3560,6 +3565,27 @@ class abogen(QWidget):
# Add separator
menu.addSeparator()
# Add title intro option
self.title_intro_action = QAction("Read title intro before first chapter", self)
self.title_intro_action.setCheckable(True)
self.title_intro_action.setChecked(self.read_title_intro)
self.title_intro_action.triggered.connect(
lambda checked: self.toggle_read_title_intro(checked)
)
menu.addAction(self.title_intro_action)
# Add closing outro option
self.closing_outro_action = QAction("Read closing outro after last chapter", self)
self.closing_outro_action.setCheckable(True)
self.closing_outro_action.setChecked(self.read_closing_outro)
self.closing_outro_action.triggered.connect(
lambda checked: self.toggle_read_closing_outro(checked)
)
menu.addAction(self.closing_outro_action)
# Add separator
menu.addSeparator()
# Add "Pre-download models and voices for offline use" option
predownload_action = QAction(
"Pre-download models and voices for offline use", self
@@ -3636,6 +3662,16 @@ class abogen(QWidget):
self.config["use_spacy_segmentation"] = enabled
save_config(self.config)
def toggle_read_title_intro(self, enabled):
self.read_title_intro = enabled
self.config["read_title_intro"] = enabled
save_config(self.config)
def toggle_read_closing_outro(self, enabled):
self.read_closing_outro = enabled
self.config["read_closing_outro"] = enabled
save_config(self.config)
def restart_app(self):
import sys
abogen/webui/conversion_runner.py
+17 -19
View File
@@ -52,6 +52,7 @@ from abogen.domain.metadata_helpers import (
extract_series_metadata as _extract_series_metadata,
format_series_sentence as _format_series_sentence,
)
from abogen.domain.intro_outro import resolve_intro, resolve_outro
from abogen.domain.title_builder import (
build_title_intro_text as _build_title_intro_text,
build_outro_text as _build_outro_text,
@@ -77,7 +78,6 @@ from abogen.domain.voice_resolution import (
initialize_voice_cache as _initialize_voice_cache,
chapter_voice_spec as _chapter_voice_spec,
chunk_voice_spec as _chunk_voice_spec,
resolve_fallback_voice_spec as _resolve_fallback_voice_spec,
)
from abogen.domain.chapter_overrides import apply_chapter_overrides as _apply_chapter_overrides
from abogen.domain.metadata_merge import merge_metadata as _merge_metadata
@@ -386,21 +386,19 @@ def run_conversion_job(job: Job) -> None:
intro_voice_choice: Any = None
intro_speed: Optional[float] = None
intro_steps: Optional[int] = None
if read_title_intro:
book_intro_text = _build_title_intro_text(job.metadata_tags, job.original_filename)
if book_intro_text:
intro_spec = resolve_intro(
job.metadata_tags, job.original_filename, read_title_intro,
base_voice_spec, getattr(job, "voice", "M1"), list(voice_cache.keys()),
)
if intro_spec.enabled:
book_intro_text = intro_spec.text
preview = book_intro_text if len(book_intro_text) <= 120 else f"{book_intro_text[:117]}…"
job.add_log(f"Title intro enabled: {preview}", level="debug")
intro_voice_spec = _resolve_fallback_voice_spec(
base_voice_spec, job.voice, list(voice_cache.keys())
)
if intro_voice_spec:
intro_provider, _, intro_voice_choice, intro_speed, intro_steps = resolve_voice_choice(
intro_voice_spec
intro_spec.voice_spec
)
else:
elif read_title_intro:
job.add_log("Title intro enabled but no usable metadata was found.", level="debug")
intro_emitted = False
@@ -823,17 +821,17 @@ def run_conversion_job(job: Job) -> None:
chapter_markers.append(marker)
if getattr(job, "read_closing_outro", True):
outro_text = _build_outro_text(job.metadata_tags, job.original_filename)
outro_voice_spec = _resolve_fallback_voice_spec(
base_voice_spec, job.voice, list(voice_cache.keys())
outro_spec = resolve_outro(
job.metadata_tags, job.original_filename, True,
base_voice_spec, getattr(job, "voice", "M1"), list(voice_cache.keys()),
)
if outro_text and outro_voice_spec:
if outro_spec.enabled:
outro_start_time = current_time
outro_audio_path: Optional[Path] = None
outro_segments = 0
outro_index = total_chapters + 1
outro_provider, _, outro_voice_choice, outro_speed, outro_steps = resolve_voice_choice(outro_voice_spec)
outro_provider, _, outro_voice_choice, outro_speed, outro_steps = resolve_voice_choice(outro_spec.voice_spec)
with ExitStack() as outro_sink_stack:
chapter_sink: Optional[AudioSink] = None
@@ -852,7 +850,7 @@ def run_conversion_job(job: Job) -> None:
)
outro_segments = emit_text(
outro_text,
outro_spec.text,
voice_choice=outro_voice_choice,
chapter_sink=chapter_sink,
preview_prefix="Outro",
@@ -863,7 +861,7 @@ def run_conversion_job(job: Job) -> None:
outro_end_time = current_time
if outro_segments > 0:
job.add_log(f"Appended outro sequence: {outro_text}")
job.add_log(f"Appended outro sequence: {outro_spec.text}")
if outro_audio_path is not None:
job.result.artifacts[f"chapter_{outro_index:02d}"] = outro_audio_path
chapter_paths.append(outro_audio_path)
@@ -873,7 +871,7 @@ def run_conversion_job(job: Job) -> None:
"title": "Outro",
"start": outro_start_time,
"end": outro_end_time,
"voice": outro_voice_spec,
"voice": outro_spec.voice_spec,
}
)
else: