"""Surface normalisation and the abbreviation index used by clustering. **This normalisation is for clustering only.** Span validation normalises whitespace and nothing else — every additional normalisation there is a hole a fabrication can fit through. Do not reuse `normalize()` in that path. """ from __future__ import annotations import re import unicodedata from ..models import AbbrevPair # Surfaces that carry no discriminating power on their own. A mention of just # "unit" or "parameter" is not a term. These are dropped as WHOLE surface forms # only, never as substrings — so no term containing them is ever lost. STOP_SURFACES = { "unit", "type", "class", "equipment", "equipment unit", "parameter", "activity", "data", "nilai", "proses", "hasil", "total", } def normalize(surface: str) -> str: s = unicodedata.normalize("NFKC", surface).casefold() s = s.replace("-", " ").replace("_", " ") s = re.sub(r"[.’']", "", s) s = re.sub(r"[^\w\s/()]", " ", s) s = re.sub(r"\s+", " ", s) return s.strip(" ()/") def is_noise(surface: str) -> bool: n = normalize(surface) if len(n) < 2: return True if n in STOP_SURFACES: return True return not re.search(r"[a-z]", n) # pure numbers / symbols class AbbrevIndex: """Bidirectional abbreviation ↔ expansion lookup built from legend blocks. This is why the legend filter runs before clustering: without it, `PA` and `Physical Availability` never meet. """ def __init__(self, pairs: list[AbbrevPair]): self.to_expansion: dict[str, str] = {} self.to_abbrev: dict[str, str] = {} for pair in pairs: abbrev, expansion = normalize(pair.abbrev), normalize(pair.expansion) if not abbrev or not expansion: continue self.to_expansion[abbrev] = expansion self.to_abbrev[expansion] = abbrev def canonical_key(self, surface: str) -> str: """Map a surface to a shared key so an abbreviation and its expansion collide into the same bucket.""" n = normalize(surface) return self.to_abbrev.get(n, n) def linked(self, a: str, b: str) -> bool: na, nb = normalize(a), normalize(b) return self.to_expansion.get(na) == nb or self.to_expansion.get(nb) == na