135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
"""Conservative OCR cleanup and spoken-form expansion for magazine TTS."""
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
import json
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
_DB_PATH = Path(__file__).with_name("pronunciations.json")
|
|
_WORD_RE = re.compile(r"\b[A-Za-zÄÖÜäöüß]{3,}\b")
|
|
_LINEBREAK_HYPHEN_RE = re.compile(
|
|
r"(?<=[A-Za-zÄÖÜäöüß])-(?P<gap>\s+)(?P<next>[a-zäöüß][A-Za-zÄÖÜäöüß]*)"
|
|
)
|
|
_SUSPENDED_COMPOUND_WORDS = {"und", "oder", "beziehungsweise", "sowie"}
|
|
_GERMAN_HINTS = {
|
|
"aber", "auch", "auf", "aus", "bei", "das", "dass", "der", "die", "ein",
|
|
"eine", "einer", "einen", "für", "hat", "ist", "mit", "nicht", "oder", "sich",
|
|
"sie", "und", "von", "werden", "wie", "zu",
|
|
}
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def pronunciation_db() -> dict:
|
|
try:
|
|
data = json.loads(_DB_PATH.read_text("utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def remove_linebreak_hyphens(text: str) -> str:
|
|
"""Fuse a word split at a printed line end, preserving real compounds."""
|
|
text = (text or "").replace("\u00ad", "")
|
|
|
|
def fuse(match: re.Match) -> str:
|
|
# ``Hardware- und Softwareentwicklung`` is a real suspended compound,
|
|
# not a printed line break. OCR historically flattened newlines to a
|
|
# space, so keep this small grammatical exception for saved layouts.
|
|
if match.group("next").casefold() in _SUSPENDED_COMPOUND_WORDS:
|
|
return match.group(0)
|
|
return match.group("next")
|
|
|
|
return _LINEBREAK_HYPHEN_RE.sub(fuse, text)
|
|
|
|
|
|
def detect_text_language(text: str) -> str:
|
|
words = re.findall(r"[A-Za-zÄÖÜäöüß]+", (text or "").lower())
|
|
if any(any(c in w for c in "äöüß") for w in words):
|
|
return "de"
|
|
hits = sum(1 for word in words if word in _GERMAN_HINTS)
|
|
return "de" if hits >= max(2, len(words) // 40) else "en"
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _german_spellchecker():
|
|
try:
|
|
from spellchecker import SpellChecker
|
|
return SpellChecker(language="de", distance=1)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _match_case(source: str, replacement: str) -> str:
|
|
if source.isupper() and len(source) > 1:
|
|
return replacement.upper()
|
|
if source[:1].isupper():
|
|
return replacement[:1].upper() + replacement[1:]
|
|
return replacement
|
|
|
|
|
|
def _umlaut_variants(word: str) -> set[str]:
|
|
positions = [(i, {"a": "ä", "o": "ö", "u": "ü"}[c])
|
|
for i, c in enumerate(word) if c in "aou"]
|
|
if not positions or len(positions) > 5:
|
|
return set()
|
|
variants = set()
|
|
for count in range(1, len(positions) + 1):
|
|
for chosen in itertools.combinations(positions, count):
|
|
chars = list(word)
|
|
for i, replacement in chosen:
|
|
chars[i] = replacement
|
|
variants.add("".join(chars))
|
|
return variants
|
|
|
|
|
|
def restore_german_umlauts(text: str) -> str:
|
|
"""Restore only dictionary-backed, unambiguous missing umlauts."""
|
|
explicit = {str(k).lower(): str(v) for k, v in
|
|
(pronunciation_db().get("de_umlaut_repairs") or {}).items()}
|
|
checker = _german_spellchecker()
|
|
|
|
def replace(match: re.Match) -> str:
|
|
token = match.group(0)
|
|
lower = token.lower()
|
|
if any(c in lower for c in "äöüß") or token.isupper():
|
|
return token
|
|
if lower in explicit:
|
|
return _match_case(token, explicit[lower])
|
|
# An unknown title-cased token is more likely to be a person's name,
|
|
# product or place than a common noun. Explicit reviewed repairs above
|
|
# may still correct a sentence-opening word such as ``Jahrlich``.
|
|
if token != lower:
|
|
return token
|
|
if checker is None or lower in checker:
|
|
return token
|
|
known = [candidate for candidate in _umlaut_variants(lower) if candidate in checker]
|
|
if len(known) != 1:
|
|
return token
|
|
return _match_case(token, known[0])
|
|
|
|
return _WORD_RE.sub(replace, text or "")
|
|
|
|
|
|
def normalize_ocr_text(text: str, language: str = "auto") -> str:
|
|
text = remove_linebreak_hyphens(text)
|
|
lang = detect_text_language(text) if language == "auto" else language
|
|
return restore_german_umlauts(text) if lang == "de" else text
|
|
|
|
|
|
def expand_abbreviations(text: str, language: str = "auto") -> str:
|
|
lang = detect_text_language(text) if language == "auto" else language
|
|
mapping = pronunciation_db().get(lang) or {}
|
|
result = text or ""
|
|
# Longest first so ``LLMs`` wins over ``LLM`` and dotted abbreviations
|
|
# are consumed as one unit. Match tokens only, never substrings in names.
|
|
for short, spoken in sorted(mapping.items(), key=lambda item: len(item[0]), reverse=True):
|
|
pattern = r"(?<!\w)" + re.escape(short) + r"(?!\w)"
|
|
result = re.sub(pattern, spoken, result)
|
|
return result
|
|
|
|
|
|
def prepare_text_for_tts(text: str, language: str = "auto") -> str:
|
|
return expand_abbreviations(normalize_ocr_text(text, language), language)
|