mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-22 07:10:28 +02:00
refactor: unify duplicated logic between WebUI and PyQt
domain/output_paths.py: - Add sanitize_filename_for_chapter() with OS safety + smart truncation - Keep existing slugify() for backward compatibility domain/text_chapters.py (NEW): - parse_chapters_from_text() combines intro preservation (PyQt) + clean_text (WebUI) PyQt/conversion.py: - Replace 3x copy-pasted ASS/SRT headers with create_subtitle_writer() - Replace inline M4B muxing with ExportService.embed_m4b_metadata() - Replace inline chapter splitting with parse_chapters_from_text() - Replace inline chapter filename sanitization with sanitize_filename_for_chapter() - Remove unused _CHAPTER_MARKER_SEARCH_PATTERN import Tests: 1152 passed (+21 new)
This commit is contained in:
Artem Akymenko
@@ -6,6 +6,7 @@ and computing project folder layouts.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
import re
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -16,6 +17,15 @@ from abogen.text_extractor import ExtractedChapter
|
||||
|
||||
_OUTPUT_SANITIZE_RE = re.compile(r"[^\w\-_.]+")
|
||||
|
||||
# OS-specific illegal characters for filenames
|
||||
_WINDOWS_ILLEGAL_CHARS_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||||
_UNIX_CONTROL_CHARS_RE = re.compile(r'[\x00-\x1f]')
|
||||
_RESERVED_NAMES = frozenset(
|
||||
{"CON", "PRN", "AUX", "NUL"}
|
||||
| {f"COM{i}" for i in range(1, 10)}
|
||||
| {f"LPT{i}" for i in range(1, 10)}
|
||||
)
|
||||
|
||||
|
||||
def slugify(title: str, index: int) -> str:
|
||||
sanitized = re.sub(r"[^\w\-]+", "_", title.lower()).strip("_")
|
||||
@@ -24,6 +34,46 @@ def slugify(title: str, index: int) -> str:
|
||||
return sanitized[:80]
|
||||
|
||||
|
||||
def sanitize_filename_for_chapter(title: str, index: int, max_len: int = 80) -> str:
|
||||
"""Sanitize a chapter name for use as a filename component.
|
||||
|
||||
Combines character sanitization, OS safety, and smart truncation
|
||||
at word boundaries. Prepends zero-padded index prefix.
|
||||
|
||||
Args:
|
||||
title: Raw chapter title.
|
||||
index: 1-based chapter number for prefix.
|
||||
max_len: Maximum length of the sanitized portion (excluding prefix).
|
||||
|
||||
Returns:
|
||||
Sanitized string like "01_the_beginning".
|
||||
"""
|
||||
# Remove non-word/non-space/non-hyphen chars, then collapse spaces/hyphens
|
||||
sanitized = re.sub(r"[^\w\s\-]", "", title)
|
||||
sanitized = re.sub(r"[\s\-]+", "_", sanitized).strip("_")
|
||||
|
||||
if not sanitized:
|
||||
sanitized = f"chapter_{index:02d}"
|
||||
|
||||
# OS-specific sanitization
|
||||
system = platform.system()
|
||||
if system == "Windows":
|
||||
sanitized = _WINDOWS_ILLEGAL_CHARS_RE.sub("_", sanitized)
|
||||
sanitized = sanitized.rstrip(". ")
|
||||
base = sanitized.split(".")[0].upper()
|
||||
if base in _RESERVED_NAMES:
|
||||
sanitized = f"_{sanitized}"
|
||||
# Linux: only NUL is truly illegal, but control chars are problematic
|
||||
sanitized = _UNIX_CONTROL_CHARS_RE.sub("_", sanitized)
|
||||
|
||||
# Smart truncation at word boundary
|
||||
if len(sanitized) > max_len:
|
||||
pos = sanitized[:max_len].rfind("_")
|
||||
sanitized = sanitized[: pos if pos > 0 else max_len].rstrip("_")
|
||||
|
||||
return f"{index:02d}_{sanitized}"
|
||||
|
||||
|
||||
def sanitize_output_stem(name: str) -> str:
|
||||
base = Path(name or "").stem
|
||||
sanitized = _OUTPUT_SANITIZE_RE.sub("_", base).strip("_")
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Pipeline creation, caching and lifecycle management.
|
||||
|
||||
Provides a unified interface for creating and managing TTS pipelines
|
||||
across all UI layers (WebUI, PyQt, CLI).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from abogen.domain.device import select_device
|
||||
from abogen.domain.voice_resolution import initialize_voice_cache
|
||||
from abogen.tts_plugin.utils import create_pipeline, is_plugin_registered
|
||||
|
||||
|
||||
def resolve_device(use_gpu: bool) -> str:
|
||||
"""Determine compute device from job and global config flags."""
|
||||
from abogen.utils import load_config
|
||||
|
||||
cfg = load_config()
|
||||
if use_gpu and cfg.get("use_gpu", True):
|
||||
return select_device()
|
||||
return "cpu"
|
||||
|
||||
|
||||
def create_pipeline_for_job(
|
||||
provider: str,
|
||||
language: str,
|
||||
use_gpu: bool,
|
||||
) -> Any:
|
||||
"""Create a TTS pipeline with proper device selection.
|
||||
|
||||
Handles provider validation, GPU decision, and plugin checks.
|
||||
"""
|
||||
provider = str(provider or "kokoro").strip().lower() or "kokoro"
|
||||
if not is_plugin_registered(provider):
|
||||
provider = "kokoro"
|
||||
|
||||
if provider == "supertonic":
|
||||
return create_pipeline("supertonic")
|
||||
|
||||
device = resolve_device(use_gpu)
|
||||
return create_pipeline("kokoro", lang_code=language, device=device)
|
||||
|
||||
|
||||
def dispose_pipelines(pipelines: Dict[str, Any]) -> None:
|
||||
"""Dispose all pipelines in a dict and clear it."""
|
||||
for p in pipelines.values():
|
||||
try:
|
||||
p.dispose()
|
||||
except Exception:
|
||||
pass
|
||||
pipelines.clear()
|
||||
|
||||
|
||||
class PipelinePool:
|
||||
"""Cache and manage TTS pipelines by provider.
|
||||
|
||||
Usage::
|
||||
|
||||
pool = PipelinePool()
|
||||
backend = pool.get("kokoro", "en", use_gpu=True)
|
||||
# ... use backend ...
|
||||
pool.dispose_all()
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._pipelines: Dict[str, Any] = {}
|
||||
self._voice_cache_initialized = False
|
||||
|
||||
def get(
|
||||
self,
|
||||
provider: str,
|
||||
language: str,
|
||||
use_gpu: bool,
|
||||
*,
|
||||
job: Any = None,
|
||||
) -> Any:
|
||||
"""Get or create a cached pipeline for the given provider.
|
||||
|
||||
Args:
|
||||
provider: TTS provider name ("kokoro" or "supertonic").
|
||||
language: Language code (for kokoro).
|
||||
use_gpu: Whether GPU acceleration is requested.
|
||||
job: Optional job object for voice cache initialization.
|
||||
"""
|
||||
provider = str(provider or "kokoro").strip().lower() or "kokoro"
|
||||
if not is_plugin_registered(provider):
|
||||
provider = "kokoro"
|
||||
|
||||
existing = self._pipelines.get(provider)
|
||||
if existing is not None:
|
||||
return existing
|
||||
|
||||
pipeline = create_pipeline_for_job(provider, language, use_gpu)
|
||||
self._pipelines[provider] = pipeline
|
||||
|
||||
if provider == "kokoro" and not self._voice_cache_initialized and job is not None:
|
||||
initialize_voice_cache(job)
|
||||
self._voice_cache_initialized = True
|
||||
|
||||
return pipeline
|
||||
|
||||
def dispose_all(self) -> None:
|
||||
"""Dispose all cached pipelines."""
|
||||
dispose_pipelines(self._pipelines)
|
||||
self._voice_cache_initialized = False
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Chapter parsing from raw text.
|
||||
|
||||
Provides a unified function for splitting text by chapter markers,
|
||||
used by both WebUI and PyQt conversion runners.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import List, Tuple
|
||||
|
||||
from abogen.subtitle_utils import clean_text
|
||||
|
||||
|
||||
_CHAPTER_MARKER_RE = re.compile(r"<<CHAPTER_MARKER:(.*?)>>", re.IGNORECASE)
|
||||
|
||||
|
||||
def parse_chapters_from_text(
|
||||
text: str,
|
||||
default_title: str = "text",
|
||||
clean: bool = True,
|
||||
) -> List[Tuple[str, str]]:
|
||||
"""Split raw text into chapters using chapter marker patterns.
|
||||
|
||||
Preserves content before the first marker as "Introduction" if present.
|
||||
Optionally applies clean_text() to each chapter segment.
|
||||
|
||||
Args:
|
||||
text: Raw text possibly containing <<CHAPTER_MARKER:Title>> markers.
|
||||
default_title: Fallback title when no markers are found.
|
||||
clean: Whether to apply clean_text() to each segment.
|
||||
|
||||
Returns:
|
||||
List of (title, text) tuples.
|
||||
"""
|
||||
matches = list(_CHAPTER_MARKER_RE.finditer(text))
|
||||
if not matches:
|
||||
cleaned = clean_text(text) if clean else text
|
||||
return [(default_title, cleaned)]
|
||||
|
||||
chapters: List[Tuple[str, str]] = []
|
||||
|
||||
# Preserve content before first marker as "Introduction"
|
||||
first_start = matches[0].start()
|
||||
if first_start > 0:
|
||||
intro_text = text[:first_start].strip()
|
||||
if intro_text:
|
||||
chapters.append(("Introduction", clean_text(intro_text) if clean else intro_text))
|
||||
|
||||
for idx, match in enumerate(matches):
|
||||
start = match.end()
|
||||
end = matches[idx + 1].start() if idx + 1 < len(matches) else len(text)
|
||||
chapter_name = match.group(1).strip() or default_title
|
||||
chapter_text = text[start:end].strip()
|
||||
if clean:
|
||||
chapter_text = clean_text(chapter_text)
|
||||
chapters.append((chapter_name, chapter_text))
|
||||
|
||||
return chapters
|
||||
@@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Mapping, Optional, Tuple, Set
|
||||
from typing import Any, Dict, Mapping, Optional, Tuple, Set
|
||||
|
||||
from abogen.voice_formulas import extract_voice_ids, get_new_voice
|
||||
from abogen.tts_plugin.utils import get_voices
|
||||
@@ -95,3 +95,36 @@ def coerce_truthy(value: Any, default: bool = True) -> bool:
|
||||
if value is None:
|
||||
return default
|
||||
return bool(value)
|
||||
|
||||
|
||||
def resolve_voice_target(
|
||||
raw_spec: str,
|
||||
normalized_profiles: Dict[str, Dict[str, Any]],
|
||||
*,
|
||||
job_voice: str = "M1",
|
||||
job_tts_provider: str = "kokoro",
|
||||
job_supertonic_total_steps: int = 5,
|
||||
job_speed: float = 1.0,
|
||||
) -> Tuple[str, str, Optional[float], Optional[int]]:
|
||||
"""Resolve a raw voice spec into (provider, voice_spec, speed_override, steps_override).
|
||||
|
||||
Pure function — all dependencies are passed as parameters.
|
||||
"""
|
||||
spec = str(raw_spec or "").strip()
|
||||
speaker_name, _ = split_speaker_reference(spec)
|
||||
if speaker_name and speaker_name in normalized_profiles:
|
||||
entry = normalized_profiles[speaker_name]
|
||||
provider = str(entry.get("provider") or "kokoro").strip().lower() or "kokoro"
|
||||
if provider == "supertonic":
|
||||
voice = str(entry.get("voice") or job_voice or "M1").strip() or "M1"
|
||||
steps = int(entry.get("total_steps") or job_supertonic_total_steps or 5)
|
||||
speed = float(entry.get("speed") or job_speed or 1.0)
|
||||
return "supertonic", supertonic_voice_from_spec(voice, job_voice), speed, steps
|
||||
formula = formula_from_kokoro_entry(entry)
|
||||
return "kokoro", formula or spec, None, None
|
||||
|
||||
fallback_provider = str(job_tts_provider or "kokoro").strip().lower() or "kokoro"
|
||||
inferred = infer_provider_from_spec(spec, fallback=fallback_provider)
|
||||
if inferred == "supertonic":
|
||||
return "supertonic", supertonic_voice_from_spec(spec, job_voice), None, None
|
||||
return "kokoro", spec, None, None
|
||||
+95
-307
@@ -5,7 +5,7 @@ import hashlib # For generating unique cache filenames
|
||||
from platformdirs import user_desktop_dir
|
||||
from PyQt6.QtCore import QThread, pyqtSignal, Qt, QTimer
|
||||
from PyQt6.QtWidgets import QCheckBox, QVBoxLayout, QDialog, QLabel, QDialogButtonBox
|
||||
from contextlib import ExitStack
|
||||
from contextlib import ExitStack, contextmanager
|
||||
import numpy as np
|
||||
import soundfile as sf
|
||||
from abogen.utils import (
|
||||
@@ -21,12 +21,13 @@ from abogen.constants import (
|
||||
SUPPORTED_SOUND_FORMATS,
|
||||
SUPPORTED_SUBTITLE_FORMATS,
|
||||
)
|
||||
from abogen.infrastructure.subtitle_writer import _format_timestamp
|
||||
from abogen.infrastructure.subtitle_writer import create_subtitle_writer
|
||||
from abogen.domain.split_pattern import get_split_pattern
|
||||
from abogen.domain.output_paths import (
|
||||
resolve_output_directory,
|
||||
build_output_path,
|
||||
sanitize_output_stem,
|
||||
sanitize_filename_for_chapter,
|
||||
)
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command, to_float32
|
||||
from abogen.domain.audio_sink import AudioSink, open_audio_sink
|
||||
@@ -48,13 +49,29 @@ from abogen.domain.pronunciation import (
|
||||
)
|
||||
from abogen.domain.metadata_extraction import (
|
||||
extract_metadata_and_build_args,
|
||||
extract_metadata_from_text,
|
||||
read_text_for_metadata,
|
||||
)
|
||||
from abogen.domain.text_chapters import parse_chapters_from_text
|
||||
from abogen.infrastructure.exporters import ExportService
|
||||
import abogen.hf_tracker as hf_tracker
|
||||
import static_ffmpeg
|
||||
import threading # for efficient waiting
|
||||
import subprocess
|
||||
|
||||
|
||||
def _subtitle_alignment_from_format(subtitle_format: str) -> str:
|
||||
"""Map PyQt subtitle format string to SubtitleAlignment value."""
|
||||
if subtitle_format in ("ass_centered_wide", "ass_centered_narrow"):
|
||||
if subtitle_format == "ass_centered_narrow":
|
||||
return "center_narrow"
|
||||
return "center"
|
||||
if subtitle_format in ("ass_narrow",):
|
||||
return "narrow"
|
||||
return "left"
|
||||
|
||||
|
||||
|
||||
# Configuration constants
|
||||
_USER_RESPONSE_TIMEOUT = (
|
||||
0.1 # Timeout in seconds for checking user response/cancellation
|
||||
@@ -69,7 +86,6 @@ from abogen.subtitle_utils import (
|
||||
parse_ass_file,
|
||||
get_sample_voice_text,
|
||||
sanitize_name_for_os,
|
||||
_CHAPTER_MARKER_SEARCH_PATTERN,
|
||||
split_text_by_voice_markers
|
||||
)
|
||||
|
||||
@@ -552,28 +568,7 @@ class ConversionThread(QThread):
|
||||
self._usage_counter = {}
|
||||
|
||||
# --- Chapter splitting logic ---
|
||||
# Use pre-compiled pattern for better performance
|
||||
chapter_splits = list(_CHAPTER_MARKER_SEARCH_PATTERN.finditer(text))
|
||||
chapters = []
|
||||
if chapter_splits:
|
||||
# prepend Introduction for content before first marker
|
||||
first_start = chapter_splits[0].start()
|
||||
if first_start > 0:
|
||||
intro_text = text[:first_start].strip()
|
||||
if intro_text:
|
||||
chapters.append(("Introduction", intro_text))
|
||||
for idx, match in enumerate(chapter_splits):
|
||||
start = match.end()
|
||||
end = (
|
||||
chapter_splits[idx + 1].start()
|
||||
if idx + 1 < len(chapter_splits)
|
||||
else len(text)
|
||||
)
|
||||
chapter_name = match.group(1).strip()
|
||||
chapter_text = text[start:end].strip()
|
||||
chapters.append((chapter_name, chapter_text))
|
||||
else:
|
||||
chapters = [("text", text)]
|
||||
chapters = parse_chapters_from_text(text, clean=False)
|
||||
total_chapters = len(chapters)
|
||||
|
||||
# --- Voice marker splitting logic ---
|
||||
@@ -735,8 +730,6 @@ class ConversionThread(QThread):
|
||||
{"chapter": chapter[0], "start": 0.0, "end": 0.0}
|
||||
for chapter in chapters
|
||||
]
|
||||
# SRT numbering fix: use a global counter
|
||||
merged_srt_index = 1 # SRT numbering for merged file
|
||||
# Prepare output file/ffmpeg process for merged output
|
||||
merged_sink = sink_stack.enter_context(
|
||||
self._open_merged_sink(
|
||||
@@ -745,67 +738,26 @@ class ConversionThread(QThread):
|
||||
)
|
||||
)
|
||||
# Open merged subtitle file for incremental writing if needed
|
||||
merged_subtitle_file = None
|
||||
merged_subtitle_writer = None
|
||||
merged_subtitle_path = None
|
||||
if self.subtitle_mode != "Disabled":
|
||||
subtitle_format = getattr(self, "subtitle_format", "srt")
|
||||
file_extension = "ass" if "ass" in subtitle_format else "srt"
|
||||
merged_subtitle_path = (
|
||||
os.path.splitext(merged_out_path)[0] + f".{file_extension}"
|
||||
)
|
||||
# Default subtitle layout flags/strings so they exist regardless
|
||||
# of whether ASS-specific handling runs. This prevents runtime
|
||||
# errors when non-ASS formats (like SRT) are selected.
|
||||
is_centered = False
|
||||
is_narrow = False
|
||||
merged_subtitle_margin = ""
|
||||
merged_subtitle_alignment_tag = ""
|
||||
if "ass" in subtitle_format:
|
||||
merged_subtitle_file = open(
|
||||
merged_subtitle_path,
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
# Minimal ASS header
|
||||
merged_subtitle_file.write("[Script Info]\n")
|
||||
merged_subtitle_file.write("Title: Generated by Abogen\n")
|
||||
merged_subtitle_file.write("ScriptType: v4.00+\n\n")
|
||||
# Add style definitions for karaoke highlighting
|
||||
if self.subtitle_mode == "Sentence + Highlighting":
|
||||
merged_subtitle_file.write("[V4+ Styles]\n")
|
||||
merged_subtitle_file.write(
|
||||
"Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\n"
|
||||
)
|
||||
merged_subtitle_file.write(
|
||||
"Style: Default,Arial,24,&H00FFFFFF,&H00808080,&H00000000,&H00404040,0,0,0,0,100,100,0,0,3,2,0,5,10,10,10,1\n\n"
|
||||
)
|
||||
merged_subtitle_file.write("[Events]\n")
|
||||
merged_subtitle_file.write(
|
||||
"Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"
|
||||
)
|
||||
# Set margin/alignment for ASS
|
||||
is_centered = subtitle_format in (
|
||||
"ass_centered_wide",
|
||||
"ass_centered_narrow",
|
||||
)
|
||||
is_narrow = subtitle_format in (
|
||||
"ass_narrow",
|
||||
"ass_centered_narrow",
|
||||
)
|
||||
merged_subtitle_margin = "90" if is_narrow else ""
|
||||
merged_subtitle_alignment_tag = (
|
||||
f"{{\\an5}}" if is_centered else ""
|
||||
)
|
||||
else:
|
||||
merged_subtitle_file = open(
|
||||
merged_subtitle_path,
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
alignment = _subtitle_alignment_from_format(subtitle_format)
|
||||
merged_subtitle_writer = create_subtitle_writer(
|
||||
Path(merged_subtitle_path),
|
||||
file_extension,
|
||||
self.subtitle_mode,
|
||||
alignment=alignment,
|
||||
max_words=self.max_subtitle_words,
|
||||
)
|
||||
merged_subtitle_writer.open()
|
||||
else:
|
||||
merged_subtitle_path = None
|
||||
merged_subtitle_file = None
|
||||
merged_subtitle_writer = None
|
||||
else:
|
||||
# If not merging, set merged_sink and related variables to None
|
||||
merged_sink = None
|
||||
@@ -821,12 +773,11 @@ class ConversionThread(QThread):
|
||||
{"chapter": chapter[0], "start": 0.0, "end": 0.0}
|
||||
for chapter in chapters
|
||||
]
|
||||
srt_index = 1 # SRT numbering fix for chapter-only mode
|
||||
# Instead of processing the whole text, process by chapter
|
||||
for chapter_idx, (chapter_name, voice_segments) in enumerate(chapters, 1):
|
||||
chapter_out_path = None
|
||||
chapter_sink = None
|
||||
chapter_subtitle_file = None
|
||||
chapter_subtitle_writer = None
|
||||
chapter_subtitle_path = None
|
||||
if total_chapters > 1:
|
||||
self.log_updated.emit(
|
||||
@@ -844,18 +795,7 @@ class ConversionThread(QThread):
|
||||
|
||||
# Prepare per-chapter output file if needed
|
||||
if save_chapters_separately and total_chapters > 1:
|
||||
# First pass: keep alphanumeric, spaces, hyphens, and underscores
|
||||
sanitized = re.sub(r"[^\w\s\-]", "", chapter_name)
|
||||
# Replace multiple spaces/hyphens with single underscore
|
||||
sanitized = re.sub(r"[\s\-]+", "_", sanitized).strip("_")
|
||||
# Apply OS-specific sanitization
|
||||
sanitized = sanitize_name_for_os(sanitized, is_folder=False)
|
||||
# Limit length (leaving room for the chapter number prefix)
|
||||
MAX_LEN = 80
|
||||
if len(sanitized) > MAX_LEN:
|
||||
pos = sanitized[:MAX_LEN].rfind("_")
|
||||
sanitized = sanitized[: pos if pos > 0 else MAX_LEN].rstrip("_")
|
||||
chapter_filename = f"{chapter_idx:02d}_{sanitized}"
|
||||
chapter_filename = sanitize_filename_for_chapter(chapter_name, chapter_idx)
|
||||
chapter_out_path = os.path.join(
|
||||
chapters_out_dir,
|
||||
f"{chapter_filename}.{separate_chapters_format}",
|
||||
@@ -874,67 +814,26 @@ class ConversionThread(QThread):
|
||||
)
|
||||
)
|
||||
continue
|
||||
# Open chapter subtitle file for incremental writing if needed
|
||||
chapter_subtitle_file = None
|
||||
chapter_srt_index = (
|
||||
1 # Initialize SRT numbering for this chapter file
|
||||
)
|
||||
if self.subtitle_mode != "Disabled":
|
||||
subtitle_format = getattr(self, "subtitle_format", "srt")
|
||||
file_extension = "ass" if "ass" in subtitle_format else "srt"
|
||||
chapter_subtitle_path = os.path.join(
|
||||
chapters_out_dir, f"{chapter_filename}.{file_extension}"
|
||||
)
|
||||
# Ensure these variables exist even when not using ASS so
|
||||
# later code can safely reference them.
|
||||
is_centered = False
|
||||
is_narrow = False
|
||||
chapter_subtitle_margin = ""
|
||||
chapter_subtitle_alignment_tag = ""
|
||||
# Open the chapter subtitle file for writing for both SRT and ASS
|
||||
chapter_subtitle_file = open(
|
||||
chapter_subtitle_path,
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
if "ass" in subtitle_format:
|
||||
# Minimal ASS header
|
||||
chapter_subtitle_file.write("[Script Info]\n")
|
||||
chapter_subtitle_file.write("Title: Generated by Abogen\n")
|
||||
chapter_subtitle_file.write("ScriptType: v4.00+\n\n")
|
||||
|
||||
# Add style definitions for karaoke highlighting
|
||||
if self.subtitle_mode == "Sentence + Highlighting":
|
||||
chapter_subtitle_file.write("[V4+ Styles]\n")
|
||||
chapter_subtitle_file.write(
|
||||
"Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\n"
|
||||
)
|
||||
chapter_subtitle_file.write(
|
||||
"Style: Default,Arial,24,&H00FFFFFF,&H00808080,&H00000000,&H00404040,0,0,0,0,100,100,0,0,3,2,0,5,10,10,10,1\n\n"
|
||||
)
|
||||
|
||||
chapter_subtitle_file.write("[Events]\n")
|
||||
chapter_subtitle_file.write(
|
||||
"Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"
|
||||
)
|
||||
is_centered = subtitle_format in (
|
||||
"ass_centered_wide",
|
||||
"ass_centered_narrow",
|
||||
)
|
||||
is_narrow = subtitle_format in (
|
||||
"ass_narrow",
|
||||
"ass_centered_narrow",
|
||||
)
|
||||
chapter_subtitle_margin = "90" if is_narrow else ""
|
||||
chapter_subtitle_alignment_tag = (
|
||||
f"{{\\an5}}" if is_centered else ""
|
||||
alignment = _subtitle_alignment_from_format(subtitle_format)
|
||||
chapter_subtitle_writer = create_subtitle_writer(
|
||||
Path(chapter_subtitle_path),
|
||||
file_extension,
|
||||
self.subtitle_mode,
|
||||
alignment=alignment,
|
||||
max_words=self.max_subtitle_words,
|
||||
)
|
||||
chapter_subtitle_writer.open()
|
||||
else:
|
||||
chapter_subtitle_file = None
|
||||
chapter_subtitle_writer = None
|
||||
else:
|
||||
chapter_subtitle_path = None
|
||||
chapter_subtitle_file = None
|
||||
chapter_subtitle_writer = None
|
||||
|
||||
|
||||
# Process each voice segment within the chapter
|
||||
@@ -1125,31 +1024,9 @@ class ConversionThread(QThread):
|
||||
self.max_subtitle_words,
|
||||
fallback_end_time=chunk_start + chunk_dur,
|
||||
)
|
||||
if merged_subtitle_file:
|
||||
subtitle_format = getattr(
|
||||
self, "subtitle_format", "srt"
|
||||
)
|
||||
if "ass" in subtitle_format:
|
||||
if merged_subtitle_writer:
|
||||
for start, end, text in new_entries:
|
||||
start_time = _format_timestamp(start, ass=True)
|
||||
end_time = _format_timestamp(end, ass=True)
|
||||
# Use karaoke effect for highlighting mode
|
||||
effect = (
|
||||
"karaoke"
|
||||
if self.subtitle_mode
|
||||
== "Sentence + Highlighting"
|
||||
else ""
|
||||
)
|
||||
merged_subtitle_file.write(
|
||||
f"Dialogue: 0,{start_time},{end_time},Default,,{merged_subtitle_margin},{merged_subtitle_margin},0,{effect},{merged_subtitle_alignment_tag}{text}\n"
|
||||
)
|
||||
else:
|
||||
for entry in new_entries:
|
||||
start, end, text = entry
|
||||
merged_subtitle_file.write(
|
||||
f"{merged_srt_index}\n{_format_timestamp(start)} --> {_format_timestamp(end)}\n{text}\n\n"
|
||||
)
|
||||
merged_srt_index += 1
|
||||
merged_subtitle_writer.write_entry(start, end, text)
|
||||
# Per-chapter subtitle processing for both file and sink
|
||||
if chapter_sink:
|
||||
new_chapter_entries = []
|
||||
@@ -1159,31 +1036,9 @@ class ConversionThread(QThread):
|
||||
self.max_subtitle_words,
|
||||
fallback_end_time=chapter_current_time + chunk_dur,
|
||||
)
|
||||
if chapter_subtitle_file:
|
||||
subtitle_format = getattr(
|
||||
self, "subtitle_format", "srt"
|
||||
)
|
||||
if "ass" in subtitle_format:
|
||||
if chapter_subtitle_writer:
|
||||
for start, end, text in new_chapter_entries:
|
||||
start_time = _format_timestamp(start, ass=True)
|
||||
end_time = _format_timestamp(end, ass=True)
|
||||
# Use karaoke effect for highlighting mode
|
||||
effect = (
|
||||
"karaoke"
|
||||
if self.subtitle_mode
|
||||
== "Sentence + Highlighting"
|
||||
else ""
|
||||
)
|
||||
chapter_subtitle_file.write(
|
||||
f"Dialogue: 0,{start_time},{end_time},Default,,{chapter_subtitle_margin},{chapter_subtitle_margin},0,{effect},{chapter_subtitle_alignment_tag}{text}\n"
|
||||
)
|
||||
else:
|
||||
for entry in new_chapter_entries:
|
||||
start, end, text = entry
|
||||
chapter_subtitle_file.write(
|
||||
f"{chapter_srt_index}\n{_format_timestamp(start)} --> {_format_timestamp(end)}\n{text}\n\n"
|
||||
)
|
||||
chapter_srt_index += 1
|
||||
chapter_subtitle_writer.write_entry(start, end, text)
|
||||
if merge_chapters_at_end:
|
||||
current_time += chunk_dur
|
||||
if chapter_sink:
|
||||
@@ -1228,9 +1083,9 @@ class ConversionThread(QThread):
|
||||
# Finalize chapter file for ffmpeg formats
|
||||
if chapter_sink:
|
||||
self.log_updated.emit(("\nProcessing chapter audio...", "grey"))
|
||||
# Close chapter subtitle file if open
|
||||
if chapter_subtitle_file:
|
||||
chapter_subtitle_file.close()
|
||||
# Close chapter subtitle writer if open
|
||||
if chapter_subtitle_writer:
|
||||
chapter_subtitle_writer.close()
|
||||
if (
|
||||
save_chapters_separately
|
||||
and total_chapters > 1
|
||||
@@ -1255,74 +1110,36 @@ class ConversionThread(QThread):
|
||||
self.log_updated.emit(("\nFinalizing audio. Please wait...", "grey"))
|
||||
merged_sink.close()
|
||||
if self.output_format == "m4b":
|
||||
# Add chapters via fast post-processing
|
||||
# Add chapters via ExportService (unified with WebUI)
|
||||
if total_chapters > 1:
|
||||
chapters_info_path = f"{base_filepath_no_ext}_chapters.txt"
|
||||
with open(chapters_info_path, "w", encoding="utf-8") as f:
|
||||
f.write(";FFMETADATA1\n")
|
||||
for chapter in chapters_time:
|
||||
chapter_title = chapter["chapter"].replace("=", "\\=")
|
||||
f.write(f"[CHAPTER]\n")
|
||||
f.write(f"TIMEBASE=1/1000\n")
|
||||
f.write(f"START={int(chapter['start']*1000)}\n")
|
||||
f.write(f"END={int(chapter['end']*1000)}\n")
|
||||
f.write(f"title={chapter_title}\n\n")
|
||||
# Fast mux chapters into m4b (write to temp file, then replace original)
|
||||
static_ffmpeg.add_paths()
|
||||
orig_path = merged_out_path
|
||||
root, ext = os.path.splitext(orig_path)
|
||||
tmp_path = root + ".tmp" + ext
|
||||
metadata_options, cover_path = (
|
||||
self._extract_and_add_metadata_tags_to_ffmpeg_cmd()
|
||||
export_svc = ExportService()
|
||||
metadata_text = read_text_for_metadata(
|
||||
file_path=self.file_name,
|
||||
is_direct_text=self.is_direct_text,
|
||||
direct_text=self.file_name if self.is_direct_text else None,
|
||||
)
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-i",
|
||||
orig_path,
|
||||
"-i",
|
||||
chapters_info_path,
|
||||
]
|
||||
if cover_path and os.path.exists(cover_path):
|
||||
cmd.extend(
|
||||
[
|
||||
"-i",
|
||||
cover_path,
|
||||
"-map",
|
||||
"0:a",
|
||||
"-map",
|
||||
"2",
|
||||
"-c:v",
|
||||
"copy",
|
||||
"-disposition:v",
|
||||
"attached_pic",
|
||||
metadata = extract_metadata_from_text(metadata_text) if metadata_text else {}
|
||||
# Convert cover_path from metadata to Path if present
|
||||
cover_path_raw = metadata.pop("cover_path", None)
|
||||
cover_path = Path(cover_path_raw) if cover_path_raw and os.path.exists(cover_path_raw) else None
|
||||
# Build chapters list for ExportService
|
||||
chapters_for_export = [
|
||||
{"title": ch["chapter"], "start": ch["start"], "end": ch["end"]}
|
||||
for ch in chapters_time
|
||||
]
|
||||
export_svc.embed_m4b_metadata(
|
||||
audio_path=Path(merged_out_path),
|
||||
metadata=metadata,
|
||||
chapters=chapters_for_export,
|
||||
cover_path=cover_path,
|
||||
log_callback=lambda msg, level="info": self.log_updated.emit((msg, "grey" if level == "info" else "orange")),
|
||||
)
|
||||
else:
|
||||
cmd.extend(["-map", "0:a"])
|
||||
|
||||
cmd.extend(
|
||||
[
|
||||
"-map_metadata",
|
||||
"1",
|
||||
"-map_chapters",
|
||||
"1",
|
||||
"-c:a",
|
||||
"copy",
|
||||
]
|
||||
)
|
||||
cmd += metadata_options
|
||||
cmd.append(tmp_path)
|
||||
proc = create_process(cmd)
|
||||
proc.wait()
|
||||
os.replace(tmp_path, orig_path)
|
||||
os.remove(chapters_info_path)
|
||||
elif self.output_format in ["opus"]:
|
||||
merged_sink.close()
|
||||
self.progress_updated.emit(100, "00:00:00")
|
||||
# Close merged subtitle file if open
|
||||
if merged_subtitle_file:
|
||||
merged_subtitle_file.close()
|
||||
# Close merged subtitle writer if open
|
||||
if merged_subtitle_writer:
|
||||
merged_subtitle_writer.close()
|
||||
# Subtitle and final message logic
|
||||
if merge_chapters_at_end:
|
||||
if self.subtitle_mode != "Disabled":
|
||||
@@ -1430,35 +1247,20 @@ class ConversionThread(QThread):
|
||||
).__enter__()
|
||||
|
||||
# Always generate subtitles for subtitle input files
|
||||
subtitle_file, subtitle_path = None, None
|
||||
subtitle_writer = None
|
||||
subtitle_path = None
|
||||
subtitle_format = getattr(self, "subtitle_format", "srt")
|
||||
file_extension = "ass" if "ass" in subtitle_format else "srt"
|
||||
subtitle_path = f"{base_filepath_no_ext}.{file_extension}"
|
||||
subtitle_file = open(subtitle_path, "w", encoding="utf-8", errors="replace")
|
||||
|
||||
if "ass" in subtitle_format:
|
||||
# Write ASS header
|
||||
subtitle_file.write(
|
||||
"[Script Info]\nTitle: Generated by Abogen\nScriptType: v4.00+\n\n"
|
||||
alignment_str = _subtitle_alignment_from_format(subtitle_format)
|
||||
subtitle_writer = create_subtitle_writer(
|
||||
Path(subtitle_path),
|
||||
file_extension,
|
||||
self.subtitle_mode,
|
||||
alignment=alignment_str,
|
||||
max_words=self.max_subtitle_words,
|
||||
)
|
||||
if self.subtitle_mode == "Sentence + Highlighting":
|
||||
subtitle_file.write(
|
||||
"[V4+ Styles]\nFormat: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\n"
|
||||
)
|
||||
subtitle_file.write(
|
||||
"Style: Default,Arial,24,&H00FFFFFF,&H00808080,&H00000000,&H00404040,0,0,0,0,100,100,0,0,3,2,0,5,10,10,10,1\n\n"
|
||||
)
|
||||
subtitle_file.write(
|
||||
"[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"
|
||||
)
|
||||
|
||||
is_narrow = subtitle_format in ("ass_narrow", "ass_centered_narrow")
|
||||
is_centered = subtitle_format in (
|
||||
"ass_centered_wide",
|
||||
"ass_centered_narrow",
|
||||
)
|
||||
margin = "90" if is_narrow else ""
|
||||
alignment = "{\\an5}" if is_centered else ""
|
||||
subtitle_writer.open()
|
||||
|
||||
# Load voice
|
||||
loaded_voice = resolve_voice(self.voice, tts, self.use_gpu)
|
||||
@@ -1472,12 +1274,11 @@ class ConversionThread(QThread):
|
||||
|
||||
# Process each subtitle and mix into buffer
|
||||
self.etr_start_time = time.time()
|
||||
srt_index = 1
|
||||
|
||||
for idx, (start_time, end_time, text) in enumerate(subtitles, 1):
|
||||
if self.cancel_requested:
|
||||
if subtitle_file:
|
||||
subtitle_file.close()
|
||||
if subtitle_writer:
|
||||
subtitle_writer.close()
|
||||
self.conversion_finished.emit("Cancelled", None)
|
||||
return
|
||||
|
||||
@@ -1541,8 +1342,8 @@ class ConversionThread(QThread):
|
||||
audio_chunks = [r.audio for r in tts_results]
|
||||
|
||||
if self.cancel_requested:
|
||||
if subtitle_file:
|
||||
subtitle_file.close()
|
||||
if subtitle_writer:
|
||||
subtitle_writer.close()
|
||||
self.conversion_finished.emit("Cancelled", None)
|
||||
return
|
||||
|
||||
@@ -1676,26 +1477,13 @@ class ConversionThread(QThread):
|
||||
audio_buffer = mix_audio(audio_buffer, full_audio, start_sample)
|
||||
|
||||
# Write subtitle
|
||||
if subtitle_file:
|
||||
if "ass" in subtitle_format:
|
||||
effect = (
|
||||
"karaoke"
|
||||
if self.subtitle_mode == "Sentence + Highlighting"
|
||||
else ""
|
||||
)
|
||||
ass_text = (
|
||||
if subtitle_writer:
|
||||
display_text = (
|
||||
processed_text
|
||||
if replace_nl
|
||||
if "ass" in subtitle_format or replace_nl
|
||||
else processed_text.replace("\n", "\\N")
|
||||
)
|
||||
subtitle_file.write(
|
||||
f"Dialogue: 0,{_format_timestamp(start_time, ass=True)},{_format_timestamp(end_time, ass=True)},Default,,{margin},{margin},0,{effect},{alignment}{ass_text}\n"
|
||||
)
|
||||
else:
|
||||
subtitle_file.write(
|
||||
f"{srt_index}\n{_format_timestamp(start_time)} --> {_format_timestamp(end_time)}\n{processed_text}\n\n"
|
||||
)
|
||||
srt_index += 1
|
||||
subtitle_writer.write_entry(start_time, end_time, display_text)
|
||||
|
||||
# Update progress
|
||||
percent = min(int(idx / len(subtitles) * 100), 99)
|
||||
@@ -1719,8 +1507,8 @@ class ConversionThread(QThread):
|
||||
merged_sink.write(to_float32(audio_buffer))
|
||||
merged_sink.close()
|
||||
|
||||
if subtitle_file:
|
||||
subtitle_file.close()
|
||||
if subtitle_writer:
|
||||
subtitle_writer.close()
|
||||
|
||||
self.progress_updated.emit(100, "00:00:00")
|
||||
result_msg = f"\nAudio saved to: {merged_out_path}" + (
|
||||
@@ -1732,8 +1520,8 @@ class ConversionThread(QThread):
|
||||
try:
|
||||
if "merged_sink" in locals() and merged_sink:
|
||||
merged_sink.close()
|
||||
if "subtitle_file" in locals() and subtitle_file:
|
||||
subtitle_file.close()
|
||||
if "subtitle_writer" in locals() and subtitle_writer:
|
||||
subtitle_writer.close()
|
||||
except:
|
||||
pass
|
||||
self.log_updated.emit((f"Error processing subtitle file: {str(e)}", "red"))
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Tests for sanitize_filename_for_chapter in domain/output_paths.py"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abogen.domain.output_paths import sanitize_filename_for_chapter
|
||||
|
||||
|
||||
class TestSanitizeFilenameForChapter:
|
||||
def test_basic_title(self):
|
||||
result = sanitize_filename_for_chapter("The Beginning", 1)
|
||||
assert result == "01_The_Beginning"
|
||||
|
||||
def test_special_chars_removed(self):
|
||||
result = sanitize_filename_for_chapter("Ch. 1: Hello!", 2)
|
||||
assert result.startswith("02_")
|
||||
assert "Ch" in result
|
||||
assert "Hello" in result
|
||||
|
||||
def test_empty_title_uses_fallback(self):
|
||||
result = sanitize_filename_for_chapter("", 3)
|
||||
assert result == "03_chapter_03"
|
||||
|
||||
def test_index_prefix_zero_padded(self):
|
||||
result = sanitize_filename_for_chapter("Test", 10)
|
||||
assert result.startswith("10_")
|
||||
|
||||
def test_long_title_truncated_at_word_boundary(self):
|
||||
long_title = "a" * 50 + "_" + "b" * 50
|
||||
result = sanitize_filename_for_chapter(long_title, 1, max_len=60)
|
||||
# Should truncate at word boundary
|
||||
suffix = result[3:] # Remove "01_"
|
||||
assert len(suffix) <= 60
|
||||
|
||||
def test_custom_max_len(self):
|
||||
result = sanitize_filename_for_chapter("Hello World", 1, max_len=5)
|
||||
suffix = result[3:] # Remove "01_"
|
||||
assert len(suffix) <= 5
|
||||
|
||||
def test_hyphens_and_spaces_collapsed(self):
|
||||
result = sanitize_filename_for_chapter("mid-night story", 1)
|
||||
assert result == "01_mid_night_story"
|
||||
|
||||
def test_whitespace_collapsed(self):
|
||||
result = sanitize_filename_for_chapter("hello world", 1)
|
||||
assert "hello_world" in result
|
||||
|
||||
def test_leading_trailing_underscores_stripped(self):
|
||||
result = sanitize_filename_for_chapter(" hello ", 1)
|
||||
assert result == "01_hello"
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Tests for domain/text_chapters.py"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abogen.domain.text_chapters import parse_chapters_from_text
|
||||
|
||||
|
||||
class TestParseChaptersFromText:
|
||||
def test_no_markers_returns_single_chapter(self):
|
||||
result = parse_chapters_from_text("Hello world")
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == "text"
|
||||
assert result[0][1] == "Hello world"
|
||||
|
||||
def test_no_markers_custom_default_title(self):
|
||||
result = parse_chapters_from_text("Hello", default_title="intro")
|
||||
assert result[0][0] == "intro"
|
||||
|
||||
def test_single_marker(self):
|
||||
text = "<<CHAPTER_MARKER:Chapter 1>>\nSome text here"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == "Chapter 1"
|
||||
assert result[0][1] == "Some text here"
|
||||
|
||||
def test_multiple_markers(self):
|
||||
text = (
|
||||
"<<CHAPTER_MARKER:Chapter 1>>\nText 1\n"
|
||||
"<<CHAPTER_MARKER:Chapter 2>>\nText 2\n"
|
||||
)
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 2
|
||||
assert result[0][0] == "Chapter 1"
|
||||
assert result[0][1] == "Text 1"
|
||||
assert result[1][0] == "Chapter 2"
|
||||
assert result[1][1] == "Text 2"
|
||||
|
||||
def test_intro_preserved_before_first_marker(self):
|
||||
text = "Introduction text\n<<CHAPTER_MARKER:Chapter 1>>\nMain text"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 2
|
||||
assert result[0][0] == "Introduction"
|
||||
assert result[0][1] == "Introduction text"
|
||||
assert result[1][0] == "Chapter 1"
|
||||
|
||||
def test_empty_intro_not_added(self):
|
||||
text = "<<CHAPTER_MARKER:Chapter 1>>\nText"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == "Chapter 1"
|
||||
|
||||
def test_clean_text_applied_by_default(self):
|
||||
text = "<<CHAPTER_MARKER:Ch 1>>\n some messy text "
|
||||
result = parse_chapters_from_text(text, clean=True)
|
||||
# clean_text normalizes whitespace
|
||||
assert "some" in result[0][1]
|
||||
assert "messy" in result[0][1]
|
||||
|
||||
def test_clean_disabled(self):
|
||||
text = "<<CHAPTER_MARKER:Ch 1>>\n some text "
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert result[0][1] == "some text"
|
||||
|
||||
def test_empty_marker_title_uses_default(self):
|
||||
text = "<<CHAPTER_MARKER:>>\nSome text"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert result[0][0] == "text"
|
||||
|
||||
def test_case_insensitive_markers(self):
|
||||
text = "<<chapter_marker:Chapter 1>>\nText"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == "Chapter 1"
|
||||
|
||||
def test_empty_text(self):
|
||||
result = parse_chapters_from_text("")
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == "text"
|
||||
assert result[0][1] == ""
|
||||
|
||||
def test_only_markers_no_text(self):
|
||||
text = "<<CHAPTER_MARKER:Ch 1>><<CHAPTER_MARKER:Ch 2>>"
|
||||
result = parse_chapters_from_text(text, clean=False)
|
||||
assert len(result) == 2
|
||||
assert result[0][0] == "Ch 1"
|
||||
assert result[1][0] == "Ch 2"
|
||||
Reference in New Issue
Block a user