From d1a84cfb8b2488c6d1415333acbb9a520a33190c Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 23 Jul 2026 08:18:55 +0000 Subject: [PATCH] refactor: move voice marker functions to domain/voice_markers.py --- abogen/application/conversion_planner.py | 4 +- abogen/domain/voice_markers.py | 117 +++++++++++++++++++++ abogen/subtitle_utils.py | 124 ++--------------------- 3 files changed, 126 insertions(+), 119 deletions(-) create mode 100644 abogen/domain/voice_markers.py diff --git a/abogen/application/conversion_planner.py b/abogen/application/conversion_planner.py index 96d63fd..1763ab5 100644 --- a/abogen/application/conversion_planner.py +++ b/abogen/application/conversion_planner.py @@ -25,7 +25,7 @@ from abogen.domain.file_type import auto_select_relevant_chapters from abogen.domain.intro_outro import resolve_intro, resolve_outro from abogen.domain.metadata_extraction import extract_metadata_for_file from abogen.domain.metadata_merge import merge_metadata -from abogen.subtitle_utils import split_text_by_voice_markers +from abogen.domain.voice_markers import split_text_by_voice_markers def build_conversion_plan(request: ConversionRequest) -> ConversionPlan: @@ -274,7 +274,7 @@ def _build_segments( # Check for voice markers (PyQt style) # Detect markers even if validation fails (voice names may not be loaded yet) - from abogen.subtitle_utils import _VOICE_MARKER_SEARCH_PATTERN + from abogen.domain.voice_markers import _VOICE_MARKER_SEARCH_PATTERN has_voice_markers = bool(_VOICE_MARKER_SEARCH_PATTERN.search(body_text)) voice_segments, last_voice, valid_count, invalid_count = split_text_by_voice_markers( diff --git a/abogen/domain/voice_markers.py b/abogen/domain/voice_markers.py new file mode 100644 index 0000000..5045daa --- /dev/null +++ b/abogen/domain/voice_markers.py @@ -0,0 +1,117 @@ +"""Voice marker parsing and text splitting. + +Handles <> markers in text, splitting text into voice-specific +segments. This is domain logic about text segmentation by voice, not subtitle +processing. +""" + +from __future__ import annotations + +import re +from typing import List, Tuple + +_VOICE_MARKER_PATTERN = re.compile(r"<]*>>") +_VOICE_MARKER_SEARCH_PATTERN = re.compile(r"<>") + + +def validate_voice_name(voice_name: str) -> Tuple[bool, str | None]: + """Validate voice name against available voices (case-insensitive). + + Handles both single voices and formulas like 'af_heart*0.5 + am_echo*0.5'. + + Returns: + Tuple of (is_valid, invalid_voice_name): + - is_valid: True if all voices in the name/formula are valid + - invalid_voice_name: The first invalid voice found, or None if all valid + """ + from abogen.tts_plugin.utils import get_voices + + voice_lookup_lower = {v.lower() for v in get_voices("kokoro")} + voice_name = voice_name.strip() + + if "*" in voice_name: + voices = voice_name.split("+") + for term in voices: + if "*" in term: + base_voice = term.split("*")[0].strip() + if base_voice.lower() not in voice_lookup_lower: + return False, base_voice + return True, None + else: + if voice_name.lower() not in voice_lookup_lower: + return False, voice_name + return True, None + + +def split_text_by_voice_markers( + text: str, default_voice: str +) -> Tuple[List[Tuple[str, str]], str, int, int]: + """Split text by voice markers, returning list of (voice, text) tuples. + + Returns the last voice used so it can persist across chapters. + Voice names are normalized to lowercase to match canonical voice names. + + Args: + text: Text potentially containing <> markers + default_voice: Voice to use if no markers found or before first marker + + Returns: + Tuple of (segments_list, last_voice_used, valid_count, invalid_count): + - segments_list: List of (voice_name, segment_text) tuples + - last_voice_used: The voice that should continue into next chapter + - valid_count: Number of valid voice markers processed + - invalid_count: Number of invalid voice markers skipped + """ + from abogen.tts_plugin.utils import get_voices + + voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text)) + + if not voice_splits: + return [(default_voice, text)], default_voice, 0, 0 + + segments: List[Tuple[str, str]] = [] + current_voice = default_voice + valid_markers = 0 + invalid_markers = 0 + + first_start = voice_splits[0].start() + if first_start > 0: + intro_text = text[:first_start].strip() + if intro_text: + segments.append((current_voice, intro_text)) + + for idx, match in enumerate(voice_splits): + voice_name = match.group(1).strip() + start = match.end() + end = voice_splits[idx + 1].start() if idx + 1 < len(voice_splits) else len(text) + segment_text = text[start:end].strip() + + is_valid, invalid_voice = validate_voice_name(voice_name) + if is_valid: + if "*" in voice_name: + normalized_parts = [] + for part in voice_name.split("+"): + part = part.strip() + if "*" in part: + voice_part, weight = part.split("*", 1) + voice_part_lower = voice_part.strip().lower() + canonical_voice = next( + (v for v in get_voices("kokoro") if v.lower() == voice_part_lower), + voice_part.strip() + ) + normalized_parts.append(f"{canonical_voice}*{weight.strip()}") + current_voice = " + ".join(normalized_parts) + else: + voice_name_lower = voice_name.lower() + current_voice = next( + (v for v in get_voices("kokoro") if v.lower() == voice_name_lower), + voice_name + ) + valid_markers += 1 + else: + invalid_markers += 1 + + if segment_text: + segments.append((current_voice, segment_text)) + + return segments, current_voice, valid_markers, invalid_markers diff --git a/abogen/subtitle_utils.py b/abogen/subtitle_utils.py index 489ed24..c314524 100644 --- a/abogen/subtitle_utils.py +++ b/abogen/subtitle_utils.py @@ -465,120 +465,10 @@ def sanitize_name_for_os(name, is_folder=True): return sanitized -def validate_voice_name(voice_name): - """Validate voice name against available voices (case-insensitive). - Handles both single voices and formulas like 'af_heart*0.5 + am_echo*0.5'. - - Args: - voice_name: Voice name or formula string to validate - - Returns: - Tuple of (is_valid, invalid_voice_name): - - is_valid: True if all voices in the name/formula are valid - - invalid_voice_name: The first invalid voice found, or None if all valid - """ - from abogen.tts_plugin.utils import get_voices - - # Create case-insensitive lookup set (done once per call) - voice_lookup_lower = {v.lower() for v in get_voices("kokoro")} - voice_name = voice_name.strip() - - # Check if it's a formula (contains *) - if "*" in voice_name: - # Extract voice names from formula - voices = voice_name.split("+") - for term in voices: - if "*" in term: - base_voice = term.split("*")[0].strip() - # Case-insensitive comparison - if base_voice.lower() not in voice_lookup_lower: - return False, base_voice - return True, None - else: - # Single voice - case-insensitive comparison - if voice_name.lower() not in voice_lookup_lower: - return False, voice_name - return True, None - - -def split_text_by_voice_markers(text, default_voice): - """Split text by voice markers, returning list of (voice, text) tuples. - - IMPORTANT: Returns the last voice used so it can persist across chapters. - Voice names are normalized to lowercase to match canonical voice names. - - Args: - text: Text potentially containing <> markers - default_voice: Voice to use if no markers found or before first marker - - Returns: - Tuple of (segments_list, last_voice_used, valid_count, invalid_count): - - segments_list: List of (voice_name, segment_text) tuples - - last_voice_used: The voice that should continue into next chapter - - valid_count: Number of valid voice markers processed - - invalid_count: Number of invalid voice markers skipped - """ - from abogen.tts_plugin.utils import get_voices - - voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text)) - - if not voice_splits: - # No voice markers, return entire text with default voice - return [(default_voice, text)], default_voice, 0, 0 - - segments = [] - current_voice = default_voice - valid_markers = 0 - invalid_markers = 0 - - # Text before first marker uses default voice - first_start = voice_splits[0].start() - if first_start > 0: - intro_text = text[:first_start].strip() - if intro_text: - segments.append((current_voice, intro_text)) - - # Process each voice marker - for idx, match in enumerate(voice_splits): - voice_name = match.group(1).strip() - start = match.end() - end = voice_splits[idx + 1].start() if idx + 1 < len(voice_splits) else len(text) - segment_text = text[start:end].strip() - - # Validate voice name - is_valid, invalid_voice = validate_voice_name(voice_name) - if is_valid: - # Normalize to lowercase to match canonical form - # Handle both single voices and formulas - if "*" in voice_name: - # Normalize each voice in the formula - normalized_parts = [] - for part in voice_name.split("+"): - part = part.strip() - if "*" in part: - voice_part, weight = part.split("*", 1) - # Find the canonical (lowercase) voice name - voice_part_lower = voice_part.strip().lower() - canonical_voice = next( - (v for v in get_voices("kokoro") if v.lower() == voice_part_lower), - voice_part.strip() - ) - normalized_parts.append(f"{canonical_voice}*{weight.strip()}") - current_voice = " + ".join(normalized_parts) - else: - # Find the canonical (lowercase) voice name - voice_name_lower = voice_name.lower() - current_voice = next( - (v for v in get_voices("kokoro") if v.lower() == voice_name_lower), - voice_name - ) - valid_markers += 1 - else: - # Invalid voice - stay with previous voice - invalid_markers += 1 - - if segment_text: - segments.append((current_voice, segment_text)) - - # Return segments, last voice, and counts - return segments, current_voice, valid_markers, invalid_markers +# Backward-compatible re-exports — canonical location is domain/voice_markers.py +from abogen.domain.voice_markers import ( # noqa: E402, F401 + validate_voice_name, + split_text_by_voice_markers, + _VOICE_MARKER_PATTERN, + _VOICE_MARKER_SEARCH_PATTERN, +)