refactor: move voice marker functions to domain/voice_markers.py

This commit is contained in:
Artem Akymenko
2026-07-24 19:17:38 +03:00
parent 332934c0cf
commit d1a84cfb8b
3 changed files with 126 additions and 119 deletions
+2 -2
View File
@@ -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(
+117
View File
@@ -0,0 +1,117 @@
"""Voice marker parsing and text splitting.
Handles <<VOICE:name>> 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:[^>]*>>")
_VOICE_MARKER_SEARCH_PATTERN = re.compile(r"<<VOICE:(.*?)>>")
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 <<VOICE:name>> 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
+6 -116
View File
@@ -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 <<VOICE:name>> 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()
# 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,
)
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