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

Reformat using black

Browse Source
This commit is contained in:
Deniz Şafak
2026-01-09 01:36:14 +03:00
parent 3c800df450
commit 5ae153f841
48 changed files with 1530 additions and 894 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/book_parser.py
+55 -40
View File
@@ -18,7 +18,9 @@ from abogen.subtitle_utils import clean_text, calculate_text_length
_BRACKETED_NUMBERS_PATTERN = re.compile(r"\[\s*\d+\s*\]")
_STANDALONE_PAGE_NUMBERS_PATTERN = re.compile(r"^\s*\d+\s*$", re.MULTILINE)
_PAGE_NUMBERS_AT_END_PATTERN = re.compile(r"\s+\d+\s*$", re.MULTILINE)
_PAGE_NUMBERS_WITH_DASH_PATTERN = re.compile(r"\s+[-–—]\s*\d+\s*[-–—]?\s*$", re.MULTILINE)
_PAGE_NUMBERS_WITH_DASH_PATTERN = re.compile(
r"\s+[-–—]\s*\d+\s*[-–—]?\s*$", re.MULTILINE
)
class BaseBookParser(ABC):
@@ -114,7 +116,7 @@ class PdfParser(BaseBookParser):
for page_num in range(len(self.pdf_doc)):
text = clean_text(self.pdf_doc[page_num].get_text())
# Clean up common PDF artifacts:
# - Remove bracketed numbers often used for citations [1]
text = _BRACKETED_NUMBERS_PATTERN.sub("", text)
@@ -128,7 +130,7 @@ class PdfParser(BaseBookParser):
page_id = f"page_{page_num + 1}"
self.content_texts[page_id] = text
self.content_lengths[page_id] = calculate_text_length(text)
return self.content_texts, self.content_lengths
def _extract_book_metadata(self):
@@ -166,7 +168,7 @@ class MarkdownParser(BaseBookParser):
def process_content(self, replace_single_newlines=True):
if self.markdown_text is None:
self.load()
self._process_markdown_content()
return self.content_texts, self.content_lengths
@@ -255,11 +257,11 @@ class MarkdownParser(BaseBookParser):
else:
self.content_texts[chapter_id] = header_name
self.content_lengths[chapter_id] = calculate_text_length(header_name)
def get_chapters(self):
chapters = super().get_chapters()
if not chapters and "markdown_content" in self.content_texts:
chapters.append(("markdown_content", "Content"))
chapters.append(("markdown_content", "Content"))
return chapters
@@ -282,17 +284,19 @@ class EpubParser(BaseBookParser):
# Patch ebooklib to skip missing files
import types
from ebooklib import epub as _epub_module
reader_class = _epub_module.EpubReader
orig_read_file = reader_class.read_file
def safe_read_file(self, name):
try:
return orig_read_file(self, name)
except KeyError:
logging.warning(f"Missing file in EPUB: {name}. Returning empty bytes.")
logging.warning(
f"Missing file in EPUB: {name}. Returning empty bytes."
)
return b""
reader_class.read_file = safe_read_file
try:
self.book = epub.read_epub(self.book_path)
@@ -302,7 +306,7 @@ class EpubParser(BaseBookParser):
def process_content(self, replace_single_newlines=True):
if not self.book:
self.load()
self.book_metadata = self._extract_book_metadata()
try:
nav_item, nav_type = self._identify_nav_item()
@@ -310,7 +314,7 @@ class EpubParser(BaseBookParser):
except Exception as e:
logging.warning(f"EPUB nav processing failed: {e}. Falling back to spine.")
self._process_epub_content_spine_fallback()
return self.content_texts, self.content_lengths
def _extract_book_metadata(self):
@@ -412,7 +416,7 @@ class EpubParser(BaseBookParser):
):
"""
Recursive parsing of NCX navigation nodes.
Logic tested by: tests/test_epub_ncx_parsing.py
"""
nav_label = nav_point.find("navLabel")
@@ -428,7 +432,9 @@ class EpubParser(BaseBookParser):
if src:
base_href, fragment = src.split("#", 1) if "#" in src else (src, None)
doc_key, doc_idx = self._find_doc_key(base_href, doc_order, doc_order_decoded)
doc_key, doc_idx = self._find_doc_key(
base_href, doc_order, doc_order_decoded
)
if not doc_key:
current_entry_node["has_content"] = False
else:
@@ -458,19 +464,20 @@ class EpubParser(BaseBookParser):
)
if title and (
current_entry_node.get("has_content", False) or current_entry_node["children"]
current_entry_node.get("has_content", False)
or current_entry_node["children"]
):
tree_structure_list.append(current_entry_node)
def _extract_nav_li_title(self, li_element, link_element=None, span_element=None):
"""Helper to extract title from a nav <li> element, handling various structures."""
title = "Untitled Section"
if link_element:
title = link_element.get_text(strip=True) or title
elif span_element:
title = span_element.get_text(strip=True) or title
# Fallback to direct text if title is empty or default
# If we used link/span but got empty string, we try fallback.
# If we didn't use link/span, we try fallback.
@@ -480,11 +487,11 @@ class EpubParser(BaseBookParser):
).strip()
if li_text:
title = li_text
# Second fallback: if we have a span but title is still empty, try span text again
# Second fallback: if we have a span but title is still empty, try span text again
# (covered by logic above mostly, but mirroring original logic's intense fallback)
if (not title.strip() or title == "Untitled Section") and span_text:
title = span_text.get_text(strip=True) or title
title = span_text.get_text(strip=True) or title
return title
@@ -499,7 +506,7 @@ class EpubParser(BaseBookParser):
):
"""
Recursive parsing of HTML5 Navigation (li) nodes.
Logic tested by: tests/test_epub_html_nav_parsing.py
"""
link = li_element.find("a", recursive=False)
@@ -509,7 +516,7 @@ class EpubParser(BaseBookParser):
if link and "href" in link.attrs:
src = link["href"]
title = self._extract_nav_li_title(li_element, link, span_text)
current_entry_node["title"] = title
@@ -521,7 +528,9 @@ class EpubParser(BaseBookParser):
fragment = None
if src:
base_href, fragment = src.split("#", 1) if "#" in src else (src, None)
doc_key, doc_idx = self._find_doc_key(base_href, doc_order, doc_order_decoded)
doc_key, doc_idx = self._find_doc_key(
base_href, doc_order, doc_order_decoded
)
if doc_key is not None:
position = find_position_func(doc_key, fragment)
entry_data = {
@@ -557,18 +566,21 @@ class EpubParser(BaseBookParser):
# 1. Check ITEM_NAVIGATION
nav_items = list(self.book.get_items_of_type(ebooklib.ITEM_NAVIGATION))
# 1.1 Support for EPUB 3 EpubNav which might be ITEM_DOCUMENT (9) but with properties=['nav']
if not nav_items:
# Look in ITEM_DOCUMENT for items with 'nav' property
for item in self.book.get_items_of_type(ebooklib.ITEM_DOCUMENT):
if hasattr(item, 'get_type') and item.get_type() == ebooklib.ITEM_DOCUMENT:
# Check properties - ebooklib stores opf properties in list
# Some versions use item.properties, some need checking
props = getattr(item, 'properties', [])
if 'nav' in props:
nav_items.append(item)
# Look in ITEM_DOCUMENT for items with 'nav' property
for item in self.book.get_items_of_type(ebooklib.ITEM_DOCUMENT):
if (
hasattr(item, "get_type")
and item.get_type() == ebooklib.ITEM_DOCUMENT
):
# Check properties - ebooklib stores opf properties in list
# Some versions use item.properties, some need checking
props = getattr(item, "properties", [])
if "nav" in props:
nav_items.append(item)
if nav_items:
nav_item = next(
(
@@ -592,7 +604,11 @@ class EpubParser(BaseBookParser):
# 2. NCX in NAV
if not nav_item and nav_items:
ncx_in_nav = next(
(item for item in nav_items if item.get_name().lower().endswith(".ncx")),
(
item
for item in nav_items
if item.get_name().lower().endswith(".ncx")
),
None,
)
if ncx_in_nav:
@@ -624,9 +640,8 @@ class EpubParser(BaseBookParser):
if not nav_item or not nav_type:
raise ValueError("No navigation document found")
return nav_item, nav_type
return nav_item, nav_type
def _execute_nav_parsing_logic(self, nav_item, nav_type):
"""Parse the identified navigation item and slice content accordingly."""
@@ -826,10 +841,10 @@ class EpubParser(BaseBookParser):
"has_content": True,
},
)
def _process_epub_content_spine_fallback(self):
"""
Process EPUB content using the spine (linear reading order)
Process EPUB content using the spine (linear reading order)
when navigation processing fails.
"""
logging.info("Using spine fallback for EPUB processing.")
@@ -877,7 +892,7 @@ class EpubParser(BaseBookParser):
if text:
self.content_texts[doc_href] = text
self.content_lengths[doc_href] = calculate_text_length(text)
def get_chapters(self):
chapters = super().get_chapters()
if not chapters:
@@ -898,7 +913,7 @@ def get_book_parser(book_path, file_type=None):
Factory function to get the appropriate parser instance.
"""
book_path = os.path.normpath(os.path.abspath(book_path))
if not file_type:
if book_path.lower().endswith(".pdf"):
file_type = "pdf"
abogen/chunking.py
+13 -5
View File
@@ -80,7 +80,9 @@ def _normalize_whitespace(value: str) -> str:
def _normalize_chunk_text(value: str) -> str:
settings = get_runtime_settings()
config = build_apostrophe_config(settings=settings, base=_PIPELINE_APOSTROPHE_CONFIG)
config = build_apostrophe_config(
settings=settings, base=_PIPELINE_APOSTROPHE_CONFIG
)
normalized = normalize_for_pipeline(value, config=config, settings=settings)
return _normalize_whitespace(normalized)
@@ -158,12 +160,16 @@ def chunk_text(
# Sentence level – flatten paragraphs into individual sentences
sentence_index = 0
for para_index, paragraph in enumerate(list(_iter_paragraphs(text)) or [text.strip()]):
for para_index, paragraph in enumerate(
list(_iter_paragraphs(text)) or [text.strip()]
):
normalized_para = _normalize_whitespace(paragraph)
if not normalized_para:
continue
sentence_pairs = _split_sentences(paragraph) or [(normalized_para, paragraph)]
for sent_local_index, (normalized_sentence, raw_sentence) in enumerate(sentence_pairs):
for sent_local_index, (normalized_sentence, raw_sentence) in enumerate(
sentence_pairs
):
normalized_sentence = _normalize_whitespace(normalized_sentence)
if not normalized_sentence:
continue
@@ -203,7 +209,9 @@ def _build_display_pattern(text: str) -> Pattern[str]:
return pattern
def _search_source_span(source: str, normalized: str, start: int) -> Optional[Tuple[int, int]]:
def _search_source_span(
source: str, normalized: str, start: int
) -> Optional[Tuple[int, int]]:
if not normalized:
return None
pattern = _build_display_pattern(normalized)
@@ -264,4 +272,4 @@ def build_chunks_for_chapters(
chunk_prefix=str(prefix),
)
all_chunks.extend(chapter_chunks)
return all_chunks
return all_chunks
abogen/debug_tts_samples.py
+17 -6
View File
@@ -287,12 +287,12 @@ def build_debug_epub(dest_path: Path, *, title: str = "abogen debug samples") ->
dest_path.parent.mkdir(parents=True, exist_ok=True)
chapter_lines: List[str] = [
"<?xml version=\"1.0\" encoding=\"utf-8\"?>",
'<?xml version="1.0" encoding="utf-8"?>',
"<!DOCTYPE html>",
"<html xmlns=\"http://www.w3.org/1999/xhtml\">",
'<html xmlns="http://www.w3.org/1999/xhtml">',
"<head>",
f" <title>{title}</title>",
" <meta charset=\"utf-8\" />",
' <meta charset="utf-8" />',
"</head>",
"<body>",
f" <h1>{title}</h1>",
@@ -303,7 +303,11 @@ def build_debug_epub(dest_path: Path, *, title: str = "abogen debug samples") ->
safe_label = sample.label.replace("&", "and")
chapter_lines.append(f" <h2>{safe_label}</h2>")
chapter_lines.append(
" <p><strong>" + marker_for(sample.code) + "</strong> " + sample.text + "</p>"
" <p><strong>"
+ marker_for(sample.code)
+ "</strong> "
+ sample.text
+ "</p>"
)
chapter_lines += ["</body>", "</html>"]
@@ -366,8 +370,15 @@ def build_debug_epub(dest_path: Path, *, title: str = "abogen debug samples") ->
# Per EPUB spec: mimetype must be the first entry and stored (no compression).
with zipfile.ZipFile(dest_path, "w") as zf:
zf.write(tmp_path / "mimetype", "mimetype", compress_type=zipfile.ZIP_STORED)
for source in (meta_inf / "container.xml", oebps / "content.opf", oebps / "chapter.xhtml", oebps / "nav.xhtml"):
zf.write(
tmp_path / "mimetype", "mimetype", compress_type=zipfile.ZIP_STORED
)
for source in (
meta_inf / "container.xml",
oebps / "content.opf",
oebps / "chapter.xhtml",
oebps / "nav.xhtml",
):
arcname = str(source.relative_to(tmp_path)).replace("\\", "/")
zf.write(source, arcname, compress_type=zipfile.ZIP_DEFLATED)
abogen/entity_analysis.py
+53 -17
View File
@@ -82,7 +82,10 @@ _EXCLUDED_NER_LABELS = {
"QUANTITY",
}
_TITLE_PATTERN = re.compile(r"^(?:" + "|".join(re.escape(prefix) for prefix in _TITLE_PREFIXES) + r")\.?\s+", re.IGNORECASE)
_TITLE_PATTERN = re.compile(
r"^(?:" + "|".join(re.escape(prefix) for prefix in _TITLE_PREFIXES) + r")\.?\s+",
re.IGNORECASE,
)
_POSSESSIVE_PATTERN = re.compile(r"(?:'s|’s|\u2019s)$", re.IGNORECASE)
_NON_WORD_PATTERN = re.compile(r"[^\w\s'-]+")
_MULTI_SPACE_PATTERN = re.compile(r"\s+")
@@ -104,7 +107,9 @@ class EntityRecord:
forms: Counter = field(default_factory=Counter)
first_position: Optional[Tuple[int, int]] = None
def register(self, *, chapter_index: int, position: int, text: str, sentence: Optional[str]) -> None:
def register(
self, *, chapter_index: int, position: int, text: str, sentence: Optional[str]
) -> None:
self.count += 1
self.chapter_indices.add(chapter_index)
self.forms[text] += 1
@@ -163,7 +168,9 @@ def _resolve_model_name(language: str) -> str:
def _load_model(language: str) -> Any:
if spacy is None:
raise EntityModelError("spaCy is not available. Install spaCy to enable entity extraction.")
raise EntityModelError(
"spaCy is not available. Install spaCy to enable entity extraction."
)
model_name = _resolve_model_name(language)
cache_key = model_name.lower()
@@ -249,7 +256,9 @@ def _extract_propn_tokens(doc: Any) -> Iterable[Any]: # type: ignore[override]
yield doc[token.i : token.i + 1]
def _empty_result(cache_key: str, error: Optional[str] = None) -> EntityExtractionResult:
def _empty_result(
cache_key: str, error: Optional[str] = None
) -> EntityExtractionResult:
payload = {
"people": [],
"entities": [],
@@ -262,7 +271,9 @@ def _empty_result(cache_key: str, error: Optional[str] = None) -> EntityExtracti
"model": None,
}
errors = [error] if error else []
return EntityExtractionResult(summary=payload, cache_key=cache_key, elapsed=0.0, errors=errors)
return EntityExtractionResult(
summary=payload, cache_key=cache_key, elapsed=0.0, errors=errors
)
def extract_entities(
@@ -319,18 +330,25 @@ def extract_entities(
key = _token_key(cleaned)
if not key:
return
category = category_hint or ("people" if span.label_ == "PERSON" else "entities")
category = category_hint or (
"people" if span.label_ == "PERSON" else "entities"
)
record_key = (category, key)
record = records.get(record_key)
if record is None:
record = EntityRecord(
key=record_key,
label=cleaned,
kind=span.label_ or ("PROPN" if category == "entities" else "PERSON"),
kind=span.label_
or ("PROPN" if category == "entities" else "PERSON"),
category=category,
)
records[record_key] = record
sentence = span.sent.text if hasattr(span, "sent") and span.sent is not None else None
sentence = (
span.sent.text
if hasattr(span, "sent") and span.sent is not None
else None
)
record.register(
chapter_index=chapter_index,
position=span.start,
@@ -361,7 +379,9 @@ def extract_entities(
elapsed = time.perf_counter() - start
people_records = [record for record in records.values() if record.category == "people"]
people_records = [
record for record in records.values() if record.category == "people"
]
people_keys = {record.key[1] for record in people_records}
entity_records = [
record
@@ -374,10 +394,16 @@ def extract_entities(
people_records.sort(key=lambda rec: (-rec.count, rec.label))
entity_records.sort(key=lambda rec: (-rec.count, rec.label))
people_payload = [record.as_dict(index + 1) for index, record in enumerate(people_records)]
entity_payload = [record.as_dict(index + 1) for index, record in enumerate(entity_records)]
people_payload = [
record.as_dict(index + 1) for index, record in enumerate(people_records)
]
entity_payload = [
record.as_dict(index + 1) for index, record in enumerate(entity_records)
]
index_payload = sorted(tokens_for_index.values(), key=lambda item: (-item["count"], item["token"]))
index_payload = sorted(
tokens_for_index.values(), key=lambda item: (-item["count"], item["token"])
)
summary = {
"people": people_payload,
@@ -397,10 +423,14 @@ def extract_entities(
},
}
return EntityExtractionResult(summary=summary, cache_key=cache_key, elapsed=elapsed, errors=[])
return EntityExtractionResult(
summary=summary, cache_key=cache_key, elapsed=elapsed, errors=[]
)
def search_tokens(index: Mapping[str, Any], query: str, *, limit: int = 15) -> List[Dict[str, Any]]:
def search_tokens(
index: Mapping[str, Any], query: str, *, limit: int = 15
) -> List[Dict[str, Any]]:
tokens = index.get("tokens") if isinstance(index, Mapping) else None
if not isinstance(tokens, list) or not query:
return []
@@ -411,14 +441,18 @@ def search_tokens(index: Mapping[str, Any], query: str, *, limit: int = 15) -> L
for entry in tokens:
token_label = str(entry.get("token", ""))
normalized_label = token_label.lower()
if normalized in normalized_label or normalized in str(entry.get("normalized", "")):
if normalized in normalized_label or normalized in str(
entry.get("normalized", "")
):
results.append(entry)
if len(results) >= limit:
break
return results
def merge_override(summary: Mapping[str, Any], overrides: Mapping[str, Mapping[str, Any]]) -> Dict[str, Any]:
def merge_override(
summary: Mapping[str, Any], overrides: Mapping[str, Mapping[str, Any]]
) -> Dict[str, Any]:
if not isinstance(summary, Mapping):
return {"people": [], "entities": []}
merged_summary: Dict[str, Any] = dict(summary)
@@ -430,7 +464,9 @@ def merge_override(summary: Mapping[str, Any], overrides: Mapping[str, Mapping[s
for entry in items:
if not isinstance(entry, Mapping):
continue
normalized = _token_key(str(entry.get("normalized") or entry.get("label") or ""))
normalized = _token_key(
str(entry.get("normalized") or entry.get("label") or "")
)
merged = dict(entry)
if normalized and normalized in overrides:
merged_override = dict(overrides[normalized])
abogen/heteronym_overrides.py
+9 -3
View File
@@ -152,7 +152,9 @@ def _word_boundary_pattern(token: str) -> re.Pattern[str]:
if cached is not None:
return cached
escaped = re.escape(token)
pattern = re.compile(rf"(?i)(?<!\w){escaped}(?P<possessive>'s|\u2019s|\u2019)?(?!\w)")
pattern = re.compile(
rf"(?i)(?<!\w){escaped}(?P<possessive>'s|\u2019s|\u2019)?(?!\w)"
)
_WORD_BOUNDARY_CACHE[key] = pattern
return pattern
@@ -167,7 +169,9 @@ def _preserve_case(replacement: str, original: str) -> str:
return replacement
def _build_replacement_sentence(sentence: str, token: str, replacement_token: str) -> str:
def _build_replacement_sentence(
sentence: str, token: str, replacement_token: str
) -> str:
pattern = _word_boundary_pattern(token)
def _repl(match: re.Match[str]) -> str:
@@ -271,7 +275,9 @@ def extract_heteronym_overrides(
options: List[Dict[str, Any]] = []
for variant in spec.variants:
replacement_sentence = _build_replacement_sentence(
sentence, token=spec.token, replacement_token=variant.replacement_token
sentence,
token=spec.token,
replacement_token=variant.replacement_token,
)
options.append(
{
abogen/kokoro_text_normalization.py
+248 -110
View File
@@ -7,7 +7,18 @@ import os
import locale
from fractions import Fraction
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
List,
Mapping,
Optional,
Sequence,
Tuple,
)
import logging
logger = logging.getLogger(__name__)
@@ -16,7 +27,9 @@ try: # pragma: no cover - optional dependency guard
from num2words import num2words
except ImportError:
num2words = None
logger.warning("num2words library not found. Number normalization will be disabled.")
logger.warning(
"num2words library not found. Number normalization will be disabled."
)
except Exception as e: # pragma: no cover - graceful degradation
num2words = None
logger.error(f"Failed to import num2words: {e}")
@@ -41,34 +54,44 @@ CONTRACTION_CATEGORY_DEFAULTS: Dict[str, bool] = {
# ---------- Configuration Dataclass ----------
@dataclass
class ApostropheConfig:
contraction_mode: str = "expand" # expand|collapse|keep
possessive_mode: str = "keep" # keep|collapse
plural_possessive_mode: str = "collapse" # keep|collapse
irregular_possessive_mode: str = "keep" # keep|expand (expand just means keep or add hints; modify if needed)
sibilant_possessive_mode: str = "mark" # keep|mark|approx
fantasy_mode: str = "keep" # keep|mark|collapse_internal
acronym_possessive_mode: str = "keep" # keep|collapse_add_s
decades_mode: str = "expand" # keep|expand
leading_elision_mode: str = "expand" # keep|expand
ambiguous_past_modal_mode: str = "contextual" # keep|expand_prefer_would|expand_prefer_had|contextual
add_phoneme_hints: bool = True # Whether to emit markers like ‹IZ›
fantasy_marker: str = "‹FAP›" # Marker inserted if fantasy_mode == mark
sibilant_iz_marker: str = "‹IZ›" # Marker for /ɪz/ insertion
joiner: str = "" # Replacement used when collapsing internal apostrophes
lowercase_for_matching: bool = True # Normalize to lower for rule matching (not output)
protect_cultural_names: bool = True # Always keep O'Brien, D'Angelo, etc.
convert_numbers: bool = True # Convert grouped numbers such as 12,500 to words
convert_currency: bool = True # Convert currency symbols to words
remove_footnotes: bool = True # Remove footnote indicators
number_lang: str = "en" # num2words language code
year_pronunciation_mode: str = "american" # off|american (extend if needed)
contraction_categories: Dict[str, bool] = field(default_factory=lambda: dict(CONTRACTION_CATEGORY_DEFAULTS))
contraction_mode: str = "expand" # expand|collapse|keep
possessive_mode: str = "keep" # keep|collapse
plural_possessive_mode: str = "collapse" # keep|collapse
irregular_possessive_mode: str = (
"keep" # keep|expand (expand just means keep or add hints; modify if needed)
)
sibilant_possessive_mode: str = "mark" # keep|mark|approx
fantasy_mode: str = "keep" # keep|mark|collapse_internal
acronym_possessive_mode: str = "keep" # keep|collapse_add_s
decades_mode: str = "expand" # keep|expand
leading_elision_mode: str = "expand" # keep|expand
ambiguous_past_modal_mode: str = (
"contextual" # keep|expand_prefer_would|expand_prefer_had|contextual
)
add_phoneme_hints: bool = True # Whether to emit markers like ‹IZ›
fantasy_marker: str = "‹FAP›" # Marker inserted if fantasy_mode == mark
sibilant_iz_marker: str = "‹IZ›" # Marker for /ɪz/ insertion
joiner: str = "" # Replacement used when collapsing internal apostrophes
lowercase_for_matching: bool = (
True # Normalize to lower for rule matching (not output)
)
protect_cultural_names: bool = True # Always keep O'Brien, D'Angelo, etc.
convert_numbers: bool = True # Convert grouped numbers such as 12,500 to words
convert_currency: bool = True # Convert currency symbols to words
remove_footnotes: bool = True # Remove footnote indicators
number_lang: str = "en" # num2words language code
year_pronunciation_mode: str = "american" # off|american (extend if needed)
contraction_categories: Dict[str, bool] = field(
default_factory=lambda: dict(CONTRACTION_CATEGORY_DEFAULTS)
)
def is_contraction_enabled(self, category: str) -> bool:
return self.contraction_categories.get(category, True)
# ---------- Dictionaries / Patterns ----------
# Common contraction expansions (type + expansion words)
@@ -124,14 +147,18 @@ _FRACTION_RE = re.compile(
_CURRENCY_RE = re.compile(
r"(?P<symbol>[$£€¥])\s*(?P<amount>\d{1,3}(?:,\d{3})*(?:\.\d+)?)(?:\s+(?P<magnitude>hundred|thousand|million|billion|trillion|quadrillion))?(?!\d)",
re.IGNORECASE
re.IGNORECASE,
)
_URL_RE = re.compile(r"(https?://)?(www\.)?(?P<domain>[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+)(/[^\s]*)?")
_URL_RE = re.compile(
r"(https?://)?(www\.)?(?P<domain>[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+)(/[^\s]*)?"
)
_FOOTNOTE_RE = re.compile(r"([a-zA-Z]+)(\d+)")
_BRACKET_FOOTNOTE_RE = re.compile(r"\[\d+\]")
_ISO_DATE_RE = re.compile(r"\b(?P<year>\d{4})[/-](?P<month>\d{1,2})[/-](?P<day>\d{1,2})\b")
_ISO_DATE_RE = re.compile(
r"\b(?P<year>\d{4})[/-](?P<month>\d{1,2})[/-](?P<day>\d{1,2})\b"
)
_MDY_DATE_RE = re.compile(
r"\b(?P<month>Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:t(?:ember)?)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)\.?\s+"
r"(?P<day>\d{1,2})(?:st|nd|rd|th)?\s*,\s*(?P<year>\d{4})\b",
@@ -143,7 +170,9 @@ _TIME_RE = re.compile(
re.IGNORECASE,
)
_ADDRESS_ABBR_RE = re.compile(r"(?P<prefix>\b\w+\s+)(?P<abbr>St|Rd|Ave|Blvd|Ln|Dr)\.(?=\s*(?:,|\.|!|\?|$))")
_ADDRESS_ABBR_RE = re.compile(
r"(?P<prefix>\b\w+\s+)(?P<abbr>St|Rd|Ave|Blvd|Ln|Dr)\.(?=\s*(?:,|\.|!|\?|$))"
)
_MONTH_MAP = {
"jan": "January",
@@ -218,25 +247,27 @@ def _normalize_dates(text: str, language: str) -> str:
day = int(match.group("day"))
if not (1 <= month <= 12 and 1 <= day <= 31):
return match.group(0)
month_name = (
[
"January",
"February",
"March",
"April",
"May",
"June",
"July",
"August",
"September",
"October",
"November",
"December",
][month - 1]
)
month_name = [
"January",
"February",
"March",
"April",
"May",
"June",
"July",
"August",
"September",
"October",
"November",
"December",
][month - 1]
ordinal = _int_to_ordinal_words(day, language) or str(day)
year_words = _year_to_words_american(year, language)
return f"{month_name} {ordinal}, {year_words}" if us else f"{ordinal} {month_name} {year_words}"
return (
f"{month_name} {ordinal}, {year_words}"
if us
else f"{ordinal} {month_name} {year_words}"
)
def _format_mdy(match: re.Match[str]) -> str:
month_raw = str(match.group("month") or "").strip().lower().rstrip(".")
@@ -247,7 +278,11 @@ def _normalize_dates(text: str, language: str) -> str:
year = int(match.group("year"))
ordinal = _int_to_ordinal_words(day, language) or str(day)
year_words = _year_to_words_american(year, language)
return f"{month_name} {ordinal}, {year_words}" if us else f"{ordinal} {month_name} {year_words}"
return (
f"{month_name} {ordinal}, {year_words}"
if us
else f"{ordinal} {month_name} {year_words}"
)
out = _ISO_DATE_RE.sub(_format_iso, text)
out = _MDY_DATE_RE.sub(_format_mdy, out)
@@ -301,6 +336,7 @@ def _normalize_internet_slang(text: str) -> str:
return re.sub(r"\b(?:pls|plz)\b", _replace, text, flags=re.IGNORECASE)
_DECIMAL_NUMBER_RE = re.compile(
rf"(?<![\w{_NUMBER_RANGE_CLASS}/])(?P<number>-?(?:\d{{1,3}}(?:,\d{{3}})+|\d+)\.(?P<fraction>\d+))(?![\w{_NUMBER_RANGE_CLASS}/])"
)
@@ -308,7 +344,18 @@ _PLAIN_NUMBER_RE = re.compile(
rf"(?<![\w{_NUMBER_RANGE_CLASS}/])(?P<number>{_NUMBER_CORE_PATTERN})(?![\w{_NUMBER_RANGE_CLASS}/])"
)
_DIGIT_WORDS = ("zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine")
_DIGIT_WORDS = (
"zero",
"one",
"two",
"three",
"four",
"five",
"six",
"seven",
"eight",
"nine",
)
def _int_to_words(value: int, language: str) -> Optional[str]:
@@ -384,7 +431,9 @@ def _pluralize_fraction_word(base: str) -> str:
return base + "s"
def _fraction_denominator_word(denominator: int, numerator: int, language: str) -> Optional[str]:
def _fraction_denominator_word(
denominator: int, numerator: int, language: str
) -> Optional[str]:
"""Return spoken form for fraction denominator respecting plurality."""
if denominator == 0:
return None
@@ -405,7 +454,9 @@ def _fraction_denominator_word(denominator: int, numerator: int, language: str)
return _pluralize_fraction_word(base)
def _format_fraction_words(numerator: int, denominator: int, language: str) -> Optional[str]:
def _format_fraction_words(
numerator: int, denominator: int, language: str
) -> Optional[str]:
"""Return spoken representation of a simple fraction."""
if denominator == 0:
return None
@@ -492,7 +543,9 @@ def _coerce_int_token(token: str) -> Optional[int]:
return int(cleaned)
except ValueError:
return None
AMBIGUOUS_D_BASES = {"i","you","he","she","we","they"}
AMBIGUOUS_D_BASES = {"i", "you", "he", "she", "we", "they"}
AMBIGUOUS_S_BASES = {
"it",
"that",
@@ -520,6 +573,7 @@ def _is_ambiguous_s(token: str) -> bool:
low = token.lower()
return low.endswith("'s") and low[:-2] in AMBIGUOUS_S_BASES
# Irregular possessives that are not formed by simple + 's logic
IRREGULAR_POSSESSIVES = {
"children's": "children's",
@@ -527,12 +581,12 @@ IRREGULAR_POSSESSIVES = {
"women's": "women's",
"people's": "people's",
"geese's": "geese's",
"mouse's": "mouse's", # singular irregular
"mouse's": "mouse's", # singular irregular
}
SIBILANT_END_RE = re.compile(r"(?:[sxz]|(?:ch|sh))$", re.IGNORECASE)
DECADE_RE = re.compile(r"^'\d0s$", re.IGNORECASE) # '90s, '80s
DECADE_RE = re.compile(r"^'\d0s$", re.IGNORECASE) # '90s, '80s
LEADING_ELISION = {
"'tis": "it is",
"'twas": "it was",
@@ -546,7 +600,7 @@ CULTURAL_NAME_PATTERNS = [
re.compile(r"^O'[A-Z][a-z]+$"),
re.compile(r"^D'[A-Z][a-z]+$"),
re.compile(r"^L'[A-Za-z].*$"),
re.compile(r"^Mc[A-Z].*$"), # not apostrophe, but often relevant (kept anyway)
re.compile(r"^Mc[A-Z].*$"), # not apostrophe, but often relevant (kept anyway)
]
ACRONYM_POSSESSIVE_RE = re.compile(r"^[A-Z]{2,}'s$")
@@ -563,7 +617,7 @@ WORD_TOKEN_RE = re.compile(
APOSTROPHE_CHARS = "’`´ꞌʼ"
TERMINAL_PUNCTUATION = {".", "?", "!", "…", ";", ":"}
CLOSING_PUNCTUATION = '"\'”’)]}»›'
CLOSING_PUNCTUATION = "\"'”’)]}»›"
ELLIPSIS_SUFFIXES = ("...", "…")
_LINE_SPLIT_RE = re.compile(r"(\n+)")
@@ -584,29 +638,38 @@ SUFFIX_ABBREVIATIONS = {
}
_TITLE_PATTERN = re.compile(
r"\b(?P<abbr>" + "|".join(sorted(TITLE_ABBREVIATIONS.keys(), key=len, reverse=True)) + r")\.",
r"\b(?P<abbr>"
+ "|".join(sorted(TITLE_ABBREVIATIONS.keys(), key=len, reverse=True))
+ r")\.",
re.IGNORECASE,
)
_SUFFIX_PATTERN = re.compile(
r"\b(?P<abbr>" + "|".join(sorted(SUFFIX_ABBREVIATIONS.keys(), key=len, reverse=True)) + r")\.",
r"\b(?P<abbr>"
+ "|".join(sorted(SUFFIX_ABBREVIATIONS.keys(), key=len, reverse=True))
+ r")\.",
re.IGNORECASE,
)
# ---------- Utility Functions ----------
def normalize_unicode_apostrophes(text: str) -> str:
text = unicodedata.normalize("NFKC", text)
for ch in APOSTROPHE_CHARS:
text = text.replace(ch, "'")
return text
def tokenize(text: str) -> List[str]:
# Simple tokenization preserving punctuation tokens
return WORD_TOKEN_RE.findall(text)
def tokenize_with_spans(text: str) -> List[Tuple[str, int, int]]:
return [(match.group(0), match.start(), match.end()) for match in WORD_TOKEN_RE.finditer(text)]
return [
(match.group(0), match.start(), match.end())
for match in WORD_TOKEN_RE.finditer(text)
]
def _cleanup_spacing(text: str) -> str:
@@ -661,7 +724,9 @@ _ROMAN_COMPOSE_ORDER = [
(1, "I"),
]
_ROMAN_PREFIX_RE = re.compile(r"^(?P<roman>[IVXLCDM]+)(?P<sep>[\s\.:,;\-–—]*)", re.IGNORECASE)
_ROMAN_PREFIX_RE = re.compile(
r"^(?P<roman>[IVXLCDM]+)(?P<sep>[\s\.:,;\-–—]*)", re.IGNORECASE
)
_ROMAN_TOKEN_RE = re.compile(r"^[IVXLCDM]+$")
_ROMAN_CARDINAL_CONTEXTS = {
@@ -816,7 +881,9 @@ def _token_is_cardinal_context(token: str) -> bool:
return token.lower() in _ROMAN_CARDINAL_CONTEXTS
def _has_cardinal_leading_context(tokens: Sequence[Tuple[str, int, int]], index: int) -> bool:
def _has_cardinal_leading_context(
tokens: Sequence[Tuple[str, int, int]], index: int
) -> bool:
j = index - 1
while j >= 0:
token, *_ = tokens[j]
@@ -927,7 +994,9 @@ def _normalize_roman_numerals(text: str, language: str) -> str:
if len(token) >= 2:
if token.isupper():
convert = True
elif numeric_value <= 200 and _has_cardinal_leading_context(tokens, index):
elif numeric_value <= 200 and _has_cardinal_leading_context(
tokens, index
):
convert = True
elif len(token) == 1:
# Only convert single letters if context is strong
@@ -1002,7 +1071,10 @@ def _should_preserve_caps_word(word: str) -> bool:
upper_base = base.upper()
if upper_base in _ACRONYM_ALLOWLIST:
return True
if all(ch in _ROMAN_NUMERAL_LETTERS for ch in letters.upper()) and len(letters) <= 7:
if (
all(ch in _ROMAN_NUMERAL_LETTERS for ch in letters.upper())
and len(letters) <= 7
):
return True
return False
@@ -1128,7 +1200,7 @@ def normalize_roman_numeral_titles(
roman_token = match.group("roman")
separator = match.group("sep") or ""
rest = stripped[match.end():]
rest = stripped[match.end() :]
if not separator and rest and rest[:1].isalnum():
normalized.append(title)
@@ -1269,7 +1341,9 @@ def _apply_contraction_policy(
return expand()
def _assemble_contraction_expansion(base_text: str, surface_text: str, expansion_word: str) -> str:
def _assemble_contraction_expansion(
base_text: str, surface_text: str, expansion_word: str
) -> str:
if not expansion_word:
return base_text
@@ -1340,6 +1414,7 @@ def _classify_ambiguous_s(token: str, cfg: ApostropheConfig) -> Tuple[str, str]:
return candidates[0][0], token
def classify_token(token: str, cfg: ApostropheConfig) -> Tuple[str, str]:
"""
Classify apostrophe usage and propose normalized form.
@@ -1398,11 +1473,15 @@ def classify_token(token: str, cfg: ApostropheConfig) -> Tuple[str, str]:
return _case_preserving_words(token, words)
collapse_value = token.replace("'", "")
normalized = _apply_contraction_policy(token, category=category, cfg=cfg, expand=_expand, collapse=collapse_value)
normalized = _apply_contraction_policy(
token, category=category, cfg=cfg, expand=_expand, collapse=collapse_value
)
return category, normalized
# 6. Suffix contractions ('m handled separately)
if low.endswith("'m") and low[:-2] in SUFFIX_CONTRACTION_BASES.get("'m", ()): # pronoun I'm
if low.endswith("'m") and low[:-2] in SUFFIX_CONTRACTION_BASES.get(
"'m", ()
): # pronoun I'm
def _expand_m() -> str:
base = token[:-2]
@@ -1488,7 +1567,10 @@ def classify_token(token: str, cfg: ApostropheConfig) -> Tuple[str, str]:
return "other", token.replace("'", cfg.joiner)
return "other", token
def normalize_apostrophes(text: str, cfg: ApostropheConfig | None = None) -> Tuple[str, List[Tuple[str,str,str]]]:
def normalize_apostrophes(
text: str, cfg: ApostropheConfig | None = None
) -> Tuple[str, List[Tuple[str, str, str]]]:
"""
Normalize apostrophes per config.
Returns normalized text AND a list of (original_token, category, normalized_token)
@@ -1502,7 +1584,10 @@ def normalize_apostrophes(text: str, cfg: ApostropheConfig | None = None) -> Tup
token_entries = tokenize_with_spans(text)
use_contextual_s = cfg.contraction_mode == "expand"
use_contextual_d = cfg.contraction_mode == "expand" and cfg.ambiguous_past_modal_mode == "contextual"
use_contextual_d = (
cfg.contraction_mode == "expand"
and cfg.ambiguous_past_modal_mode == "contextual"
)
need_contextual = False
if (use_contextual_s or use_contextual_d) and token_entries:
@@ -1514,7 +1599,9 @@ def normalize_apostrophes(text: str, cfg: ApostropheConfig | None = None) -> Tup
need_contextual = True
break
contextual_resolutions = resolve_ambiguous_contractions(text) if need_contextual else {}
contextual_resolutions = (
resolve_ambiguous_contractions(text) if need_contextual else {}
)
results: List[Tuple[str, str, str]] = []
normalized_tokens: List[str] = []
@@ -1522,7 +1609,9 @@ def normalize_apostrophes(text: str, cfg: ApostropheConfig | None = None) -> Tup
for tok, start, end in token_entries:
category, norm = classify_token(tok, cfg)
resolution = contextual_resolutions.get((start, end)) if contextual_resolutions else None
resolution = (
contextual_resolutions.get((start, end)) if contextual_resolutions else None
)
if resolution is not None and cfg.contraction_mode == "expand":
if cfg.is_contraction_enabled(resolution.category):
category = resolution.category
@@ -1537,6 +1626,7 @@ def normalize_apostrophes(text: str, cfg: ApostropheConfig | None = None) -> Tup
normalized_text = _cleanup_spacing(" ".join(filtered))
return normalized_text, results
def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
if not text:
return text
@@ -1609,14 +1699,14 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
first_two = value // 100
last_two = value % 100
prefix = _words(first_two)
if not prefix:
return None
if last_two == 0:
return f"{prefix} hundred"
if last_two < 10:
# Use "oh X" format (e.g. "nineteen oh five")
return f"{prefix} oh {_DIGIT_WORDS[last_two]}"
@@ -1673,9 +1763,11 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
# Check for "address" or "addresses" as a whole word
has_address = bool(re.search(r"\baddress(es)?\b", context))
# Check for year markers as whole words
has_year_marker = bool(re.search(r"\b(bc|ad|bce|ce|b\.c\.|a\.d\.|b\.c\.e\.|c\.e\.)\b", context))
has_year_marker = bool(
re.search(r"\b(bc|ad|bce|ce|b\.c\.|a\.d\.|b\.c\.e\.|c\.e\.)\b", context)
)
should_try_year = True
if has_address and not has_year_marker:
@@ -1740,7 +1832,7 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
def _replace_currency(match: re.Match[str]) -> str:
if num2words is None:
return match.group(0)
symbol = match.group("symbol")
amount_str = match.group("amount").replace(",", "")
magnitude = match.group("magnitude")
@@ -1756,17 +1848,17 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
integer_part, fraction_part = amount_str.split(".", 1)
integer_val = int(integer_part)
integer_words = _int_to_words(integer_val, language)
# Spell out fraction digits
digit_words = []
for digit in fraction_part:
if digit.isdigit():
digit_words.append(_DIGIT_WORDS[int(digit)])
amount_spoken = f"{integer_words} point {' '.join(digit_words)}"
else:
amount_spoken = _int_to_words(int(amount), language)
currency_names = {
"$": "dollars",
"£": "pounds",
@@ -1774,7 +1866,7 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
"¥": "yen",
}
currency_name = currency_names.get(symbol, "dollars")
return f"{amount_spoken} {magnitude} {currency_name}"
# Handle $0.99 -> ninety-nine cents (avoid "zero dollars and...").
@@ -1787,7 +1879,9 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
except ValueError:
cents_value = 0
if cents_value > 0:
cents_words = _int_to_words(cents_value, language) or str(cents_value)
cents_words = _int_to_words(cents_value, language) or str(
cents_value
)
subunit = {
"$": "cent",
"€": "cent",
@@ -1797,7 +1891,11 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
if symbol == "£":
subunit = "pence" if cents_value != 1 else "penny"
else:
subunit = (subunit + "s") if cents_value != 1 and subunit not in {"pence", "yen"} else subunit
subunit = (
(subunit + "s")
if cents_value != 1 and subunit not in {"pence", "yen"}
else subunit
)
return f"{cents_words} {subunit}".strip()
currency_map = {
@@ -1807,12 +1905,14 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
"¥": "JPY",
}
currency_code = currency_map.get(symbol, "USD")
try:
# Always use float to avoid num2words treating int as cents (if that's what it does)
# or to ensure consistent behavior.
words = num2words(amount, to="currency", currency=currency_code, lang=language)
words = num2words(
amount, to="currency", currency=currency_code, lang=language
)
# Remove "zero cents" if present
# Patterns: ", zero cents", " and zero cents"
words = words.replace(", zero cents", "").replace(" and zero cents", "")
@@ -1829,21 +1929,21 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
parts = [p for p in domain.split(".") if p]
if parts and all(p.isalpha() and len(p) <= 2 for p in parts):
return match.group(0)
# Avoid matching numbers like 1.05 or 12.34.56
# If the domain consists only of digits and dots, ignore it (unless it has http/www prefix)
has_prefix = match.group(1) or match.group(2)
if not has_prefix and all(c.isdigit() or c == '.' or c == '-' for c in domain):
# Check if it really looks like a number (e.g. 1.05)
# If it has multiple dots like 1.2.3.4 it might be an IP, which we might want to speak as dot?
# But 1.05 is definitely a number.
# Let's be safe: if it looks like a float, skip.
try:
float(domain)
return match.group(0)
except ValueError:
pass
if not has_prefix and all(c.isdigit() or c == "." or c == "-" for c in domain):
# Check if it really looks like a number (e.g. 1.05)
# If it has multiple dots like 1.2.3.4 it might be an IP, which we might want to speak as dot?
# But 1.05 is definitely a number.
# Let's be safe: if it looks like a float, skip.
try:
float(domain)
return match.group(0)
except ValueError:
pass
if domain.startswith("www."):
domain = domain[4:]
spoken = domain.replace(".", " dot ")
@@ -1865,15 +1965,23 @@ def _normalize_grouped_numbers(text: str, cfg: ApostropheConfig) -> str:
normalized = _CURRENCY_RE.sub(_replace_currency, normalized)
if cfg.convert_numbers:
normalized = _NUMBER_RANGE_RE.sub(lambda m: _replace_number_range(m, language), normalized)
normalized = _NUMBER_SPACE_RANGE_RE.sub(lambda m: _replace_space_separated_range(m, language), normalized)
normalized = _FRACTION_RE.sub(lambda m: _replace_fraction(m, language), normalized)
normalized = _NUMBER_RANGE_RE.sub(
lambda m: _replace_number_range(m, language), normalized
)
normalized = _NUMBER_SPACE_RANGE_RE.sub(
lambda m: _replace_space_separated_range(m, language), normalized
)
normalized = _FRACTION_RE.sub(
lambda m: _replace_fraction(m, language), normalized
)
normalized = _DECIMAL_NUMBER_RE.sub(_replace_decimal, normalized)
normalized = _NUMBER_WITH_GROUP_RE.sub(_replace_grouped, normalized)
normalized = _PLAIN_NUMBER_RE.sub(_replace_plain, normalized)
normalized = _normalize_roman_numerals(normalized, language)
return normalized
def _normalize_dotted_acronyms(text: str) -> str:
def _replace(match: re.Match[str]) -> str:
value = match.group(0)
@@ -1882,8 +1990,10 @@ def _normalize_dotted_acronyms(text: str) -> str:
return re.sub(r"\b(?:[A-Z]\.){1,}[A-Z]\.?(?=\W|$)", _replace, text)
# ---------- Optional phoneme hint post-processing ----------
def apply_phoneme_hints(text: str, iz_marker="‹IZ›") -> str:
"""
Replace markers with an orthographic sequence that
@@ -1951,7 +2061,10 @@ _LLM_REGEX_TOOL = {
},
},
}
_LLM_REGEX_TOOL_CHOICE = {"type": "function", "function": {"name": _LLM_REGEX_TOOL_NAME}}
_LLM_REGEX_TOOL_CHOICE = {
"type": "function",
"function": {"name": _LLM_REGEX_TOOL_NAME},
}
_LLM_ALLOWED_REGEX_FLAGS = {
"IGNORECASE": re.IGNORECASE,
"MULTILINE": re.MULTILINE,
@@ -1974,7 +2087,9 @@ _SENTENCE_CAPTURE_RE = re.compile(r"[^.!?]+[.!?]+|[^.!?]+$", re.MULTILINE)
def _split_sentences_for_llm(text: str) -> List[str]:
sentences = [segment.strip() for segment in _SENTENCE_CAPTURE_RE.findall(text or "")]
sentences = [
segment.strip() for segment in _SENTENCE_CAPTURE_RE.findall(text or "")
]
return [segment for segment in sentences if segment]
@@ -1984,7 +2099,10 @@ def _normalize_with_llm(
settings: Mapping[str, Any],
config: ApostropheConfig,
) -> str:
from abogen.normalization_settings import build_llm_configuration, DEFAULT_LLM_PROMPT
from abogen.normalization_settings import (
build_llm_configuration,
DEFAULT_LLM_PROMPT,
)
from abogen.llm_client import generate_completion, LLMClientError
llm_config = build_llm_configuration(settings)
@@ -2001,7 +2119,7 @@ def _normalize_with_llm(
newline = ""
if raw_line.endswith(("\r", "\n")):
stripped_newline = raw_line.rstrip("\r\n")
newline = raw_line[len(stripped_newline):]
newline = raw_line[len(stripped_newline) :]
line_body = stripped_newline
else:
line_body = raw_line
@@ -2011,7 +2129,7 @@ def _normalize_with_llm(
continue
leading_ws = line_body[: len(line_body) - len(line_body.lstrip())]
trailing_ws = line_body[len(line_body.rstrip()):]
trailing_ws = line_body[len(line_body.rstrip()) :]
core = line_body[len(leading_ws) : len(line_body) - len(trailing_ws)]
sentences = _split_sentences_for_llm(core)
@@ -2136,7 +2254,11 @@ def _normalize_flag_field(raw: Any) -> List[str]:
seen: set[str] = set()
for value in raw_iterable:
candidate = str(value or "").strip().upper()
if not candidate or candidate not in _LLM_ALLOWED_REGEX_FLAGS or candidate in seen:
if (
not candidate
or candidate not in _LLM_ALLOWED_REGEX_FLAGS
or candidate in seen
):
continue
seen.add(candidate)
normalized.append(candidate)
@@ -2153,7 +2275,9 @@ def _apply_single_regex_replacement(text: str, spec: Mapping[str, Any]) -> str:
flag_names = spec.get("flags")
if isinstance(flag_names, str):
flag_iterable: Iterable[Any] = [flag_names]
elif isinstance(flag_names, Iterable) and not isinstance(flag_names, (bytes, str, Mapping)):
elif isinstance(flag_names, Iterable) and not isinstance(
flag_names, (bytes, str, Mapping)
):
flag_iterable = flag_names
else:
flag_iterable = []
@@ -2179,7 +2303,10 @@ def normalize_for_pipeline(
) -> str:
"""Normalize text for the synthesis pipeline with runtime settings."""
from abogen.normalization_settings import build_apostrophe_config, get_runtime_settings
from abogen.normalization_settings import (
build_apostrophe_config,
get_runtime_settings,
)
from abogen.llm_client import LLMClientError
runtime_settings = settings or get_runtime_settings()
@@ -2201,15 +2328,25 @@ def normalize_for_pipeline(
if mode == "off":
normalized = normalize_unicode_apostrophes(normalized)
if cfg.convert_numbers or cfg.convert_currency or getattr(cfg, "remove_footnotes", False):
if (
cfg.convert_numbers
or cfg.convert_currency
or getattr(cfg, "remove_footnotes", False)
):
normalized = _normalize_grouped_numbers(normalized, cfg)
normalized = _cleanup_spacing(normalized)
elif mode == "llm":
try:
normalized = _normalize_with_llm(normalized, settings=runtime_settings, config=cfg)
normalized = _normalize_with_llm(
normalized, settings=runtime_settings, config=cfg
)
except LLMClientError:
raise
if cfg.convert_numbers or cfg.convert_currency or getattr(cfg, "remove_footnotes", False):
if (
cfg.convert_numbers
or cfg.convert_currency
or getattr(cfg, "remove_footnotes", False)
):
normalized = _normalize_grouped_numbers(normalized, cfg)
normalized = _cleanup_spacing(normalized)
else:
@@ -2227,6 +2364,7 @@ def normalize_for_pipeline(
return normalized
# ---------- Example Usage ----------
if __name__ == "__main__":
@@ -2237,4 +2375,4 @@ if __name__ == "__main__":
print("Original:", sample)
print("Normalized:", norm_text)
for orig, cat, norm in details:
print(f"{orig:15} -> {norm:15} [{cat}]")
print(f"{orig:15} -> {norm:15} [{cat}]")
abogen/llm_client.py
+13 -5
View File
@@ -52,8 +52,10 @@ def _build_url(base_url: str, path: str) -> str:
normalized = _normalized_base_url(base_url)
trimmed_path = path.lstrip("/")
parsed = parse.urlparse(normalized)
if parsed.path.rstrip("/").lower().endswith("/v1") and trimmed_path.startswith("v1/"):
trimmed_path = trimmed_path[len("v1/"):]
if parsed.path.rstrip("/").lower().endswith("/v1") and trimmed_path.startswith(
"v1/"
):
trimmed_path = trimmed_path[len("v1/") :]
return parse.urljoin(normalized, trimmed_path)
@@ -77,7 +79,9 @@ def _perform_request(
if payload is not None:
data_bytes = json.dumps(payload).encode("utf-8")
request_headers = dict(headers or {})
req = request.Request(url, data=data_bytes, headers=request_headers, method=method.upper())
req = request.Request(
url, data=data_bytes, headers=request_headers, method=method.upper()
)
try:
with request.urlopen(req, timeout=timeout) as response:
body = response.read()
@@ -102,7 +106,9 @@ def list_models(configuration: LLMConfiguration) -> List[Dict[str, str]]:
raise LLMClientError("LLM configuration is incomplete")
url = _build_url(configuration.base_url, "v1/models")
headers = _build_headers(configuration.api_key)
payload = _perform_request("GET", url, headers=headers, timeout=configuration.timeout)
payload = _perform_request(
"GET", url, headers=headers, timeout=configuration.timeout
)
if not isinstance(payload, Mapping):
raise LLMClientError("Unexpected response when listing models")
data = payload.get("data")
@@ -153,7 +159,9 @@ def generate_completion(
if response_format:
payload["response_format"] = dict(response_format)
response = _perform_request("POST", url, headers=headers, payload=payload, timeout=configuration.timeout)
response = _perform_request(
"POST", url, headers=headers, payload=payload, timeout=configuration.timeout
)
if not isinstance(response, Mapping):
raise LLMClientError("Unexpected response from LLM")
choices = response.get("choices")
abogen/normalization_settings.py
+38 -11
View File
@@ -5,7 +5,10 @@ from dataclasses import replace
from functools import lru_cache
from typing import Any, Dict, Mapping, Optional
from abogen.kokoro_text_normalization import ApostropheConfig, CONTRACTION_CATEGORY_DEFAULTS
from abogen.kokoro_text_normalization import (
ApostropheConfig,
CONTRACTION_CATEGORY_DEFAULTS,
)
from abogen.llm_client import LLMConfiguration
from abogen.utils import load_config
@@ -149,7 +152,9 @@ def _extract_settings(source: Mapping[str, Any]) -> Dict[str, Any]:
elif isinstance(default, float):
extracted[key] = _coerce_float(raw_value, default)
else:
extracted[key] = str(raw_value or "") if isinstance(default, str) else raw_value
extracted[key] = (
str(raw_value or "") if isinstance(default, str) else raw_value
)
_apply_llm_migrations(extracted)
return extracted
@@ -177,20 +182,38 @@ def build_apostrophe_config(
config.convert_numbers = bool(settings.get("normalization_numbers", True))
config.convert_currency = bool(settings.get("normalization_currency", True))
config.remove_footnotes = bool(settings.get("normalization_footnotes", True))
config.year_pronunciation_mode = str(settings.get("normalization_numbers_year_style", "american") or "").strip().lower()
config.year_pronunciation_mode = (
str(settings.get("normalization_numbers_year_style", "american") or "")
.strip()
.lower()
)
config.add_phoneme_hints = bool(settings.get("normalization_phoneme_hints", True))
config.contraction_mode = "expand" if settings.get("normalization_apostrophes_contractions", True) else "keep"
config.contraction_mode = (
"expand"
if settings.get("normalization_apostrophes_contractions", True)
else "keep"
)
config.plural_possessive_mode = (
"collapse" if settings.get("normalization_apostrophes_plural_possessives", True) else "keep"
"collapse"
if settings.get("normalization_apostrophes_plural_possessives", True)
else "keep"
)
config.sibilant_possessive_mode = (
"mark" if settings.get("normalization_apostrophes_sibilant_possessives", True) else "keep"
"mark"
if settings.get("normalization_apostrophes_sibilant_possessives", True)
else "keep"
)
config.decades_mode = (
"expand" if settings.get("normalization_apostrophes_decades", True) else "keep"
)
config.decades_mode = "expand" if settings.get("normalization_apostrophes_decades", True) else "keep"
config.leading_elision_mode = (
"expand" if settings.get("normalization_apostrophes_leading_elisions", True) else "keep"
"expand"
if settings.get("normalization_apostrophes_leading_elisions", True)
else "keep"
)
config.ambiguous_past_modal_mode = (
"contextual" if config.contraction_mode == "expand" else "keep"
)
config.ambiguous_past_modal_mode = "contextual" if config.contraction_mode == "expand" else "keep"
category_flags = dict(CONTRACTION_CATEGORY_DEFAULTS)
for setting_key, category in _CONTRACTION_SETTING_MAP.items():
default_value = bool(_SETTINGS_DEFAULTS.get(setting_key, True))
@@ -205,11 +228,15 @@ def build_llm_configuration(settings: Mapping[str, Any]) -> LLMConfiguration:
base_url=str(settings.get("llm_base_url") or ""),
api_key=str(settings.get("llm_api_key") or ""),
model=str(settings.get("llm_model") or ""),
timeout=_coerce_float(settings.get("llm_timeout"), float(_SETTINGS_DEFAULTS["llm_timeout"])),
timeout=_coerce_float(
settings.get("llm_timeout"), float(_SETTINGS_DEFAULTS["llm_timeout"])
),
)
def apply_overrides(base: Mapping[str, Any], overrides: Mapping[str, Any]) -> Dict[str, Any]:
def apply_overrides(
base: Mapping[str, Any], overrides: Mapping[str, Any]
) -> Dict[str, Any]:
merged: Dict[str, Any] = dict(base)
for key, value in overrides.items():
if key not in _SETTINGS_DEFAULTS:
abogen/pronunciation_store.py
+38 -29
View File
@@ -31,7 +31,7 @@ def _migrate_legacy_sqlite(target_json_path: Path) -> None:
base_dir = Path(get_user_settings_dir())
except ModuleNotFoundError:
base_dir = Path(get_internal_cache_path("pronunciations"))
sqlite_path = base_dir / "pronunciations.db"
if not sqlite_path.exists():
return
@@ -39,23 +39,25 @@ def _migrate_legacy_sqlite(target_json_path: Path) -> None:
try:
conn = sqlite3.connect(sqlite_path)
conn.row_factory = sqlite3.Row
# Check if table exists
cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='overrides'")
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='overrides'"
)
if not cursor.fetchone():
conn.close()
return
cursor = conn.execute("SELECT * FROM overrides")
rows = cursor.fetchall()
data = {"version": _SCHEMA_VERSION, "overrides": {}}
for row in rows:
lang = row["language"]
if lang not in data["overrides"]:
data["overrides"][lang] = {}
entry = {
"id": str(row["id"]),
"normalized": row["normalized"],
@@ -70,16 +72,16 @@ def _migrate_legacy_sqlite(target_json_path: Path) -> None:
"updated_at": row["updated_at"],
}
data["overrides"][lang][row["normalized"]] = entry
conn.close()
# Save to JSON
with open(target_json_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# Rename old DB
sqlite_path.rename(sqlite_path.with_suffix(".db.bak"))
except Exception:
pass
@@ -110,11 +112,11 @@ def load_overrides(language: str, tokens: Iterable[str]) -> Dict[str, Dict[str,
normalized_tokens = {normalize_token(token) for token in tokens if token}
if not normalized_tokens:
return {}
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
results: Dict[str, Dict[str, Any]] = {}
for normalized in normalized_tokens:
if normalized in lang_overrides:
@@ -122,22 +124,27 @@ def load_overrides(language: str, tokens: Iterable[str]) -> Dict[str, Dict[str,
return results
def search_overrides(language: str, query: str, *, limit: int = 15) -> List[Dict[str, Any]]:
def search_overrides(
language: str, query: str, *, limit: int = 15
) -> List[Dict[str, Any]]:
if not query:
return []
query = query.lower()
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
matches = []
for entry in lang_overrides.values():
if query in entry["normalized"] or query in entry["token"].lower():
matches.append(entry)
# Sort by usage count desc, then updated_at desc
matches.sort(key=lambda x: (x.get("usage_count", 0), x.get("updated_at", 0)), reverse=True)
matches.sort(
key=lambda x: (x.get("usage_count", 0), x.get("updated_at", 0)),
reverse=True,
)
return matches[:limit]
@@ -153,15 +160,15 @@ def save_override(
normalized = normalize_token(token)
if not normalized:
raise ValueError("Provide a token to override")
timestamp = time.time()
with _DB_LOCK:
db = _load_db()
overrides = db.setdefault("overrides", {})
lang_overrides = overrides.setdefault(language, {})
existing = lang_overrides.get(normalized)
if existing:
entry = existing
entry["token"] = token
@@ -185,7 +192,7 @@ def save_override(
"updated_at": timestamp,
}
lang_overrides[normalized] = entry
_save_db(db)
return entry
@@ -194,11 +201,11 @@ def delete_override(*, language: str, token: str) -> None:
normalized = normalize_token(token)
if not normalized:
return
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
if normalized in lang_overrides:
del lang_overrides[normalized]
_save_db(db)
@@ -208,7 +215,7 @@ def all_overrides(language: str) -> List[Dict[str, Any]]:
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
results = list(lang_overrides.values())
results.sort(key=lambda x: x.get("updated_at", 0), reverse=True)
return results
@@ -218,11 +225,11 @@ def increment_usage(*, language: str, token: str, amount: int = 1) -> None:
normalized = normalize_token(token)
if not normalized:
return
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
if normalized in lang_overrides:
entry = lang_overrides[normalized]
entry["usage_count"] = entry.get("usage_count", 0) + amount
@@ -234,11 +241,13 @@ def get_override_stats(language: str) -> Dict[str, int]:
with _DB_LOCK:
db = _load_db()
lang_overrides = db.get("overrides", {}).get(language, {})
total = len(lang_overrides)
with_pronunciation = sum(1 for x in lang_overrides.values() if x.get("pronunciation"))
with_pronunciation = sum(
1 for x in lang_overrides.values() if x.get("pronunciation")
)
with_voice = sum(1 for x in lang_overrides.values() if x.get("voice"))
return {
"total": total,
"filtered": total,
abogen/spacy_contraction_resolver.py
+12 -5
View File
@@ -49,7 +49,9 @@ def _load_spacy_model(model: str = _DEFAULT_MODEL) -> Optional[Language]:
return nlp
def resolve_ambiguous_contractions(text: str, *, model: Optional[str] = None) -> Dict[Tuple[int, int], ContractionResolution]:
def resolve_ambiguous_contractions(
text: str, *, model: Optional[str] = None
) -> Dict[Tuple[int, int], ContractionResolution]:
"""Use spaCy to disambiguate ambiguous contractions in *text*.
Returns a mapping from (start, end) spans to their resolved expansion.
@@ -80,7 +82,9 @@ def resolve_ambiguous_contractions(text: str, *, model: Optional[str] = None) ->
return resolutions
def _resolution(prev: Token, token: Token, expansion_word: str, category: str, lemma_hint: str) -> Optional[ContractionResolution]:
def _resolution(
prev: Token, token: Token, expansion_word: str, category: str, lemma_hint: str
) -> Optional[ContractionResolution]:
if token is None or prev is None:
return None
@@ -186,10 +190,14 @@ def _resolve_apostrophe_d(token: Token) -> Optional[ContractionResolution]:
next_lemma = ""
if next_tag == "VB":
return _resolution(prev, token, "would", "contraction_modal_would", lemma or "will")
return _resolution(
prev, token, "would", "contraction_modal_would", lemma or "will"
)
if token.tag_ == "MD" or lemma in {"will", "would", "shall"}:
return _resolution(prev, token, "would", "contraction_modal_would", lemma or "will")
return _resolution(
prev, token, "would", "contraction_modal_would", lemma or "will"
)
if next_lemma in {"been", "gone", "had", "better"} or next_tag in {"VBN", "VBD"}:
return _resolution(prev, token, "had", "contraction_aux_have", "have")
@@ -255,4 +263,3 @@ def _context_prefers_had(token: Token) -> bool:
if next_lemma == "better":
return True
return False
abogen/spacy_utils.py
+3 -3
View File
@@ -83,12 +83,12 @@ def get_spacy_model(lang_code, log_callback=None):
model_name,
disable=["ner", "tagger", "lemmatizer", "attribute_ruler"],
)
# Ensure a sentence segmentation strategy is in place
# The parser provides sents, but if it's missing (unlikely for core models), fallback to sentencizer
if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names:
nlp.add_pipe("sentencizer")
_nlp_cache[lang_code] = nlp
return nlp
except OSError:
@@ -105,7 +105,7 @@ def get_spacy_model(lang_code, log_callback=None):
)
if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names:
nlp.add_pipe("sentencizer")
_nlp_cache[lang_code] = nlp
log(f"spaCy model '{model_name}' downloaded and loaded")
return nlp
abogen/speaker_analysis.py
+68 -21
View File
@@ -164,10 +164,16 @@ class SpeakerGuess:
sample_excerpt: Optional[str] = None,
) -> None:
self.count += 1
if _CONFIDENCE_RANK.get(confidence, 0) > _CONFIDENCE_RANK.get(self.confidence, 0):
if _CONFIDENCE_RANK.get(confidence, 0) > _CONFIDENCE_RANK.get(
self.confidence, 0
):
self.confidence = confidence
excerpt = sample_excerpt if sample_excerpt is not None else _build_excerpt(text, quote)
excerpt = (
sample_excerpt
if sample_excerpt is not None
else _build_excerpt(text, quote)
)
gender_hint = _format_gender_hint(male_votes, female_votes)
if excerpt:
payload = {"excerpt": excerpt, "gender_hint": gender_hint}
@@ -180,9 +186,13 @@ class SpeakerGuess:
self.male_votes += male_votes
if female_votes:
self.female_votes += female_votes
self.detected_gender = _derive_gender(self.male_votes, self.female_votes, self.detected_gender)
self.detected_gender = _derive_gender(
self.male_votes, self.female_votes, self.detected_gender
)
if self.gender in {"unknown", "male", "female"}:
self.gender = _derive_gender(self.male_votes, self.female_votes, self.gender)
self.gender = _derive_gender(
self.male_votes, self.female_votes, self.gender
)
def as_dict(self) -> Dict[str, Any]:
return {
@@ -211,7 +221,10 @@ class SpeakerAnalysis:
"version": self.version,
"narrator": self.narrator,
"assignments": dict(self.assignments),
"speakers": {speaker_id: guess.as_dict() for speaker_id, guess in self.speakers.items()},
"speakers": {
speaker_id: guess.as_dict()
for speaker_id, guess in self.speakers.items()
},
"suppressed": list(self.suppressed),
"stats": dict(self.stats),
}
@@ -226,7 +239,9 @@ def analyze_speakers(
) -> SpeakerAnalysis:
narrator_id = "narrator"
speaker_guesses: Dict[str, SpeakerGuess] = {
narrator_id: SpeakerGuess(speaker_id=narrator_id, label="Narrator", confidence="low")
narrator_id: SpeakerGuess(
speaker_id=narrator_id, label="Narrator", confidence="low"
)
}
label_index: Dict[str, str] = {"Narrator": narrator_id}
assignments: Dict[str, str] = {}
@@ -265,21 +280,31 @@ def analyze_speakers(
if record_id is None:
record_id = _dedupe_slug(_slugify(label), speaker_guesses)
label_index[label] = record_id
speaker_guesses[record_id] = SpeakerGuess(speaker_id=record_id, label=label)
speaker_guesses[record_id] = SpeakerGuess(
speaker_id=record_id, label=label
)
guess = speaker_guesses[record_id]
assignments[chunk_id] = record_id
unique_speakers.add(record_id)
if record_id != narrator_id and record_id != speaker_id and speaker_id == last_explicit:
if (
record_id != narrator_id
and record_id != speaker_id
and speaker_id == last_explicit
):
last_explicit = record_id
sample_excerpt = None
if record_id != narrator_id:
sample_excerpt = _select_sample_excerpt(ordered_chunks, index, guess.label, quote, confidence)
sample_excerpt = _select_sample_excerpt(
ordered_chunks, index, guess.label, quote, confidence
)
male_votes, female_votes = _count_gender_votes(text, guess.label)
guess.register_occurrence(confidence, text, quote, male_votes, female_votes, sample_excerpt)
guess.register_occurrence(
confidence, text, quote, male_votes, female_votes, sample_excerpt
)
active_speakers = [sid for sid in speaker_guesses if sid != narrator_id]
# Apply minimum occurrence threshold.
@@ -302,7 +327,9 @@ def analyze_speakers(
active_speakers = active_speakers[:max_speakers]
narrator_guess = speaker_guesses[narrator_id]
narrator_guess.count = sum(1 for value in assignments.values() if value == narrator_id)
narrator_guess.count = sum(
1 for value in assignments.values() if value == narrator_id
)
narrator_guess.confidence = "low"
stats = {
@@ -322,7 +349,9 @@ def analyze_speakers(
)
def _infer_chunk_speaker(text: str, last_explicit: Optional[str]) -> Tuple[Optional[str], str, Optional[str]]:
def _infer_chunk_speaker(
text: str, last_explicit: Optional[str]
) -> Tuple[Optional[str], str, Optional[str]]:
normalized = text.strip()
if not normalized:
return None, "low", None
@@ -376,7 +405,9 @@ def _match_name_near_quote(before: str, after: str) -> Optional[str]:
if _looks_like_name(name):
return name
match = re.search(rf"({_NAME_PATTERN})\s*,?\s*{_VERB_PATTERN}", leading, flags=re.IGNORECASE)
match = re.search(
rf"({_NAME_PATTERN})\s*,?\s*{_VERB_PATTERN}", leading, flags=re.IGNORECASE
)
if match:
name = match.group(1)
if _looks_like_name(name):
@@ -466,8 +497,14 @@ def _count_gender_votes(text: str, label: Optional[str]) -> Tuple[int, int]:
if windows:
mapped: List[Tuple[int, int]] = []
for start, end in windows:
start_idx = min(len(search_text) - 1, int(start * len(search_text) / max(len(ascii_text), 1)))
end_idx = min(len(search_text), int(end * len(search_text) / max(len(ascii_text), 1)))
start_idx = min(
len(search_text) - 1,
int(start * len(search_text) / max(len(ascii_text), 1)),
)
end_idx = min(
len(search_text),
int(end * len(search_text) / max(len(ascii_text), 1)),
)
mapped.append((start_idx, end_idx))
windows = mapped
else:
@@ -485,7 +522,9 @@ def _count_gender_votes(text: str, label: Optional[str]) -> Tuple[int, int]:
except IndexError:
content_start, content_end = match.span()
if content_start < content_end:
quote_spans.append((content_start, content_end, search_text[content_start:content_end]))
quote_spans.append(
(content_start, content_end, search_text[content_start:content_end])
)
normalized_label = _normalize_candidate_name(label) if label else None
normalized_label_lower = normalized_label.lower() if normalized_label else None
@@ -511,7 +550,11 @@ def _count_gender_votes(text: str, label: Optional[str]) -> Tuple[int, int]:
name_matches = list(re.finditer(_NAME_PATTERN, tail))
if name_matches:
last_name = _normalize_candidate_name(name_matches[-1].group(0))
if normalized_label_lower and last_name and last_name.lower() == normalized_label_lower:
if (
normalized_label_lower
and last_name
and last_name.lower() == normalized_label_lower
):
return 0.6
return 0.05
if re.search(r"[.!?]\s", prefix):
@@ -607,8 +650,12 @@ def _contains_dialogue_attribution(label: str, text: str, quote: Optional[str])
if not label or not text:
return False
escaped_label = re.escape(label)
direct_pattern = re.compile(rf"\b{escaped_label}\b\s+(?:{_VERB_PATTERN})\b", re.IGNORECASE)
reverse_pattern = re.compile(rf"(?:{_VERB_PATTERN})\s+\b{escaped_label}\b", re.IGNORECASE)
direct_pattern = re.compile(
rf"\b{escaped_label}\b\s+(?:{_VERB_PATTERN})\b", re.IGNORECASE
)
reverse_pattern = re.compile(
rf"(?:{_VERB_PATTERN})\s+\b{escaped_label}\b", re.IGNORECASE
)
colon_pattern = re.compile(rf"^\s*{escaped_label}\s*:\s*", re.IGNORECASE)
if colon_pattern.search(text):
@@ -679,7 +726,7 @@ def _format_gender_hint(male_votes: int, female_votes: int) -> str:
def _normalize_candidate_name(raw: str) -> Optional[str]:
if not raw:
return None
cleaned = raw.strip().strip('"“”\'’.,:;!')
cleaned = raw.strip().strip("\"“”'’.,:;!")
cleaned = re.sub(r"\s+", " ", cleaned).strip()
if not cleaned:
return None
@@ -712,4 +759,4 @@ def _normalize_candidate_name(raw: str) -> Optional[str]:
lowered = candidate.lower()
if lowered in _PRONOUN_LABELS or lowered in _STOP_LABELS:
return None
return candidate
return candidate
abogen/speaker_configs.py
+5 -1
View File
@@ -135,7 +135,11 @@ def _sanitize_speaker(entry: Dict[str, Any]) -> Dict[str, Any]:
normalized_langs.append(code.lower())
resolved_voice = entry.get("resolved_voice") or voice_formula or voice
resolved_label = label or entry.get("id") or ""
slug = entry.get("id") if isinstance(entry.get("id"), str) else slugify_label(resolved_label)
slug = (
entry.get("id")
if isinstance(entry.get("id"), str)
else slugify_label(resolved_label)
)
return {
"id": slug,
"label": resolved_label,
abogen/subtitle_utils.py
+2
View File
@@ -37,6 +37,7 @@ def clean_subtitle_text(text):
text = _CHAPTER_MARKER_PATTERN.sub("", text)
return text.strip()
def calculate_text_length(text):
# Use pre-compiled patterns for better performance
# Ignore chapter markers and metadata patterns in a single pass
@@ -48,6 +49,7 @@ def calculate_text_length(text):
char_count = len(text)
return char_count
def clean_text(text, *args, **kwargs):
# Remove metadata tags first
text = _METADATA_TAG_PATTERN.sub("", text)
abogen/text_extractor.py
+112 -35
View File
@@ -110,13 +110,19 @@ def _extract_from_string(raw: str, default_title: str) -> ExtractionResult:
normalized_tags = _normalize_metadata_keys(raw_metadata)
chapter_count = len(chapters)
artist_value = normalized_tags.get("artist")
authors = [name.strip() for name in artist_value.split(",") if name.strip()] if artist_value else []
authors = (
[name.strip() for name in artist_value.split(",") if name.strip()]
if artist_value
else []
)
metadata_source = MetadataSource(
title=normalized_tags.get("title") or default_title,
authors=authors,
publication_year=normalized_tags.get("year"),
)
metadata = _build_metadata_payload(metadata_source, chapter_count, "text", default_title)
metadata = _build_metadata_payload(
metadata_source, chapter_count, "text", default_title
)
metadata.update(normalized_tags)
if not chapters:
chapters = [ExtractedChapter(title=default_title, text="")]
@@ -148,9 +154,11 @@ def _split_chapters(content: str, default_title: str) -> List[ExtractedChapter]:
current_title = default_title
for match in matches:
segment = content[last_index:match.start()]
segment = content[last_index : match.start()]
if segment.strip():
chapters.append(ExtractedChapter(title=current_title, text=clean_text(segment)))
chapters.append(
ExtractedChapter(title=current_title, text=clean_text(segment))
)
current_title = match.group(1).strip() or default_title
last_index = match.end()
@@ -271,19 +279,27 @@ def _extract_markdown(path: Path) -> ExtractionResult:
raw = path.read_text(encoding=encoding, errors="replace")
metadata_source, chapters = _parse_markdown(raw, path.stem)
if not chapters:
chapters = [ExtractedChapter(title=metadata_source.title or path.stem, text=clean_text(raw))]
metadata = _build_metadata_payload(metadata_source, len(chapters), "markdown", path.stem)
chapters = [
ExtractedChapter(
title=metadata_source.title or path.stem, text=clean_text(raw)
)
]
metadata = _build_metadata_payload(
metadata_source, len(chapters), "markdown", path.stem
)
return ExtractionResult(chapters=chapters, metadata=metadata)
def _parse_markdown(raw: str, default_title: str) -> Tuple[MetadataSource, List[ExtractedChapter]]:
def _parse_markdown(
raw: str, default_title: str
) -> Tuple[MetadataSource, List[ExtractedChapter]]:
metadata = MetadataSource()
text = textwrap.dedent(raw)
frontmatter_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", text, re.DOTALL)
if frontmatter_match:
frontmatter = frontmatter_match.group(1)
_parse_markdown_frontmatter(frontmatter, metadata)
text_body = text[frontmatter_match.end():]
text_body = text[frontmatter_match.end() :]
else:
text_body = text
@@ -324,7 +340,11 @@ def _parse_markdown(raw: str, default_title: str) -> Tuple[MetadataSource, List[
chapters: List[ExtractedChapter] = []
for index, (header_id, start, name) in enumerate(header_positions):
end = header_positions[index + 1][1] if index + 1 < len(header_positions) else len(html)
end = (
header_positions[index + 1][1]
if index + 1 < len(header_positions)
else len(html)
)
section_html = html[start:end]
section_soup = BeautifulSoup(section_html, "html.parser")
header_tag = section_soup.find(attrs={"id": header_id})
@@ -336,7 +356,14 @@ def _parse_markdown(raw: str, default_title: str) -> Tuple[MetadataSource, List[
chapters.append(ExtractedChapter(title=name.strip(), text=section_text))
if not metadata.title:
first_h1 = next((header for header in headers if header.get("level") == 1 and header.get("name")), None)
first_h1 = next(
(
header
for header in headers
if header.get("level") == 1 and header.get("name")
),
None,
)
if first_h1:
metadata.title = str(first_h1["name"])
@@ -344,21 +371,27 @@ def _parse_markdown(raw: str, default_title: str) -> Tuple[MetadataSource, List[
def _parse_markdown_frontmatter(frontmatter: str, metadata: MetadataSource) -> None:
title_match = re.search(r"^title:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE)
title_match = re.search(
r"^title:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE
)
if title_match:
metadata.title = title_match.group(1).strip().strip('"\'')
metadata.title = title_match.group(1).strip().strip("\"'")
author_match = re.search(r"^author:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE)
author_match = re.search(
r"^author:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE
)
if author_match:
metadata.authors = [author_match.group(1).strip().strip('"\'')]
metadata.authors = [author_match.group(1).strip().strip("\"'")]
desc_match = re.search(r"^description:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE)
desc_match = re.search(
r"^description:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE
)
if desc_match:
metadata.description = desc_match.group(1).strip().strip('"\'')
metadata.description = desc_match.group(1).strip().strip("\"'")
date_match = re.search(r"^date:\s*(.+)$", frontmatter, re.MULTILINE | re.IGNORECASE)
if date_match:
date_str = date_match.group(1).strip().strip('\"\'')
date_str = date_match.group(1).strip().strip("\"'")
year_match = re.search(r"\b(19|20)\d{2}\b", date_str)
if year_match:
metadata.publication_year = year_match.group(0)
@@ -390,7 +423,9 @@ class EpubExtractor:
chapters = self._process_spine_fallback()
if not chapters:
chapters = [ExtractedChapter(title=self.path.stem, text="")]
metadata = _build_metadata_payload(metadata_source, len(chapters), "epub", self.path.stem)
metadata = _build_metadata_payload(
metadata_source, len(chapters), "epub", self.path.stem
)
metadata.setdefault("chapter_count", str(len(chapters)))
if metadata_source.series:
series_text = str(metadata_source.series).strip()
@@ -424,7 +459,9 @@ class EpubExtractor:
try:
author_items = self.book.get_metadata("DC", "creator")
if author_items:
metadata.authors = [author[0] for author in author_items if author and author[0]]
metadata.authors = [
author[0] for author in author_items if author and author[0]
]
except Exception as exc:
logger.debug("Failed to extract EPUB author metadata: %s", exc)
@@ -447,7 +484,9 @@ class EpubExtractor:
if date_items:
date_str = date_items[0][0]
year_match = re.search(r"\b(19|20)\d{2}\b", date_str)
metadata.publication_year = year_match.group(0) if year_match else date_str
metadata.publication_year = (
year_match.group(0) if year_match else date_str
)
except Exception as exc:
logger.debug("Failed to extract EPUB publication year metadata: %s", exc)
@@ -482,7 +521,16 @@ class EpubExtractor:
if name in {"calibre:series", "series"} and series_name is None:
series_name = candidate_text
continue
if name in {"calibre:series_index", "calibre:seriesindex", "series_index", "seriesindex"} and series_index is None:
if (
name
in {
"calibre:series_index",
"calibre:seriesindex",
"series_index",
"seriesindex",
}
and series_index is None
):
series_index = candidate_text
continue
@@ -533,7 +581,9 @@ class EpubExtractor:
self.spine_docs = self._build_spine_docs()
doc_order = {href: index for index, href in enumerate(self.spine_docs)}
doc_order_decoded = {urllib.parse.unquote(href): index for href, index in doc_order.items()}
doc_order_decoded = {
urllib.parse.unquote(href): index for href, index in doc_order.items()
}
nav_targets = self._collect_nav_targets(nav_soup, nav_type)
self._cache_relevant_documents(doc_order, nav_targets)
@@ -544,7 +594,9 @@ class EpubExtractor:
if not nav_map:
raise ValueError("NCX navigation missing <navMap>")
for nav_point in nav_map.find_all("navPoint", recursive=False):
self._parse_ncx_navpoint(nav_point, ordered_entries, doc_order, doc_order_decoded)
self._parse_ncx_navpoint(
nav_point, ordered_entries, doc_order, doc_order_decoded
)
else:
toc_nav = nav_soup.find("nav", attrs={"epub:type": "toc"})
if toc_nav is None:
@@ -558,7 +610,9 @@ class EpubExtractor:
if top_ol is None:
raise ValueError("TOC navigation missing <ol>")
for li in top_ol.find_all("li", recursive=False):
self._parse_html_nav_li(li, ordered_entries, doc_order, doc_order_decoded)
self._parse_html_nav_li(
li, ordered_entries, doc_order, doc_order_decoded
)
if not ordered_entries:
raise ValueError("No navigation entries found")
@@ -591,7 +645,9 @@ class EpubExtractor:
text = self._html_to_text(html_content)
if not text:
continue
title = self._resolve_document_title(html_content, fallback=f"Untitled Chapter {index + 1}")
title = self._resolve_document_title(
html_content, fallback=f"Untitled Chapter {index + 1}"
)
chapters.append(ExtractedChapter(title=title, text=text))
return chapters
@@ -605,7 +661,8 @@ class EpubExtractor:
(
item
for item in nav_items
if "nav" in item.get_name().lower() and item.get_name().lower().endswith((".xhtml", ".html"))
if "nav" in item.get_name().lower()
and item.get_name().lower().endswith((".xhtml", ".html"))
),
None,
)
@@ -627,7 +684,11 @@ class EpubExtractor:
if not nav_item and nav_items:
ncx_candidate = next(
(item for item in nav_items if item.get_name().lower().endswith(".ncx")),
(
item
for item in nav_items
if item.get_name().lower().endswith(".ncx")
),
None,
)
if ncx_candidate:
@@ -682,7 +743,9 @@ class EpubExtractor:
targets.append(href_value.split("#", 1)[0])
return targets
def _cache_relevant_documents(self, doc_order: Dict[str, int], nav_targets: List[str]) -> None:
def _cache_relevant_documents(
self, doc_order: Dict[str, int], nav_targets: List[str]
) -> None:
needed: set[str] = set(doc_order.keys())
for target in nav_targets:
needed.add(target)
@@ -718,7 +781,9 @@ class EpubExtractor:
if src:
base_href, fragment = src.split("#", 1) if "#" in src else (src, None)
doc_key, doc_idx = self._find_doc_key(base_href, doc_order, doc_order_decoded)
doc_key, doc_idx = self._find_doc_key(
base_href, doc_order, doc_order_decoded
)
if doc_key is not None and doc_idx is not None:
position = self._find_position_robust(doc_key, fragment)
ordered_entries.append(
@@ -738,7 +803,9 @@ class EpubExtractor:
)
for child_navpoint in nav_point.find_all("navPoint", recursive=False):
self._parse_ncx_navpoint(child_navpoint, ordered_entries, doc_order, doc_order_decoded)
self._parse_ncx_navpoint(
child_navpoint, ordered_entries, doc_order, doc_order_decoded
)
def _parse_html_nav_li(
self,
@@ -767,7 +834,9 @@ class EpubExtractor:
if src:
base_href, fragment = src.split("#", 1) if "#" in src else (src, None)
doc_key, doc_idx = self._find_doc_key(base_href, doc_order, doc_order_decoded)
doc_key, doc_idx = self._find_doc_key(
base_href, doc_order, doc_order_decoded
)
if doc_key is not None and doc_idx is not None:
position = self._find_position_robust(doc_key, fragment)
ordered_entries.append(
@@ -788,7 +857,9 @@ class EpubExtractor:
for child_ol in li_element.find_all("ol", recursive=False):
for child_li in child_ol.find_all("li", recursive=False):
self._parse_html_nav_li(child_li, ordered_entries, doc_order, doc_order_decoded)
self._parse_html_nav_li(
child_li, ordered_entries, doc_order, doc_order_decoded
)
def _find_doc_key(
self,
@@ -825,7 +896,9 @@ class EpubExtractor:
if pos != -1:
return pos
except Exception:
logger.debug("BeautifulSoup failed to locate id '%s' in %s", fragment_id, doc_href)
logger.debug(
"BeautifulSoup failed to locate id '%s' in %s", fragment_id, doc_href
)
safe_fragment_id = re.escape(fragment_id)
id_name_pattern = re.compile(
@@ -844,13 +917,17 @@ class EpubExtractor:
tag_start = html_content.rfind("<", 0, pos)
return tag_start if tag_start != -1 else pos
logger.warning("Anchor '%s' not found in %s. Defaulting to start.", fragment_id, doc_href)
logger.warning(
"Anchor '%s' not found in %s. Defaulting to start.", fragment_id, doc_href
)
return 0
def _slice_entries(self, ordered_entries: List[NavEntry]) -> List[ExtractedChapter]:
chapters: List[ExtractedChapter] = []
for index, entry in enumerate(ordered_entries):
next_entry = ordered_entries[index + 1] if index + 1 < len(ordered_entries) else None
next_entry = (
ordered_entries[index + 1] if index + 1 < len(ordered_entries) else None
)
slice_html = self._slice_entry(entry, next_entry)
text = self._html_to_text(slice_html)
if not text:
abogen/tts_supertonic.py
+19 -7
View File
@@ -47,7 +47,9 @@ def _resample_linear(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndar
return np.interp(x_new, x_old, audio).astype("float32", copy=False)
def _split_text(text: str, *, split_pattern: Optional[str], max_chunk_length: int) -> list[str]:
def _split_text(
text: str, *, split_pattern: Optional[str], max_chunk_length: int
) -> list[str]:
stripped = (text or "").strip()
if not stripped:
return []
@@ -81,13 +83,17 @@ def _split_text(text: str, *, split_pattern: Optional[str], max_chunk_length: in
return result
_UNSUPPORTED_CHARS_RE = re.compile(r"unsupported character\(s\):\s*(\[[^\]]*\])", re.IGNORECASE)
_UNSUPPORTED_CHARS_RE = re.compile(
r"unsupported character\(s\):\s*(\[[^\]]*\])", re.IGNORECASE
)
def _parse_unsupported_characters(error: BaseException) -> list[str]:
"""Best-effort extraction of unsupported characters from SuperTonic errors."""
message = " ".join(str(part) for part in getattr(error, "args", ()) if part is not None) or str(error)
message = " ".join(
str(part) for part in getattr(error, "args", ()) if part is not None
) or str(error)
match = _UNSUPPORTED_CHARS_RE.search(message)
if not match:
return []
@@ -127,8 +133,9 @@ def _configure_supertonic_gpu() -> None:
"""Patch supertonic's config to enable GPU acceleration if available."""
try:
import onnxruntime as ort
available = ort.get_available_providers()
# Use CUDA if available, skip TensorRT (requires extra libs not always present)
# TensorrtExecutionProvider may be listed as available but fail at runtime
# if TensorRT libraries (libnvinfer.so) are not installed
@@ -136,11 +143,12 @@ def _configure_supertonic_gpu() -> None:
if "CUDAExecutionProvider" in available:
providers.append("CUDAExecutionProvider")
providers.append("CPUExecutionProvider")
# Patch supertonic's config and loader before TTS import
# We must patch both because loader imports the value at module load time
import supertonic.config as supertonic_config
import supertonic.loader as supertonic_loader
supertonic_config.DEFAULT_ONNX_PROVIDERS = providers
supertonic_loader.DEFAULT_ONNX_PROVIDERS = providers
logger.info("Supertonic ONNX providers configured: %s", providers)
@@ -191,7 +199,9 @@ class SupertonicPipeline:
speed_value = max(0.7, min(2.0, speed_value))
style = self._tts.get_voice_style(voice_name=voice_name)
chunks = _split_text(text, split_pattern=split_pattern, max_chunk_length=self.max_chunk_length)
chunks = _split_text(
text, split_pattern=split_pattern, max_chunk_length=self.max_chunk_length
)
for chunk in chunks:
chunk_to_speak = chunk
removed: set[str] = set()
@@ -217,7 +227,9 @@ class SupertonicPipeline:
raise
removed.update(unsupported)
sanitized = _remove_unsupported_characters(chunk_to_speak, unsupported).strip()
sanitized = _remove_unsupported_characters(
chunk_to_speak, unsupported
).strip()
# If we didn't change anything, don't loop forever.
if sanitized == chunk_to_speak.strip():
abogen/utils.py
+20 -10
View File
@@ -14,6 +14,7 @@ from functools import lru_cache
from dotenv import load_dotenv, find_dotenv
def _load_environment() -> None:
explicit_path = os.environ.get("ABOGEN_ENV_FILE")
if explicit_path:
@@ -56,6 +57,7 @@ def detect_encoding(file_path):
encoding = detected_encoding if detected_encoding else "utf-8"
return encoding.lower()
def get_resource_path(package, resource):
"""
Get the path to a resource file, with fallback to local file system.
@@ -146,7 +148,9 @@ def get_user_settings_dir():
if os.path.exists(legacy_dir):
return ensure_directory(legacy_dir)
config_dir = user_config_dir("abogen", appauthor=False, roaming=True, ensure_exists=True)
config_dir = user_config_dir(
"abogen", appauthor=False, roaming=True, ensure_exists=True
)
return ensure_directory(config_dir)
@@ -250,7 +254,9 @@ def get_user_cache_root():
def get_internal_cache_root():
root = os.environ.get("ABOGEN_INTERNAL_CACHE_ROOT") or os.environ.get("XDG_CACHE_HOME")
root = os.environ.get("ABOGEN_INTERNAL_CACHE_ROOT") or os.environ.get(
"XDG_CACHE_HOME"
)
if root:
return ensure_directory(root)
home_dir = os.environ.get("HOME") or os.path.join("/tmp", "abogen-home")
@@ -274,9 +280,8 @@ def get_user_cache_path(folder=None):
@lru_cache(maxsize=1)
def get_user_output_root():
override = (
os.environ.get("ABOGEN_OUTPUT_DIR")
or os.environ.get("ABOGEN_OUTPUT_ROOT")
override = os.environ.get("ABOGEN_OUTPUT_DIR") or os.environ.get(
"ABOGEN_OUTPUT_ROOT"
)
if override:
return ensure_directory(override)
@@ -290,7 +295,10 @@ def get_user_output_path(folder=None):
return base
_sleep_procs: Dict[str, Optional[subprocess.Popen[str]]] = {"Darwin": None, "Linux": None} # Store sleep prevention processes
_sleep_procs: Dict[str, Optional[subprocess.Popen[str]]] = {
"Darwin": None,
"Linux": None,
} # Store sleep prevention processes
def clean_text(text, *args, **kwargs):
@@ -328,7 +336,9 @@ def create_process(cmd, stdin=None, text=True, capture_output=False):
# Determine shell usage: use shell only for string commands
use_shell = isinstance(cmd, str)
if use_shell:
logger.warning("Security Warning: create_process called with string command. Prefer using a list of arguments to avoid shell injection risks.")
logger.warning(
"Security Warning: create_process called with string command. Prefer using a list of arguments to avoid shell injection risks."
)
kwargs = {
"shell": use_shell,
@@ -439,18 +449,18 @@ def get_gpu_acceleration(enabled):
if not enabled:
return "GPU available but using CPU.", False
# Check for Apple Silicon MPS
if platform.system() == "Darwin" and platform.processor() == "arm":
if torch.backends.mps.is_available():
return "MPS GPU available and enabled.", True
else:
return "MPS GPU not available on Apple Silicon. Using CPU.", False
# Check for CUDA
if cuda_available():
return "CUDA GPU available and enabled.", True
# Gather CUDA diagnostic info if not available
try:
cuda_devices = torch.cuda.device_count()
abogen/voice_cache.py
+3 -1
View File
@@ -12,9 +12,11 @@ except Exception: # pragma: no cover - import fallback
LocalEntryNotFoundError = None # type: ignore[assignment]
if LocalEntryNotFoundError is None: # pragma: no cover - fallback for tests
class LocalEntryNotFoundError(Exception):
pass
from abogen.constants import VOICES_INTERNAL
_CACHE_LOCK = threading.Lock()
@@ -140,4 +142,4 @@ def _ensure_single_voice_asset(
pass
hf_hub_download(resume_download=True, **common_kwargs)
return True
return True
abogen/voice_profiles.py
+11 -3
View File
@@ -110,9 +110,17 @@ def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
return {
"provider": "supertonic",
"language": language,
"voice": _normalize_supertonic_voice(entry.get("voice") or entry.get("voice_name") or entry.get("name")),
"total_steps": _coerce_supertonic_steps(entry.get("total_steps") or entry.get("supertonic_total_steps") or entry.get("quality")),
"speed": _coerce_supertonic_speed(entry.get("speed") or entry.get("supertonic_speed")),
"voice": _normalize_supertonic_voice(
entry.get("voice") or entry.get("voice_name") or entry.get("name")
),
"total_steps": _coerce_supertonic_steps(
entry.get("total_steps")
or entry.get("supertonic_total_steps")
or entry.get("quality")
),
"speed": _coerce_supertonic_speed(
entry.get("speed") or entry.get("supertonic_speed")
),
}
voices = _normalize_voice_entries(entry.get("voices", []))