File size: 2,374 Bytes
024c30a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """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
|