325 lines
14 KiB
Python
325 lines
14 KiB
Python
"""German/English spelling findings for magazine text.
|
||
|
||
Magazine text is full of names, brands, product codes and long German
|
||
compounds that no dictionary contains, so this module never rewrites text on
|
||
its own. It produces *findings* (token, position, suggestions) that a person
|
||
sees marked on the page, or that an agent (vision re-read / LLM) may resolve
|
||
under the guards in ``accept_agent_fix``.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
from core import pronunciations as _pron
|
||
from core.pronunciations import (
|
||
_match_case,
|
||
_umlaut_variants,
|
||
detect_article_language,
|
||
detect_text_language,
|
||
language_for_block,
|
||
)
|
||
|
||
_TOKEN_RE = re.compile(r"[A-Za-zÄÖÜäöüß]+(?:-[A-Za-zÄÖÜäöüß]+)*")
|
||
_MIN_LENGTH = 4
|
||
_MIN_PART = 4
|
||
_MAX_SUGGESTIONS = 4
|
||
_LINKING = ("", "s", "es", "n", "en", "e", "er")
|
||
_STEM_ENDINGS = ("", "e", "en", "ung", "s", "er")
|
||
_MAX_ISSUES_PER_TEXT = 60
|
||
_RECURRING = 3
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def known_names() -> frozenset[str]:
|
||
"""Company, product and technology names the dictionaries cannot know (core/known_names.json)."""
|
||
try:
|
||
data = json.loads((Path(__file__).with_name("known_names.json")).read_text("utf-8"))
|
||
return frozenset(str(n).casefold() for n in data.get("names", []))
|
||
except (OSError, ValueError):
|
||
return frozenset()
|
||
|
||
|
||
def _checker(language: str):
|
||
return _pron._german_spellchecker() if language == "de" else _pron._english_spellchecker()
|
||
|
||
|
||
_INFLECTIONS = ("e", "en", "er", "es", "em", "n", "s", "t", "te", "ten", "tes", "st", "ste", "sten", "end", "ende")
|
||
_PREFIXES = ("ein", "aus", "auf", "ab", "an", "be", "ge", "ver", "zer", "ent", "er", "vor", "nach", "zu",
|
||
"mit", "über", "unter", "um", "durch", "weg", "her", "hin", "zurück", "zusammen", "wieder", "fort")
|
||
|
||
|
||
def _known(word: str, checker) -> bool:
|
||
return word.casefold() in checker
|
||
|
||
|
||
def _known_form(word: str, checker, language: str) -> bool:
|
||
"""Known as printed, or as an inflected / prefixed / ß-variant form of a known word.
|
||
|
||
The frequency word lists behind the checker hold common surface forms but
|
||
miss many inflections (``ließe``, ``einfließen``). Accepting a form whose
|
||
stem or prefix-stripped stem is a real word trades a few missed typos for
|
||
far fewer false alarms, which is the right way round for a marking tool.
|
||
"""
|
||
w = word.casefold()
|
||
if w in checker:
|
||
return True
|
||
if language != "de":
|
||
return False
|
||
if "ß" in w and w.replace("ß", "ss") in checker:
|
||
return True
|
||
if "ss" in w and w.replace("ss", "ß") in checker:
|
||
return True
|
||
for suffix in _INFLECTIONS:
|
||
if w.endswith(suffix) and len(w) - len(suffix) >= 4:
|
||
stem = w[: -len(suffix)]
|
||
if stem in checker or stem + "e" in checker:
|
||
return True
|
||
for prefix in _PREFIXES:
|
||
if w.startswith(prefix) and len(w) - len(prefix) >= 5:
|
||
rest = w[len(prefix):]
|
||
if rest in checker or any(rest.endswith(x) and rest[: -len(x)] in checker
|
||
for x in _INFLECTIONS if len(rest) - len(x) >= 4):
|
||
return True
|
||
return False
|
||
|
||
|
||
def is_known_compound(word: str, checker) -> bool:
|
||
"""True when a German word splits into known parts (with a linking element).
|
||
|
||
``Softwareentwicklung`` is not in any word list but is plainly
|
||
``Software`` + ``Entwicklung``. Every part must be a known form of at least
|
||
four letters, so random letter runs do not pass by accident.
|
||
"""
|
||
w = word.casefold()
|
||
n = len(w)
|
||
if n < 2 * _MIN_PART or w in checker:
|
||
return False
|
||
memo: dict[int, bool] = {}
|
||
|
||
def splits(pos: int) -> bool:
|
||
if pos in memo:
|
||
return memo[pos]
|
||
ok = False
|
||
for end in range(pos + _MIN_PART, n + 1):
|
||
if not _known_form(w[pos:end], checker, "de"):
|
||
continue
|
||
if end == n:
|
||
ok = True
|
||
break
|
||
if any(w.startswith(link, end) and end + len(link) < n and splits(end + len(link))
|
||
for link in _LINKING):
|
||
ok = True
|
||
break
|
||
memo[pos] = ok
|
||
return ok
|
||
|
||
if not splits(0):
|
||
return False
|
||
# ...and it must really be more than one part, not one inflected word.
|
||
return not _known_form(w, checker, "de") or any(
|
||
_known_form(w[:end], checker, "de") and end < n - _MIN_PART + 1
|
||
and any(w.startswith(link, end) and end + len(link) < n for link in _LINKING)
|
||
for end in range(_MIN_PART, n - _MIN_PART + 1))
|
||
|
||
|
||
_GLUED_CONJUNCTION_RE = re.compile(r"^(?P<stem>[A-Za-zÄÖÜäöüß]{4,}?)(?P<conj>und|oder)$", re.I)
|
||
|
||
|
||
def _suspended_compound_suggestion(token: str, checker) -> str | None:
|
||
"""``Programmierund`` -> ``Programmier- und``: a suspended compound whose hyphen was lost."""
|
||
match = _GLUED_CONJUNCTION_RE.match(token)
|
||
if not match:
|
||
return None
|
||
stem = match.group("stem").casefold()
|
||
if any(stem + tail in checker for tail in ("", "en", "ung", "e", "er", "s")) or is_known_compound(stem, checker):
|
||
return f"{match.group('stem')}- {match.group('conj').lower()}"
|
||
return None
|
||
|
||
|
||
def _suggestions(token: str, checker, language: str) -> list[str]:
|
||
low = token.casefold()
|
||
found: list[str] = []
|
||
if language == "de":
|
||
glued = _suspended_compound_suggestion(token, checker)
|
||
if glued:
|
||
return [glued]
|
||
if language == "de":
|
||
found.extend(v for v in sorted(_umlaut_variants(low)) if v in checker)
|
||
try:
|
||
ranked = checker.candidates(low) or set()
|
||
except Exception:
|
||
ranked = set()
|
||
freq = getattr(checker, "word_usage_frequency", None)
|
||
ordered = sorted(ranked, key=lambda w: -(freq(w) if callable(freq) else 0))
|
||
for cand in ordered:
|
||
if cand != low and cand not in found:
|
||
found.append(cand)
|
||
return [_match_case(token, c) for c in found[:_MAX_SUGGESTIONS]]
|
||
|
||
|
||
_OPENING_QUOTES = "»«\"„“”'‘’("
|
||
|
||
|
||
def _starts_sentence(text: str, index: int) -> bool:
|
||
before = text[:index].rstrip(_OPENING_QUOTES + " \n")
|
||
return not before or before[-1] in ".!?…:"
|
||
|
||
|
||
_SEAM_RE = re.compile(r"(?<=[A-Za-zÄÖÜäöüß])(?:-\s+|\u00ad)(?=[a-zäöüß])")
|
||
_SUSPENDED = {"und", "oder", "beziehungsweise", "sowie"}
|
||
|
||
|
||
def _close_line_breaks(text: str) -> tuple[str, list[int]]:
|
||
"""Text with printed line-break hyphens closed up, and each kept character's raw offset.
|
||
|
||
``Prüfungs- formate`` is one word broken at a line end; narration already
|
||
closes it, so it is checked (and, if wrong, fixed) as one word. ``Hardware-
|
||
und Software`` is a suspended compound and stays as printed.
|
||
"""
|
||
out: list[str] = []
|
||
origin: list[int] = []
|
||
pos = 0
|
||
for match in _SEAM_RE.finditer(text):
|
||
following = re.match(r"[A-Za-zÄÖÜäöüß]+", text[match.end():])
|
||
if following and following.group(0).casefold() in _SUSPENDED:
|
||
continue
|
||
for i in range(pos, match.start()):
|
||
out.append(text[i]); origin.append(i)
|
||
pos = match.end()
|
||
for i in range(pos, len(text)):
|
||
out.append(text[i]); origin.append(i)
|
||
origin.append(len(text))
|
||
return "".join(out), origin
|
||
|
||
|
||
def find_spelling_issues(text: str, language: str = "auto",
|
||
ignore: Iterable[str] = (), other_language: bool = True,
|
||
vocabulary: Iterable[str] = ()) -> list[dict]:
|
||
"""Dictionary-unknown words in `text`, with position and suggestions.
|
||
|
||
A word is only a finding when it is unknown in the text's language, is not
|
||
a known compound, is not known in the *other* language either (an English
|
||
term in a German article is not a typo), and is not in the reviewer's own
|
||
dictionary. Acronyms, short words and, outside German, capitalised words
|
||
(names) are skipped.
|
||
"""
|
||
lang = detect_text_language(text) if language == "auto" else language
|
||
checker = _checker(lang)
|
||
if checker is None:
|
||
return []
|
||
other = _checker("en" if lang == "de" else "de") if other_language else None
|
||
ignored = {str(w).casefold() for w in ignore} | known_names()
|
||
vocab = {str(w).casefold() for w in vocabulary}
|
||
issues: list[dict] = []
|
||
raw = text or ""
|
||
text, origin = _close_line_breaks(raw)
|
||
for match in _TOKEN_RE.finditer(text):
|
||
whole = match.group(0)
|
||
offset = match.start()
|
||
parts = whole.split("-")
|
||
for index, part in enumerate(parts):
|
||
start = offset
|
||
offset += len(part) + 1
|
||
# A bare stem before a hyphen (``Lern-Sparringspartner``, ``Programmier- und``) is
|
||
# not a word on its own; it is fine when a normal form of it is.
|
||
before_hyphen = index < len(parts) - 1 or text[start + len(part):start + len(part) + 1] == "-"
|
||
if before_hyphen and any(_known_form(part + tail, checker, lang) for tail in _STEM_ENDINGS):
|
||
continue
|
||
if len(part) < _MIN_LENGTH or part.isupper() or part.casefold() in ignored:
|
||
continue
|
||
if lang != "de" and part[:1].isupper() and not _starts_sentence(text, match.start()):
|
||
continue # mid-sentence capital: a name or product
|
||
if part.casefold() in vocab and not (
|
||
lang == "de" and any(v in checker for v in _umlaut_variants(part.casefold()))):
|
||
continue # printed repeatedly: a name or term, not OCR noise
|
||
if _known_form(part, checker, lang) or (other is not None and _known_form(part, other, "en" if lang == "de" else "de")):
|
||
continue
|
||
if lang == "de" and is_known_compound(part, checker):
|
||
continue
|
||
# camelCase / mixed-case identifiers (PyMuPDF, iPhone) are brands, not typos
|
||
if part != part.capitalize() and part != part.lower() and not part.isupper():
|
||
continue
|
||
raw_start, raw_end = origin[start], origin[start + len(part) - 1] + 1
|
||
issues.append({
|
||
# token: the word as read (line-break hyphens closed); raw: the exact printed
|
||
# slice at start:end, which is what an edit must still find in the text.
|
||
"token": part, "raw": raw[raw_start:raw_end], "start": raw_start, "end": raw_end, "language": lang,
|
||
"suggestions": _suggestions(part, checker, lang),
|
||
})
|
||
if len(issues) >= _MAX_ISSUES_PER_TEXT:
|
||
break
|
||
# Two unknown capitalised words side by side (``Juliane Jarke``) are a person's name.
|
||
starts = {i["start"] for i in issues if i["token"][:1].isupper()}
|
||
ends = {i["end"] + 1 for i in issues if i["token"][:1].isupper()}
|
||
return [i for i in issues
|
||
if not (i["token"][:1].isupper() and (i["start"] in ends or i["end"] + 1 in starts))]
|
||
|
||
|
||
def block_issues(blocks: list[dict], article_language: str | None = None,
|
||
ignore: Iterable[str] = (), vocabulary: Iterable[str] = ()) -> list[dict]:
|
||
"""Findings for a run of blocks, each with its block id, page and context."""
|
||
ignore = set(ignore)
|
||
language = article_language or detect_article_language(b.get("text", "") for b in blocks)
|
||
out: list[dict] = []
|
||
for block in blocks:
|
||
text = block.get("text") or ""
|
||
lang = language_for_block(text, language)
|
||
for issue in find_spelling_issues(text, lang, ignore, vocabulary=vocabulary):
|
||
lo = max(0, issue["start"] - 60)
|
||
hi = min(len(text), issue["end"] + 60)
|
||
out.append(dict(issue, block_id=block["id"], page=block.get("page"),
|
||
context=text[lo:hi].replace("\n", " ")))
|
||
# A misspelling that recurs identically is far more likely a name, brand or
|
||
# technical term than OCR noise (which differs every time): treat it as vocabulary.
|
||
counts: dict[str, int] = {}
|
||
for issue in out:
|
||
counts[issue["token"].casefold()] = counts.get(issue["token"].casefold(), 0) + 1
|
||
return [i for i in out if counts[i["token"].casefold()] < _RECURRING]
|
||
|
||
|
||
def _edit_distance(a: str, b: str, limit: int = 3) -> int:
|
||
if abs(len(a) - len(b)) > limit:
|
||
return limit + 1
|
||
prev = list(range(len(b) + 1))
|
||
for i, ca in enumerate(a, 1):
|
||
cur = [i]
|
||
for j, cb in enumerate(b, 1):
|
||
cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
|
||
prev = cur
|
||
return prev[-1]
|
||
|
||
|
||
def accept_agent_fix(issue: dict, replacement: str) -> bool:
|
||
"""Guard for an agent-proposed correction: never trust a model's rewrite blindly.
|
||
|
||
The replacement must be one of the dictionary suggestions, or a single
|
||
dictionary-known word (or known compound) within two edits of the printed
|
||
token. Anything else - a different word, a phrase, new content - is refused
|
||
and the finding stays marked for a human.
|
||
"""
|
||
replacement = (replacement or "").strip()
|
||
token = issue["token"]
|
||
if (not replacement or replacement == token
|
||
or not re.fullmatch(r"[A-Za-zÄÖÜäöüß]+(?:- (?:und|oder))?", replacement)):
|
||
return False
|
||
if replacement.casefold() in {s.casefold() for s in issue.get("suggestions") or []}:
|
||
return True
|
||
checker = _checker(issue.get("language") or "de")
|
||
if checker is None:
|
||
return False
|
||
if _edit_distance(token.casefold(), replacement.casefold(), 2) > 2:
|
||
return False
|
||
return _known(replacement, checker) or (
|
||
issue.get("language") == "de" and is_known_compound(replacement, checker))
|
||
|
||
|
||
def apply_fix(text: str, start: int, end: int, token: str, replacement: str) -> str | None:
|
||
"""Replace text[start:end] when it still is `token`; None when the text has moved on."""
|
||
if text[start:end] != token:
|
||
return None
|
||
return text[:start] + replacement + text[end:]
|