src/knowledge_extraction/cluster/normalize.py

"""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