mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 19:50:59 +02:00
Compare commits
2
Commits
ffac4a4da9
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
08e2ee8b85 | ||
|
|
be74c69507 |
@@ -5,10 +5,10 @@ from __future__ import annotations
|
||||
from abogen.domain.enums import Language, SubtitleMode
|
||||
|
||||
# Canonical punctuation sets covering all supported scripts:
|
||||
# ASCII (. ! ?), Arabic ؟, CJK (。!?), Devanagari ।
|
||||
PUNCTUATION_SENTENCE = r".!?؟。!?।"
|
||||
# ASCII (. ! ?), ellipsis (…), Arabic ؟, CJK (。!?), Devanagari ।
|
||||
PUNCTUATION_SENTENCE = r".!?…؟。!?।"
|
||||
# Commas: ASCII , CJK fullwidth ,CJK ideographic 、
|
||||
PUNCTUATION_SENTENCE_COMMA = r".!?,?。!?،,、।"
|
||||
PUNCTUATION_SENTENCE_COMMA = r".!?…,?。!?،,、।"
|
||||
PUNCTUATION_COMMAS = ",,、"
|
||||
|
||||
|
||||
|
||||
@@ -13,6 +13,34 @@ from typing import List, Optional, Tuple
|
||||
from abogen.domain.enums import Language, SubtitleMode
|
||||
from abogen.domain.split_pattern import PUNCTUATION_SENTENCE, PUNCTUATION_SENTENCE_COMMA
|
||||
|
||||
_CLOSING_DELIMS = "\"\"\"\"'\"”’»›)]}」』"
|
||||
|
||||
|
||||
def _is_sentence_boundary(
|
||||
token: dict,
|
||||
current_sentence: List[dict],
|
||||
separator: str,
|
||||
) -> bool:
|
||||
"""Check whether token ends a sentence, considering closing quotes and brackets."""
|
||||
ws = token.get("whitespace", "") or ""
|
||||
if not ws:
|
||||
return False
|
||||
|
||||
# For Line mode, a newline in whitespace or text marks line boundary
|
||||
if separator == r"\n":
|
||||
return "\n" in ws or "\n" in str(token.get("text", ""))
|
||||
|
||||
text = str(token.get("text", ""))
|
||||
if re.search(rf"{separator}[{re.escape(_CLOSING_DELIMS)}]*$", text):
|
||||
return True
|
||||
|
||||
if len(current_sentence) >= 2 and text and all(c in _CLOSING_DELIMS for c in text):
|
||||
prev_text = str(current_sentence[-2].get("text", ""))
|
||||
if re.search(rf"{separator}$", prev_text):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def process_subtitle_tokens(
|
||||
tokens_with_timestamps: List[dict],
|
||||
@@ -42,37 +70,55 @@ def process_subtitle_tokens(
|
||||
if not tokens_with_timestamps:
|
||||
return
|
||||
|
||||
if not isinstance(language, Language):
|
||||
try:
|
||||
language = Language.from_str(str(language))
|
||||
except ValueError:
|
||||
language = Language.EN_US
|
||||
|
||||
if isinstance(subtitle_mode, SubtitleMode):
|
||||
subtitle_mode_str = subtitle_mode.value
|
||||
else:
|
||||
subtitle_mode_str = str(subtitle_mode)
|
||||
|
||||
processed_tokens = tokens_with_timestamps
|
||||
|
||||
# For English with spaCy enabled and sentence-based modes, use spaCy for sentence boundaries
|
||||
# spaCy is disabled when subtitle mode is "Disabled" or "Line"
|
||||
use_spacy_for_english = (
|
||||
use_spacy_segmentation
|
||||
and subtitle_mode not in [SubtitleMode.DISABLED, SubtitleMode.LINE]
|
||||
and subtitle_mode_str not in [SubtitleMode.DISABLED.value, SubtitleMode.LINE.value, "Disabled", "Line"]
|
||||
and language in [Language.EN_US, Language.EN_GB]
|
||||
and subtitle_mode in [SubtitleMode.SENTENCE, SubtitleMode.SENTENCE_COMMA]
|
||||
and subtitle_mode_str in [SubtitleMode.SENTENCE.value, SubtitleMode.SENTENCE_COMMA.value, "Sentence", "Sentence + Comma"]
|
||||
)
|
||||
|
||||
if subtitle_mode == SubtitleMode.SENTENCE_HIGHLIGHT:
|
||||
if subtitle_mode_str in (SubtitleMode.SENTENCE_HIGHLIGHT.value, "Sentence + Highlighting"):
|
||||
_process_karaoke_highlighting(
|
||||
processed_tokens, subtitle_entries, max_subtitle_words, fallback_end_time
|
||||
)
|
||||
elif subtitle_mode in [SubtitleMode.SENTENCE, SubtitleMode.SENTENCE_COMMA, SubtitleMode.LINE]:
|
||||
if use_spacy_for_english and subtitle_mode != SubtitleMode.LINE:
|
||||
elif subtitle_mode_str in [
|
||||
SubtitleMode.SENTENCE.value,
|
||||
SubtitleMode.SENTENCE_COMMA.value,
|
||||
SubtitleMode.LINE.value,
|
||||
"Sentence",
|
||||
"Sentence + Comma",
|
||||
"Line",
|
||||
]:
|
||||
if use_spacy_for_english and subtitle_mode_str not in (SubtitleMode.LINE.value, "Line"):
|
||||
_process_spacy_sentences(
|
||||
processed_tokens, subtitle_entries, max_subtitle_words,
|
||||
subtitle_mode, language, fallback_end_time
|
||||
subtitle_mode_str, language, fallback_end_time
|
||||
)
|
||||
else:
|
||||
_process_regex_sentences(
|
||||
processed_tokens, subtitle_entries, max_subtitle_words,
|
||||
subtitle_mode, fallback_end_time
|
||||
subtitle_mode_str, fallback_end_time
|
||||
)
|
||||
else:
|
||||
# Word count-based grouping (e.g., "5" for 5-word groups)
|
||||
_process_word_count(
|
||||
processed_tokens, subtitle_entries, max_subtitle_words,
|
||||
subtitle_mode, fallback_end_time
|
||||
subtitle_mode_str, fallback_end_time
|
||||
)
|
||||
|
||||
|
||||
@@ -91,10 +137,8 @@ def _process_karaoke_highlighting(
|
||||
current_sentence.append(token)
|
||||
word_count += 1
|
||||
|
||||
# Split sentences based on separator or word count
|
||||
if (
|
||||
re.search(separator, token["text"]) and token.get("whitespace") == " "
|
||||
) or word_count >= max_subtitle_words:
|
||||
is_boundary = _is_sentence_boundary(token, current_sentence, separator)
|
||||
if is_boundary or word_count >= max_subtitle_words:
|
||||
if current_sentence:
|
||||
# Create karaoke subtitle entry for this sentence
|
||||
start_time = current_sentence[0]["start"]
|
||||
@@ -109,12 +153,17 @@ def _process_karaoke_highlighting(
|
||||
if t.get("end") is not None and t.get("start") is not None
|
||||
else 0.5
|
||||
)
|
||||
try:
|
||||
duration_cs = int(duration * 100)
|
||||
except (ValueError, OverflowError, TypeError):
|
||||
duration_cs = 50
|
||||
# Add karaoke effect
|
||||
karaoke_text += f"{{\\kf{duration_cs}}}{t['text']}{t.get('whitespace', '') or ''}"
|
||||
karaoke_text += f"{{\\kf{duration_cs}}}{t.get('text', '')}{t.get('whitespace', '') or ''}"
|
||||
|
||||
text_stripped = karaoke_text.strip()
|
||||
if text_stripped:
|
||||
subtitle_entries.append(
|
||||
(start_time, end_time, karaoke_text.strip())
|
||||
(start_time, end_time, text_stripped)
|
||||
)
|
||||
current_sentence = []
|
||||
word_count = 0
|
||||
@@ -128,9 +177,14 @@ def _process_karaoke_highlighting(
|
||||
karaoke_text = ""
|
||||
for t in current_sentence:
|
||||
duration = t["end"] - t["start"] if t.get("end") and t.get("start") else 0.5
|
||||
try:
|
||||
duration_cs = int(duration * 100)
|
||||
karaoke_text += f"{{\\kf{duration_cs}}}{t['text']}{t.get('whitespace', '') or ''}"
|
||||
subtitle_entries.append((start_time, end_time, karaoke_text.strip()))
|
||||
except (ValueError, OverflowError, TypeError):
|
||||
duration_cs = 50
|
||||
karaoke_text += f"{{\\kf{duration_cs}}}{t.get('text', '')}{t.get('whitespace', '') or ''}"
|
||||
text_stripped = karaoke_text.strip()
|
||||
if text_stripped:
|
||||
subtitle_entries.append((start_time, end_time, text_stripped))
|
||||
|
||||
# Fallback for last entry
|
||||
_apply_fallback_end_time(subtitle_entries, fallback_end_time)
|
||||
@@ -166,7 +220,7 @@ def _process_spacy_sentences(
|
||||
# Build full text and track character positions to token indices
|
||||
full_text = ""
|
||||
for token in tokens:
|
||||
text_part = token["text"] + (token.get("whitespace") or "")
|
||||
text_part = str(token.get("text", "")) + (token.get("whitespace") or "")
|
||||
full_text += text_part
|
||||
|
||||
# Get sentence boundaries from spaCy
|
||||
@@ -174,7 +228,7 @@ def _process_spacy_sentences(
|
||||
sentence_boundaries = [sent.end_char for sent in doc.sents]
|
||||
|
||||
# For "Sentence + Comma" mode, also split on commas
|
||||
if subtitle_mode == SubtitleMode.SENTENCE_COMMA:
|
||||
if subtitle_mode in (SubtitleMode.SENTENCE_COMMA.value, "Sentence + Comma"):
|
||||
comma_positions = [
|
||||
i + 1 for i, c in enumerate(full_text) if c == ","
|
||||
]
|
||||
@@ -182,6 +236,56 @@ def _process_spacy_sentences(
|
||||
set(sentence_boundaries + comma_positions)
|
||||
)
|
||||
|
||||
# spaCy does not treat ellipsis ("...", "..", "…") as a sentence
|
||||
# boundary ("Lorem ipsum... Lorem..." stays one sentence), so ellipsis
|
||||
# runs followed by whitespace/end would merge into a single subtitle
|
||||
# entry. Add explicit boundaries after them. Single dots ("Mr.") stay
|
||||
# spaCy's responsibility so abbreviations don't regress.
|
||||
for m in re.finditer(r"\.{2,}(?=[\s\"'”’»›)\]}]|$)|…(?=[\s\"'”’»›)\]}]|$)", full_text):
|
||||
sentence_boundaries.append(m.end())
|
||||
# Double newlines are paragraph breaks: always split, even when spaCy
|
||||
# sees no sentence boundary.
|
||||
for m in re.finditer(r"\n{2,}", full_text):
|
||||
sentence_boundaries.append(m.end())
|
||||
sentence_boundaries = sorted(set(sentence_boundaries))
|
||||
|
||||
# Multi-sentence single FakeToken handling
|
||||
if len(tokens) == 1 and len(sentence_boundaries) > 1:
|
||||
single = tokens[0]
|
||||
start_time = single.get("start", 0.0) or 0.0
|
||||
end_time = single.get("end")
|
||||
duration = (end_time - start_time) if (end_time is not None and end_time > start_time) else 0.0
|
||||
|
||||
prev_pos = 0
|
||||
cur_start = start_time
|
||||
total_chars = max(len(full_text), 1)
|
||||
|
||||
for i, b_pos in enumerate(sentence_boundaries):
|
||||
piece = full_text[prev_pos:b_pos].strip()
|
||||
if not piece:
|
||||
prev_pos = b_pos
|
||||
continue
|
||||
if i == len(sentence_boundaries) - 1:
|
||||
cur_end = end_time if end_time is not None else (cur_start + 1.0)
|
||||
else:
|
||||
cur_end = cur_start + duration * len(piece) / total_chars
|
||||
subtitle_entries.append((cur_start, cur_end, piece))
|
||||
cur_start = cur_end
|
||||
prev_pos = b_pos
|
||||
|
||||
if prev_pos < len(full_text):
|
||||
remainder = full_text[prev_pos:].strip()
|
||||
if remainder:
|
||||
remainder_end = end_time
|
||||
if remainder_end is None:
|
||||
remainder_end = fallback_end_time
|
||||
if remainder_end is None:
|
||||
remainder_end = cur_start
|
||||
subtitle_entries.append((cur_start, remainder_end, remainder))
|
||||
|
||||
_apply_fallback_end_time(subtitle_entries, fallback_end_time)
|
||||
return
|
||||
|
||||
# Group tokens by sentence boundaries
|
||||
current_sentence = []
|
||||
word_count = 0
|
||||
@@ -191,7 +295,7 @@ def _process_spacy_sentences(
|
||||
for token in tokens:
|
||||
current_sentence.append(token)
|
||||
word_count += 1
|
||||
text_len = len(token["text"]) + len(token.get("whitespace") or "")
|
||||
text_len = len(str(token.get("text", ""))) + len(token.get("whitespace") or "")
|
||||
current_char_pos += text_len
|
||||
|
||||
# Check if we've hit a sentence boundary or max words
|
||||
@@ -204,15 +308,19 @@ def _process_spacy_sentences(
|
||||
start_time = current_sentence[0]["start"]
|
||||
end_time = current_sentence[-1]["end"]
|
||||
sentence_text = "".join(
|
||||
t["text"] + (t.get("whitespace") or "")
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "")
|
||||
for t in current_sentence
|
||||
)
|
||||
).strip()
|
||||
if sentence_text:
|
||||
subtitle_entries.append(
|
||||
(start_time, end_time, sentence_text.strip())
|
||||
(start_time, end_time, sentence_text)
|
||||
)
|
||||
current_sentence = []
|
||||
word_count = 0
|
||||
if at_boundary:
|
||||
while (
|
||||
boundary_idx < len(sentence_boundaries)
|
||||
and current_char_pos >= sentence_boundaries[boundary_idx]
|
||||
):
|
||||
boundary_idx += 1
|
||||
|
||||
# Add remaining tokens
|
||||
@@ -220,11 +328,12 @@ def _process_spacy_sentences(
|
||||
start_time = current_sentence[0]["start"]
|
||||
end_time = current_sentence[-1]["end"]
|
||||
sentence_text = "".join(
|
||||
t["text"] + (t.get("whitespace") or "")
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "")
|
||||
for t in current_sentence
|
||||
)
|
||||
).strip()
|
||||
if sentence_text:
|
||||
subtitle_entries.append(
|
||||
(start_time, end_time, sentence_text.strip())
|
||||
(start_time, end_time, sentence_text)
|
||||
)
|
||||
|
||||
# Fallback for last entry
|
||||
@@ -240,9 +349,9 @@ def _process_regex_sentences(
|
||||
) -> None:
|
||||
"""Process tokens using regex for sentence boundary detection."""
|
||||
# Define separator pattern based on mode
|
||||
if subtitle_mode == SubtitleMode.LINE:
|
||||
if subtitle_mode in (SubtitleMode.LINE.value, "Line"):
|
||||
separator = r"\n"
|
||||
elif subtitle_mode == SubtitleMode.SENTENCE:
|
||||
elif subtitle_mode in (SubtitleMode.SENTENCE.value, "Sentence"):
|
||||
separator = rf"[{PUNCTUATION_SENTENCE}]"
|
||||
else: # Sentence + Comma
|
||||
separator = rf"[{PUNCTUATION_SENTENCE_COMMA}]"
|
||||
@@ -255,21 +364,21 @@ def _process_regex_sentences(
|
||||
word_count += 1
|
||||
|
||||
# Split sentences based on separator or word count
|
||||
if (
|
||||
re.search(separator, token["text"]) and token.get("whitespace") == " "
|
||||
) or word_count >= max_subtitle_words:
|
||||
is_boundary = _is_sentence_boundary(token, current_sentence, separator)
|
||||
if is_boundary or word_count >= max_subtitle_words:
|
||||
if current_sentence:
|
||||
# Create subtitle entry for this sentence
|
||||
start_time = current_sentence[0]["start"]
|
||||
end_time = current_sentence[-1]["end"]
|
||||
|
||||
# Simplified text joining logic
|
||||
sentence_text = ""
|
||||
for t in current_sentence:
|
||||
sentence_text += t["text"] + (t.get("whitespace") or "")
|
||||
sentence_text = "".join(
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "")
|
||||
for t in current_sentence
|
||||
).strip()
|
||||
|
||||
if sentence_text:
|
||||
subtitle_entries.append(
|
||||
(start_time, end_time, sentence_text.strip())
|
||||
(start_time, end_time, sentence_text)
|
||||
)
|
||||
current_sentence = []
|
||||
word_count = 0
|
||||
@@ -279,23 +388,39 @@ def _process_regex_sentences(
|
||||
start_time = current_sentence[0]["start"]
|
||||
end_time = current_sentence[-1]["end"]
|
||||
|
||||
sentence_text = ""
|
||||
for t in current_sentence:
|
||||
sentence_text += t["text"] + (t.get("whitespace") or "")
|
||||
sentence_text = sentence_text.strip()
|
||||
sentence_text = "".join(
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "")
|
||||
for t in current_sentence
|
||||
).strip()
|
||||
|
||||
if len(current_sentence) == 1:
|
||||
parts = re.split(rf"(?<={separator})\s+", sentence_text)
|
||||
split_pat = (
|
||||
r"\n+"
|
||||
if separator == r"\n"
|
||||
else rf"(?<={separator})\s+|(?<={separator}[{re.escape(_CLOSING_DELIMS)}])\s+"
|
||||
)
|
||||
parts = [p.strip() for p in re.split(split_pat, sentence_text) if p.strip()]
|
||||
if len(parts) > 1:
|
||||
d = end_time - start_time
|
||||
d = (end_time - start_time) if (end_time is not None and start_time is not None and end_time > start_time) else 0.0
|
||||
total_len = max(len(sentence_text), 1)
|
||||
cur_s = start_time if start_time is not None else 0.0
|
||||
for i, p in enumerate(parts):
|
||||
e = end_time if i == len(parts) - 1 else start_time + d * len(p) / len(sentence_text)
|
||||
subtitle_entries.append((start_time, e, p.strip()))
|
||||
start_time = e
|
||||
if i == len(parts) - 1 and end_time is not None:
|
||||
e = end_time
|
||||
else:
|
||||
e = cur_s + d * len(p) / total_len
|
||||
subtitle_entries.append((cur_s, e, p))
|
||||
cur_s = e
|
||||
current_sentence = []
|
||||
|
||||
if current_sentence:
|
||||
subtitle_entries.append((start_time, end_time, sentence_text))
|
||||
if current_sentence and sentence_text:
|
||||
safe_start = start_time if start_time is not None else 0.0
|
||||
safe_end = end_time
|
||||
if safe_end is None:
|
||||
safe_end = fallback_end_time
|
||||
if safe_end is None:
|
||||
safe_end = safe_start
|
||||
subtitle_entries.append((safe_start, safe_end, sentence_text))
|
||||
|
||||
# Fallback for last entry
|
||||
_apply_fallback_end_time(subtitle_entries, fallback_end_time)
|
||||
@@ -328,14 +453,15 @@ def _process_word_count(
|
||||
# Split after counting N spaces
|
||||
if space_count >= word_count:
|
||||
text = "".join(
|
||||
t["text"] + (t.get("whitespace") or "")
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "")
|
||||
for t in current_group
|
||||
)
|
||||
).strip()
|
||||
if text:
|
||||
subtitle_entries.append(
|
||||
(
|
||||
current_group[0]["start"],
|
||||
current_group[-1]["end"],
|
||||
text.strip(),
|
||||
text,
|
||||
)
|
||||
)
|
||||
current_group = []
|
||||
@@ -344,10 +470,11 @@ def _process_word_count(
|
||||
# Add any remaining tokens
|
||||
if current_group:
|
||||
text = "".join(
|
||||
t["text"] + (t.get("whitespace") or "") for t in current_group
|
||||
)
|
||||
str(t.get("text", "")) + (t.get("whitespace") or "") for t in current_group
|
||||
).strip()
|
||||
if text:
|
||||
subtitle_entries.append(
|
||||
(current_group[0]["start"], current_group[-1]["end"], text.strip())
|
||||
(current_group[0]["start"], current_group[-1]["end"], text)
|
||||
)
|
||||
|
||||
# Fallback for last entry
|
||||
|
||||
@@ -672,6 +672,15 @@ def tokenize_with_spans(text: str) -> List[Tuple[str, int, int]]:
|
||||
]
|
||||
|
||||
|
||||
_OPENING_PUNCTUATION_CHARS = "«‹“‘([{¡¿「『"
|
||||
_CLOSING_PUNCTUATION_CHARS = "»›”’)]}」』"
|
||||
_STANDARD_PUNCTUATION_CHARS = ",.;:!?%"
|
||||
|
||||
_OPENING_PUNCT_CLASS = re.escape(_OPENING_PUNCTUATION_CHARS)
|
||||
_CLOSING_PUNCT_CLASS = re.escape(_CLOSING_PUNCTUATION_CHARS)
|
||||
_STANDARD_PUNCT_CLASS = re.escape(_STANDARD_PUNCTUATION_CHARS)
|
||||
|
||||
|
||||
def _cleanup_spacing(text: str) -> str:
|
||||
if not text:
|
||||
return text
|
||||
@@ -679,22 +688,39 @@ def _cleanup_spacing(text: str) -> str:
|
||||
for marker in ("\ufeff", "\u200b", "\u200c", "\u200d", "\u2060"):
|
||||
text = text.replace(marker, "")
|
||||
|
||||
# Collapse spaces before closing punctuation.
|
||||
text = re.sub(r"\s+([,.;:!?%])", r"\1", text)
|
||||
text = re.sub(r"\s+([’\"”»›)\]\}])", r"\1", text)
|
||||
# Collapse spaces before standard punctuation and unambiguous closing quotes/brackets.
|
||||
text = re.sub(rf"\s+([{_STANDARD_PUNCT_CLASS}])", r"\1", text)
|
||||
text = re.sub(rf"\s+([{_CLOSING_PUNCT_CLASS}])", r"\1", text)
|
||||
|
||||
# Remove spaces directly after opening punctuation/quotes.
|
||||
text = re.sub(r"([«‹“‘\"'(\[\{])\s+", r"\1", text)
|
||||
# Remove spaces directly after unambiguous opening punctuation/quotes.
|
||||
text = re.sub(rf"([{_OPENING_PUNCT_CLASS}])\s+", r"\1", text)
|
||||
|
||||
# Handle ambiguous straight quotes (\", ')
|
||||
# 1. Remove spaces directly after opening straight quotes:
|
||||
# e.g. ' \" word' -> ' \"word', '^\" word' -> '\"word', '(\" word' -> '(\"word'
|
||||
text = re.sub(rf"(^|[\s{_OPENING_PUNCT_CLASS}])([\"\'])\s+", r"\1\2", text)
|
||||
# 2. Collapse spaces directly before closing straight quotes:
|
||||
# e.g. 'word \" ' -> 'word\" ', 'word \".' -> 'word\".'
|
||||
text = re.sub(rf"\s+([\"\'])([\s{_STANDARD_PUNCT_CLASS}{_CLOSING_PUNCT_CLASS}]|$)", r"\1\2", text)
|
||||
|
||||
# Ensure spaces exist after sentence punctuation when followed by a word/quote.
|
||||
text = re.sub(r"([,.;:!?%])(?![\s”'\"’»›)])", r"\1 ", text)
|
||||
text = re.sub(r"([”\"’])(?![\s.,;:!?\"”’»›)])", r"\1 ", text)
|
||||
# Runs of punctuation ("...", "?!?", "!!") must stay together: no space
|
||||
# inside the run, only after it ("a...b" -> "a... b").
|
||||
text = re.sub(rf"([{_STANDARD_PUNCT_CLASS}])(?![\s{_STANDARD_PUNCT_CLASS}{_CLOSING_PUNCT_CLASS}\"\'”’»›)])", r"\1 ", text)
|
||||
# Ensure space after unambiguous closing quote when followed by a word (e.g. '”Next' -> '” Next')
|
||||
text = re.sub(rf"([{_CLOSING_PUNCT_CLASS}])(?![\s{_STANDARD_PUNCT_CLASS}{_CLOSING_PUNCT_CLASS}\"\'”’»›)])", r"\1 ", text)
|
||||
# Straight double quote closing (preceded by non-whitespace) followed directly by a word/number/opening
|
||||
text = re.sub(rf"(\S\")([A-Za-z0-9{_OPENING_PUNCT_CLASS}])", r"\1 \2", text)
|
||||
# Straight single quote closing (preceded by punctuation, not internal word apostrophe) followed by a word
|
||||
text = re.sub(rf"([{_STANDARD_PUNCT_CLASS}{_CLOSING_PUNCT_CLASS}]\')([A-Za-z0-9{_OPENING_PUNCT_CLASS}])", r"\1 \2", text)
|
||||
|
||||
# Tighten hyphen/em dash spacing between word characters.
|
||||
text = re.sub(r"(?<=\w)\s*([-–—])\s*(?=\w)", r"\1", text)
|
||||
|
||||
# Normalize multiple spaces.
|
||||
text = re.sub(r"\s{2,}", " ", text)
|
||||
# Normalize multiple spaces, preserving paragraph breaks (double
|
||||
# newlines must survive so the TTS engine can split on them).
|
||||
text = re.sub(r"[^\S\n]{2,}", " ", text)
|
||||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||||
return text.strip()
|
||||
|
||||
|
||||
@@ -1622,8 +1648,18 @@ def normalize_apostrophes(
|
||||
results.append((tok, category, norm))
|
||||
normalized_tokens.append(norm)
|
||||
|
||||
filtered = [token for token in normalized_tokens if token]
|
||||
normalized_text = _cleanup_spacing(" ".join(filtered))
|
||||
out_pieces: List[str] = []
|
||||
last_end = 0
|
||||
for (tok, start, end), norm in zip(token_entries, normalized_tokens):
|
||||
if start > last_end:
|
||||
out_pieces.append(text[last_end:start])
|
||||
out_pieces.append(norm)
|
||||
last_end = end
|
||||
if last_end < len(text):
|
||||
out_pieces.append(text[last_end:])
|
||||
|
||||
reconstructed = "".join(out_pieces)
|
||||
normalized_text = _cleanup_spacing(reconstructed)
|
||||
return normalized_text, results
|
||||
|
||||
|
||||
@@ -1824,7 +1860,10 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
|
||||
for digit in trimmed_fraction:
|
||||
if not digit.isdigit():
|
||||
return token
|
||||
try:
|
||||
digit_words.append(_DIGIT_WORDS[int(digit)])
|
||||
except (ValueError, IndexError):
|
||||
return token
|
||||
|
||||
spoken = f"{integer_words} point {' '.join(digit_words)}"
|
||||
return f"minus {spoken}" if is_negative else spoken
|
||||
@@ -1846,18 +1885,27 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
|
||||
# Magnitude case: $2.5 million -> two point five million dollars
|
||||
if "." in amount_str:
|
||||
integer_part, fraction_part = amount_str.split(".", 1)
|
||||
try:
|
||||
integer_val = int(integer_part)
|
||||
except ValueError:
|
||||
return match.group(0)
|
||||
integer_words = _int_to_words(integer_val, language)
|
||||
|
||||
# Spell out fraction digits
|
||||
digit_words = []
|
||||
for digit in fraction_part:
|
||||
if digit.isdigit():
|
||||
try:
|
||||
digit_words.append(_DIGIT_WORDS[int(digit)])
|
||||
except (ValueError, IndexError):
|
||||
return match.group(0)
|
||||
|
||||
amount_spoken = f"{integer_words} point {' '.join(digit_words)}"
|
||||
else:
|
||||
try:
|
||||
amount_spoken = _int_to_words(int(amount), language)
|
||||
except (ValueError, OverflowError):
|
||||
return match.group(0)
|
||||
|
||||
currency_names = {
|
||||
"$": "dollars",
|
||||
|
||||
+78
-50
@@ -137,6 +137,28 @@ class ThreadSafeLogSignal(QObject):
|
||||
self.log_signal.emit(message)
|
||||
|
||||
|
||||
_UPDATE_CHECK_URL = "https://raw.githubusercontent.com/denizsafak/abogen/refs/heads/main/abogen/VERSION"
|
||||
_UPDATE_CHECK_TIMEOUT = 8 # seconds; bounds offline/DNS hangs so the GUI never blocks
|
||||
|
||||
|
||||
class _UpdateCheckThread(QThread):
|
||||
"""Fetch the remote VERSION file off the GUI thread."""
|
||||
|
||||
succeeded = pyqtSignal(str)
|
||||
failed = pyqtSignal(str)
|
||||
|
||||
def run(self):
|
||||
import urllib.request
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
_UPDATE_CHECK_URL, timeout=_UPDATE_CHECK_TIMEOUT
|
||||
) as response:
|
||||
self.succeeded.emit(response.read().decode().strip())
|
||||
except Exception as exc: # offline, DNS hang, HTTP error, ...
|
||||
self.failed.emit(str(exc))
|
||||
|
||||
|
||||
class IconProvider(QFileIconProvider):
|
||||
def icon(self, fileInfo):
|
||||
return super().icon(fileInfo)
|
||||
@@ -4077,9 +4099,63 @@ Categories=AudioVideo;Audio;Utility;
|
||||
self.check_for_updates_startup()
|
||||
|
||||
def check_for_updates_startup(self):
|
||||
import urllib.request
|
||||
# Network I/O runs in a worker thread: urlopen without a timeout on
|
||||
# the GUI thread froze the whole app when offline (DNS/connect can
|
||||
# hang for minutes). Results return via signals on the GUI thread.
|
||||
thread = getattr(self, "_update_check_thread", None)
|
||||
if thread is not None:
|
||||
try:
|
||||
if thread.isRunning():
|
||||
return
|
||||
except RuntimeError:
|
||||
pass # previous thread already finished/deleted
|
||||
show_result = (
|
||||
hasattr(self, "_show_update_check_result")
|
||||
and self._show_update_check_result
|
||||
)
|
||||
self._show_update_check_result = False
|
||||
self._update_check_thread = _UpdateCheckThread(self)
|
||||
self._update_check_thread.succeeded.connect(
|
||||
lambda remote_raw: self._on_update_check_done(remote_raw, show_result)
|
||||
)
|
||||
self._update_check_thread.failed.connect(
|
||||
lambda err: self._on_update_check_failed(err, show_result)
|
||||
)
|
||||
self._update_check_thread.finished.connect(
|
||||
self._update_check_thread.deleteLater
|
||||
)
|
||||
self._update_check_thread.start()
|
||||
|
||||
def show_update_message(remote_version, local_version):
|
||||
def _on_update_check_done(self, remote_raw, show_result):
|
||||
remote_version = remote_raw.strip()
|
||||
local_version = VERSION
|
||||
try:
|
||||
remote_num = int("".join(remote_version.split(".")))
|
||||
local_num = int("".join(local_version.split(".")))
|
||||
except ValueError:
|
||||
return
|
||||
if remote_num > local_num:
|
||||
# Use QTimer to ensure UI is ready, then show update message.
|
||||
QTimer.singleShot(
|
||||
1000,
|
||||
lambda: self._show_update_message(remote_version, local_version),
|
||||
)
|
||||
elif show_result:
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"Up to Date",
|
||||
f"You are running the latest version of {PROGRAM_NAME} ({local_version}).",
|
||||
)
|
||||
|
||||
def _on_update_check_failed(self, err, show_result):
|
||||
if show_result:
|
||||
QMessageBox.warning(
|
||||
self,
|
||||
"Update Check Failed",
|
||||
f"Could not check for updates:\n{err}",
|
||||
)
|
||||
|
||||
def _show_update_message(self, remote_version, local_version):
|
||||
msg_box = QMessageBox(self)
|
||||
msg_box.setIcon(QMessageBox.Icon.Information)
|
||||
msg_box.setWindowTitle("Update Available")
|
||||
@@ -4103,50 +4179,6 @@ Categories=AudioVideo;Audio;Utility;
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Reset flag to track if we should show "no updates" message
|
||||
show_result = (
|
||||
hasattr(self, "_show_update_check_result")
|
||||
and self._show_update_check_result
|
||||
)
|
||||
self._show_update_check_result = False
|
||||
|
||||
try:
|
||||
update_url = "https://raw.githubusercontent.com/denizsafak/abogen/refs/heads/main/abogen/VERSION"
|
||||
with urllib.request.urlopen(update_url) as response:
|
||||
remote_raw = response.read().decode().strip()
|
||||
local_raw = VERSION
|
||||
|
||||
# Parse version numbers
|
||||
remote_version = remote_raw
|
||||
local_version = local_raw
|
||||
|
||||
try:
|
||||
remote_num = int("".join(remote_version.split(".")))
|
||||
local_num = int("".join(local_version.split(".")))
|
||||
except ValueError as ve:
|
||||
return
|
||||
|
||||
if remote_num > local_num:
|
||||
# Use QTimer to ensure UI is ready, then show update message.
|
||||
QTimer.singleShot(
|
||||
1000, lambda: show_update_message(remote_version, local_version)
|
||||
)
|
||||
elif show_result:
|
||||
# Show "no updates" message if manually checking
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"Up to Date",
|
||||
f"You are running the latest version of {PROGRAM_NAME} ({local_version}).",
|
||||
)
|
||||
except Exception as e:
|
||||
if show_result:
|
||||
QMessageBox.warning(
|
||||
self,
|
||||
"Update Check Failed",
|
||||
f"Could not check for updates:\n{str(e)}",
|
||||
)
|
||||
pass
|
||||
|
||||
def clear_cache_files(self):
|
||||
"""Clear cache files created by the program."""
|
||||
import glob
|
||||
@@ -4248,8 +4280,6 @@ Categories=AudioVideo;Audio;Utility;
|
||||
|
||||
def set_max_log_lines(self):
|
||||
"""Open a dialog to set the maximum lines in the log window."""
|
||||
from PyQt6.QtWidgets import QInputDialog
|
||||
|
||||
value, ok = QInputDialog.getInt(
|
||||
self,
|
||||
"Max Lines in Log Window",
|
||||
@@ -4271,8 +4301,6 @@ Categories=AudioVideo;Audio;Utility;
|
||||
|
||||
def set_max_subtitle_words(self):
|
||||
"""Open a dialog to set the maximum words per subtitle"""
|
||||
from PyQt6.QtWidgets import QInputDialog
|
||||
|
||||
current_value = self.config.get("max_subtitle_words", _DEFAULTS["max_subtitle_words"])
|
||||
|
||||
value, ok = QInputDialog.getInt(
|
||||
|
||||
@@ -90,3 +90,10 @@ def test_manual_override_normalization():
|
||||
assert normalize_manual_override_token("The") == "the"
|
||||
assert normalize_manual_override_token(" A ") == "a"
|
||||
assert normalize_manual_override_token("word") == "word"
|
||||
|
||||
|
||||
def test_paragraph_breaks_and_ellipsis_preserved():
|
||||
normalized = normalize("Test. Lorem ipsum...\n\nLorem...\n\nLorem ...")
|
||||
assert "\n\n" in normalized
|
||||
assert ". . ." not in normalized
|
||||
assert "Lorem..." in normalized
|
||||
|
||||
@@ -0,0 +1,402 @@
|
||||
"""Comprehensive tests for subtitle generation across different models, modes, and text scenarios.
|
||||
|
||||
Tests include:
|
||||
- Quotation mark handling (straight quotes, curly quotes, guillemets, dialogs)
|
||||
- No spurious spaces after opening quotes (e.g., test "word word" vs test " word word")
|
||||
- Proper sentence boundary detection for quoted dialogues (e.g., "Hello." She said.)
|
||||
- Paragraph handling and multi-line text
|
||||
- All subtitle modes (Line, Sentence, Sentence + Comma, Sentence + Highlighting, N-words)
|
||||
- Both TTS model token styles (Kokoro per-word tokens and Supertonic FakeTokens)
|
||||
- Non-English and multilingual scenarios
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from abogen.domain.enums import Language, SubtitleMode
|
||||
from abogen.domain.normalization import prepare_text_for_tts
|
||||
from abogen.domain.subtitle_generation import (
|
||||
process_subtitle_tokens,
|
||||
PUNCTUATION_SENTENCE,
|
||||
PUNCTUATION_SENTENCE_COMMA,
|
||||
)
|
||||
|
||||
|
||||
class TestQuoteNormalizationAndSpacing:
|
||||
"""Verify text normalization correctly preserves quotation mark spacing."""
|
||||
|
||||
def test_straight_quote_mid_sentence_no_extra_space(self):
|
||||
"""Input 'test "word word"' should keep space before quote and no space after."""
|
||||
text = 'test "word word"'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert '" word' not in normalized
|
||||
assert 'test "' in normalized or 'test "word' in normalized
|
||||
|
||||
def test_straight_quote_at_start_no_extra_space(self):
|
||||
"""Input '"word word"' should not have a leading space after opening quote."""
|
||||
text = '"word word"'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert not normalized.startswith('" ')
|
||||
assert normalized.startswith('"word')
|
||||
|
||||
def test_dialogue_quote_spacing(self):
|
||||
"""He said, "Hello world." should preserve proper comma-space-quote-word sequence."""
|
||||
text = 'He said, "Hello world."'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert 'said, "' in normalized or 'said,"' not in normalized
|
||||
assert '" Hello' not in normalized
|
||||
assert '"Hello' in normalized
|
||||
|
||||
def test_quote_with_contraction(self):
|
||||
"""Contraction inside quotes like "Don't go!" should expand cleanly without extra spaces."""
|
||||
text = '"Don\'t go!"'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert '" Do not' not in normalized
|
||||
assert '"Do not' in normalized or '"Don\'t' in normalized
|
||||
|
||||
def test_curly_quotes_preserved(self):
|
||||
"""Curly quotes like “Hello world.” should not have spurious spacing."""
|
||||
text = '“Hello world.”'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert '“ ' not in normalized
|
||||
assert ' ”' not in normalized
|
||||
|
||||
def test_spanish_opening_punctuation(self):
|
||||
"""Spanish inverted exclamation ¡Hola! should not have space after ¡."""
|
||||
text = '¡Hola mundo!'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert '¡ ' not in normalized
|
||||
|
||||
def test_french_guillemets_spacing(self):
|
||||
"""French guillemets « Bonjour » should clean up spaces properly."""
|
||||
text = '« Bonjour »'
|
||||
normalized = prepare_text_for_tts(text)
|
||||
assert '« ' not in normalized
|
||||
assert ' »' not in normalized
|
||||
|
||||
|
||||
class TestKokoroPerWordTokenSubtitles:
|
||||
"""Tests using Kokoro-style per-word tokens with individual timestamps."""
|
||||
|
||||
def test_quoted_phrase_subtitles(self):
|
||||
"""Tokens for 'test "word word"' produce subtitle without space after quote."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.3, "text": "test", "whitespace": " "},
|
||||
{"start": 0.3, "end": 0.35, "text": '"', "whitespace": ""},
|
||||
{"start": 0.35, "end": 0.7, "text": "word", "whitespace": " "},
|
||||
{"start": 0.7, "end": 1.0, "text": "word", "whitespace": ""},
|
||||
{"start": 1.0, "end": 1.05, "text": '"', "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 1
|
||||
assert entries[0][2] == 'test "word word"'
|
||||
|
||||
def test_dialogue_sentence_splitting_regex(self):
|
||||
"""Dialogue ending with ." should split into separate sentence subtitles."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.05, "text": '"', "whitespace": ""},
|
||||
{"start": 0.05, "end": 0.5, "text": "Hello", "whitespace": " "},
|
||||
{"start": 0.5, "end": 0.9, "text": "world.", "whitespace": ""},
|
||||
{"start": 0.9, "end": 0.95, "text": '"', "whitespace": " "},
|
||||
{"start": 0.95, "end": 1.4, "text": "She", "whitespace": " "},
|
||||
{"start": 1.4, "end": 1.8, "text": "smiled.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert entries[0][2] == '"Hello world."'
|
||||
assert entries[1][2] == "She smiled."
|
||||
assert entries[0][0] == 0.0
|
||||
assert entries[0][1] == 0.95
|
||||
assert entries[1][0] == 0.95
|
||||
assert entries[1][1] == 1.8
|
||||
|
||||
def test_question_exclamation_dialogue_splitting(self):
|
||||
"""Dialogue with ?" and !" should split sentences cleanly."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.05, "text": '"', "whitespace": ""},
|
||||
{"start": 0.05, "end": 0.4, "text": "Why?", "whitespace": ""},
|
||||
{"start": 0.4, "end": 0.45, "text": '"', "whitespace": " "},
|
||||
{"start": 0.45, "end": 0.8, "text": "she", "whitespace": " "},
|
||||
{"start": 0.8, "end": 1.2, "text": "asked.", "whitespace": " "},
|
||||
{"start": 1.2, "end": 1.25, "text": '"', "whitespace": ""},
|
||||
{"start": 1.25, "end": 1.7, "text": "Because!", "whitespace": ""},
|
||||
{"start": 1.7, "end": 1.75, "text": '"', "whitespace": " "},
|
||||
{"start": 1.75, "end": 2.0, "text": "he", "whitespace": " "},
|
||||
{"start": 2.0, "end": 2.4, "text": "replied.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 4
|
||||
assert entries[0][2] == '"Why?"'
|
||||
assert entries[1][2] == "she asked."
|
||||
assert entries[2][2] == '"Because!"'
|
||||
assert entries[3][2] == "he replied."
|
||||
|
||||
def test_sentence_comma_mode_with_quotes(self):
|
||||
"""Sentence + Comma mode splits at commas and sentence boundaries."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.4, "text": "First,", "whitespace": " "},
|
||||
{"start": 0.4, "end": 0.8, "text": "she", "whitespace": " "},
|
||||
{"start": 0.8, "end": 1.2, "text": "said,", "whitespace": " "},
|
||||
{"start": 1.2, "end": 1.25, "text": '"', "whitespace": ""},
|
||||
{"start": 1.25, "end": 1.6, "text": "wait.", "whitespace": ""},
|
||||
{"start": 1.6, "end": 1.65, "text": '"', "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence + Comma",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) >= 2
|
||||
assert "First," in entries[0][2]
|
||||
|
||||
def test_karaoke_highlighting_with_quotes(self):
|
||||
"""Sentence + Highlighting generates valid karaoke tags with quotes."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.05, "text": '"', "whitespace": ""},
|
||||
{"start": 0.05, "end": 0.5, "text": "Hello", "whitespace": " "},
|
||||
{"start": 0.5, "end": 0.9, "text": "world.", "whitespace": ""},
|
||||
{"start": 0.9, "end": 0.95, "text": '"', "whitespace": " "},
|
||||
{"start": 0.95, "end": 1.4, "text": "She", "whitespace": " "},
|
||||
{"start": 1.4, "end": 1.8, "text": "said.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence + Highlighting",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert '{\\kf' in entries[0][2]
|
||||
assert '{\\kf' in entries[1][2]
|
||||
assert '"' in entries[0][2]
|
||||
|
||||
def test_word_count_mode_with_quotes(self):
|
||||
"""N-words mode (e.g. '3 words') groups tokens by space count."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.3, "text": "One", "whitespace": " "},
|
||||
{"start": 0.3, "end": 0.35, "text": '"', "whitespace": ""},
|
||||
{"start": 0.35, "end": 0.7, "text": "two", "whitespace": " "},
|
||||
{"start": 0.7, "end": 1.0, "text": "three", "whitespace": ""},
|
||||
{"start": 1.0, "end": 1.05, "text": '"', "whitespace": " "},
|
||||
{"start": 1.05, "end": 1.4, "text": "four", "whitespace": " "},
|
||||
{"start": 1.4, "end": 1.8, "text": "five", "whitespace": " "},
|
||||
{"start": 1.8, "end": 2.2, "text": "six.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="3",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert entries[0][2] == 'One "two three"'
|
||||
assert entries[1][2] == "four five six."
|
||||
|
||||
|
||||
class TestSupertonicAndFakeTokenSubtitles:
|
||||
"""Tests using Supertonic / non-English Kokoro FakeTokens (segment-level stubs)."""
|
||||
|
||||
def test_faketoken_multi_sentence_regex_split(self):
|
||||
"""A single FakeToken containing multiple sentences should split proportionally."""
|
||||
tokens = [
|
||||
{
|
||||
"start": 0.0,
|
||||
"end": 6.0,
|
||||
"text": 'First sentence. "Second quoted sentence." Third sentence.',
|
||||
"whitespace": "",
|
||||
}
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.ES, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 3
|
||||
assert entries[0][2] == "First sentence."
|
||||
assert entries[1][2] == '"Second quoted sentence."'
|
||||
assert entries[2][2] == "Third sentence."
|
||||
assert entries[0][0] == 0.0
|
||||
assert entries[2][1] == 6.0
|
||||
|
||||
def test_faketoken_multi_sentence_spacy_split(self):
|
||||
"""A single FakeToken in English with spaCy should split into separate sentences."""
|
||||
tokens = [
|
||||
{
|
||||
"start": 0.0,
|
||||
"end": 6.0,
|
||||
"text": 'The sun rose high. "Are you ready?" she asked. "Always," he replied.',
|
||||
"whitespace": "",
|
||||
}
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=True
|
||||
)
|
||||
assert len(entries) >= 2
|
||||
assert entries[0][0] == 0.0
|
||||
assert entries[-1][1] == 6.0
|
||||
for e in entries:
|
||||
assert not e[2].startswith('" ')
|
||||
|
||||
def test_faketoken_single_sentence_with_quotes(self):
|
||||
"""Single sentence FakeToken preserves quotes cleanly."""
|
||||
tokens = [
|
||||
{
|
||||
"start": 1.0,
|
||||
"end": 3.5,
|
||||
"text": '"This is a single quoted thought."',
|
||||
"whitespace": "",
|
||||
}
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.FR, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 1
|
||||
assert entries[0][2] == '"This is a single quoted thought."'
|
||||
assert entries[0][0] == 1.0
|
||||
assert entries[0][1] == 3.5
|
||||
|
||||
def test_line_mode_with_faketokens(self):
|
||||
"""Line mode emits one subtitle per line / segment."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 2.0, "text": 'Line 1 with "quotes"', "whitespace": "\n"},
|
||||
{"start": 2.0, "end": 4.0, "text": 'Line 2 with "more quotes"', "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Line",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert entries[0][2] == 'Line 1 with "quotes"'
|
||||
assert entries[1][2] == 'Line 2 with "more quotes"'
|
||||
|
||||
|
||||
class TestComplexParagraphsAndEdgeCases:
|
||||
"""Tests for paragraphs, multiple newlines, and unusual punctuation combinations."""
|
||||
|
||||
def test_paragraph_multi_line_token_flow(self):
|
||||
"""Text spanning paragraphs with multiple sentences."""
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.5, "text": "Paragraph", "whitespace": " "},
|
||||
{"start": 0.5, "end": 1.0, "text": "one.", "whitespace": "\n\n"},
|
||||
{"start": 1.0, "end": 1.5, "text": "Paragraph", "whitespace": " "},
|
||||
{"start": 1.5, "end": 2.0, "text": "two.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert entries[0][2] == "Paragraph one."
|
||||
assert entries[1][2] == "Paragraph two."
|
||||
|
||||
def test_nested_quotes_and_parentheses(self):
|
||||
"""Sentence with nested quotes and parentheses: He said, "(Wait) 'now'!" """
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.3, "text": "He", "whitespace": " "},
|
||||
{"start": 0.3, "end": 0.6, "text": "said,", "whitespace": " "},
|
||||
{"start": 0.6, "end": 0.65, "text": '"', "whitespace": ""},
|
||||
{"start": 0.65, "end": 0.7, "text": "(", "whitespace": ""},
|
||||
{"start": 0.7, "end": 1.0, "text": "Wait", "whitespace": ""},
|
||||
{"start": 1.0, "end": 1.05, "text": ")", "whitespace": " "},
|
||||
{"start": 1.05, "end": 1.1, "text": "'", "whitespace": ""},
|
||||
{"start": 1.1, "end": 1.4, "text": "now", "whitespace": ""},
|
||||
{"start": 1.4, "end": 1.45, "text": "'!", "whitespace": ""},
|
||||
{"start": 1.45, "end": 1.5, "text": '"', "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 1
|
||||
assert entries[0][2] == 'He said, "(Wait) \'now\'!"'
|
||||
|
||||
def test_trailing_quotes_and_ellipsis(self):
|
||||
"""Sentence ending with ellipsis and quote: "I wonder..." """
|
||||
tokens = [
|
||||
{"start": 0.0, "end": 0.05, "text": '"', "whitespace": ""},
|
||||
{"start": 0.05, "end": 0.3, "text": "I", "whitespace": " "},
|
||||
{"start": 0.3, "end": 0.8, "text": "wonder...", "whitespace": ""},
|
||||
{"start": 0.8, "end": 0.85, "text": '"', "whitespace": " "},
|
||||
{"start": 0.85, "end": 1.2, "text": "he", "whitespace": " "},
|
||||
{"start": 1.2, "end": 1.6, "text": "mused.", "whitespace": ""},
|
||||
]
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
tokens, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=False
|
||||
)
|
||||
assert len(entries) == 2
|
||||
assert entries[0][2] == '"I wonder..."'
|
||||
assert entries[1][2] == "he mused."
|
||||
|
||||
|
||||
class TestEllipsisAndParagraphBreaks:
|
||||
"""Regression: '...' sentences merged into one entry; '\\n\\n' flattened.
|
||||
|
||||
spaCy does not treat ellipsis as a sentence boundary, so
|
||||
'Lorem ipsum... Lorem...' stayed a single subtitle entry. And
|
||||
_cleanup_spacing collapsed paragraph breaks before TTS.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _tok(text_ws, dur=0.5):
|
||||
toks, t = [], 0.0
|
||||
for text, ws in text_ws:
|
||||
toks.append({"start": t, "end": t + dur, "text": text, "whitespace": ws})
|
||||
t += dur
|
||||
return toks, t
|
||||
|
||||
def test_spacy_splits_ellipsis_sentences(self):
|
||||
# Kokoro-style tokens: '...' arrives as 3 dot tokens.
|
||||
toks, end = self._tok([
|
||||
("Test", ""), (".", " "), ("Lorem", " "), ("ipsum", ""),
|
||||
(".", ""), (".", ""), (".", " "),
|
||||
("Lorem", ""), (".", ""), (".", ""), (".", ""),
|
||||
])
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
toks, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=True,
|
||||
fallback_end_time=end,
|
||||
)
|
||||
assert [e[2] for e in entries] == ["Test.", "Lorem ipsum...", "Lorem..."]
|
||||
|
||||
def test_spacy_keeps_abbreviations_intact(self):
|
||||
toks, end = self._tok([
|
||||
("Mr.", " "), ("Smith", " "), ("went", " "), ("home", ""),
|
||||
(".", " "), ("He", " "), ("slept", ""), (".", ""),
|
||||
])
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
toks, entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=True,
|
||||
fallback_end_time=end,
|
||||
)
|
||||
assert [e[2] for e in entries] == ["Mr. Smith went home.", "He slept."]
|
||||
|
||||
def test_spacy_splits_faketoken_ellipsis(self):
|
||||
entries = []
|
||||
process_subtitle_tokens(
|
||||
[{"start": 0.0, "end": 3.0,
|
||||
"text": "Lorem ipsum... Lorem... Lorem...", "whitespace": ""}],
|
||||
entries, max_subtitle_words=50, subtitle_mode="Sentence",
|
||||
language=Language.EN_US, use_spacy_segmentation=True,
|
||||
fallback_end_time=3.0,
|
||||
)
|
||||
assert [e[2] for e in entries] == ["Lorem ipsum...", "Lorem...", "Lorem..."]
|
||||
Reference in New Issue
Block a user