297 lines
12 KiB
Python
297 lines
12 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",
|
|
}
|
|
_ENGLISH_HINTS = {
|
|
"the", "and", "that", "with", "for", "are", "was", "were", "this", "which", "from",
|
|
"have", "has", "not", "but", "they", "their", "will", "would", "been", "into", "than",
|
|
}
|
|
_SPELL_WORD_RE = re.compile(r"[A-Za-zÄÖÜäöüß]{3,}")
|
|
# Two to six capitals (digits allowed after the first letter): SSD, HDMI, MP3, WLAN.
|
|
_ACRONYM_RE = re.compile(r"(?<![\w-])[A-ZÄÖÜ][A-ZÄÖÜ0-9]{1,5}(?![\w-])")
|
|
_ROMAN_RE = re.compile(r"[IVXLCDM]+")
|
|
|
|
|
|
@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 _language_hits(text: str) -> tuple[int, int, int]:
|
|
words = re.findall(r"[A-Za-zÄÖÜäöüß]+", (text or "").lower())
|
|
german = sum(1 for word in words if word in _GERMAN_HINTS)
|
|
english = sum(1 for word in words if word in _ENGLISH_HINTS)
|
|
return len(words), german, english
|
|
|
|
|
|
def detect_article_language(texts) -> str:
|
|
"""One language for the whole article; short fragments alone are unreliable."""
|
|
return detect_text_language(" ".join(t for t in texts if t))
|
|
|
|
|
|
def language_for_block(text: str, default: str) -> str:
|
|
"""The article language, unless this block is decisively in the other one.
|
|
|
|
An English quotation inside a German article (or the reverse) must not be
|
|
spoken with the wrong acronym expansions, but a short fragment or a
|
|
heading is never enough evidence to override the article.
|
|
"""
|
|
if default not in ("de", "en"):
|
|
return detect_text_language(text)
|
|
total, german, english = _language_hits(text)
|
|
if total < 12:
|
|
return default
|
|
if default == "de" and english >= 3 and german == 0:
|
|
return "en"
|
|
if default == "en" and german >= 3 and english == 0:
|
|
return "de"
|
|
return default
|
|
|
|
|
|
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
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _english_spellchecker():
|
|
try:
|
|
from spellchecker import SpellChecker
|
|
return SpellChecker(language="en", distance=1)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def suspect_ocr_tokens(text: str, language: str = "auto", ignore=()) -> list[str]:
|
|
"""Return dictionary-unknown tokens worth checking against the page image.
|
|
|
|
This deliberately does not provide replacements. Magazine text contains
|
|
names, brands and technical compounds that a spellchecker cannot know;
|
|
dictionary output is only a trigger for a second OCR reading of the real
|
|
crop, never authority to rewrite printed text by itself.
|
|
"""
|
|
from core.spelling import find_spelling_issues
|
|
skip = {str(w).casefold() for w in ignore}
|
|
return list(dict.fromkeys(i["token"] for i in find_spelling_issues(text or "", language)
|
|
if i["token"].casefold() not in skip))
|
|
|
|
|
|
def recurring_tokens(texts, minimum: int = 2) -> set[str]:
|
|
"""Words that appear at least `minimum` times across `texts` (lower-cased).
|
|
|
|
A misreading differs each time it happens; a word printed again and again is a name, a
|
|
brand or a term of the article's subject, however unfamiliar it is to a dictionary.
|
|
"""
|
|
counts: dict[str, int] = {}
|
|
for text in texts:
|
|
for token in _SPELL_WORD_RE.findall(text or ""):
|
|
counts[token.casefold()] = counts.get(token.casefold(), 0) + 1
|
|
return {token for token, n in counts.items() if n >= minimum}
|
|
|
|
|
|
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 spell_letters(acronym: str) -> str:
|
|
"""Spell an acronym letter by letter (``SSD`` -> ``S S D``) for the TTS engine."""
|
|
return " ".join(acronym)
|
|
|
|
|
|
def _mapping_for(lang: str, glossary: dict[str, str] | None) -> dict[str, str]:
|
|
db = pronunciation_db()
|
|
mapping = dict(db.get(lang) or {})
|
|
for acronym in db.get(f"{lang}_letters") or []:
|
|
mapping.setdefault(str(acronym), spell_letters(str(acronym)))
|
|
# The article's own glossary wins: it was decided from that article's context.
|
|
for short, spoken in (glossary or {}).items():
|
|
if isinstance(short, str) and isinstance(spoken, str) and short.strip() and spoken.strip():
|
|
mapping[short.strip()] = spoken.strip()
|
|
return mapping
|
|
|
|
|
|
def expand_abbreviations(text: str, language: str = "auto",
|
|
glossary: dict[str, str] | None = None) -> str:
|
|
lang = detect_text_language(text) if language == "auto" else language
|
|
mapping = _mapping_for(lang, glossary)
|
|
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, lambda _m, spoken=spoken: spoken, result)
|
|
return result
|
|
|
|
|
|
def find_unknown_acronyms(text: str, language: str = "auto",
|
|
glossary: dict[str, str] | None = None) -> list[dict]:
|
|
"""All-caps tokens the pronunciation database and glossary do not cover yet.
|
|
|
|
Each entry carries one sentence of context so an LLM (or the reviewer) can
|
|
decide how to speak it. Headline-style all-caps sentences and Roman
|
|
numerals are skipped: capitals there are typography, not acronyms.
|
|
"""
|
|
lang = detect_text_language(text) if language == "auto" else language
|
|
known = set(_mapping_for(lang, glossary))
|
|
found: dict[str, dict] = {}
|
|
for sentence in re.split(r"(?<=[.!?…])\s+|\n+", text or ""):
|
|
words = re.findall(r"[A-Za-zÄÖÜäöüß]{2,}", sentence)
|
|
if len(words) >= 3 and sum(w.isupper() for w in words) / len(words) > 0.6:
|
|
continue
|
|
for match in _ACRONYM_RE.finditer(sentence):
|
|
token = match.group(0)
|
|
if token in known or _ROMAN_RE.fullmatch(token) or token.isdigit():
|
|
continue
|
|
entry = found.setdefault(token, {"acronym": token, "count": 0, "context": sentence.strip()[:240]})
|
|
entry["count"] += 1
|
|
return sorted(found.values(), key=lambda e: (-e["count"], e["acronym"]))
|
|
|
|
|
|
_MAX_SPOKEN = 80
|
|
_MAX_GLOSSARY = 300
|
|
|
|
|
|
def clean_glossary(raw) -> dict[str, str]:
|
|
"""Validate a user- or LLM-supplied glossary: bounded, single-line, str -> str."""
|
|
if not isinstance(raw, dict):
|
|
return {}
|
|
out: dict[str, str] = {}
|
|
for short, spoken in raw.items():
|
|
if not (isinstance(short, str) and isinstance(spoken, str)):
|
|
continue
|
|
short, spoken = short.strip(), " ".join(spoken.split())
|
|
if 1 <= len(short) <= 12 and 1 <= len(spoken) <= _MAX_SPOKEN and "\x00" not in spoken:
|
|
out[short] = spoken
|
|
if len(out) >= _MAX_GLOSSARY:
|
|
break
|
|
return out
|
|
|
|
|
|
def parse_acronym_decisions(items, unknown: list[dict]) -> dict[str, str]:
|
|
"""Turn an LLM answer into glossary entries, trusting only what can be checked.
|
|
|
|
An answer for an acronym we did not ask about is dropped. ``letters`` and
|
|
``word`` are decided here, not by the model, so the model cannot inject
|
|
arbitrary text through them; only ``expansion`` carries model-written words.
|
|
"""
|
|
asked = {entry["acronym"] for entry in unknown}
|
|
decided: dict[str, str] = {}
|
|
for item in items if isinstance(items, list) else []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
acronym, kind, spoken = item.get("acronym"), item.get("kind"), item.get("spoken")
|
|
if acronym not in asked:
|
|
continue
|
|
if kind == "letters":
|
|
decided[acronym] = spell_letters(acronym)
|
|
elif kind == "word":
|
|
decided[acronym] = acronym
|
|
elif kind == "expansion" and isinstance(spoken, str):
|
|
decided.update(clean_glossary({acronym: spoken}))
|
|
return decided
|
|
|
|
|
|
def prepare_text_for_tts(text: str, language: str = "auto",
|
|
glossary: dict[str, str] | None = None) -> str:
|
|
return expand_abbreviations(normalize_ocr_text(text, language), language, glossary)
|