anya/services/normalization.py

import logging
import re
import string
import tomllib

logger = logging.getLogger(__name__)


def get_aliases(path):
    """Load headline aliases from a TOML file with an ``[aliases]`` table.

    Each ``"full phrase" = "canonical"`` entry maps a multi-word name to a
    canonical short form (e.g. ``"federal reserve" = "fed"``). Keys are
    lowercased, punctuation-stripped, and whitespace-collapsed to match the
    text form that ``normalize_headline`` applies aliases to.
    """
    logger.info("Attempting to load aliases from file: '%s'", path)
    try:
        with open(path, 'rb') as aliases_file:
            data = tomllib.load(aliases_file)
    except FileNotFoundError:
        logger.error("Aliases file not found at path: '%s'", path, exc_info=True)
        raise
    except PermissionError:
        logger.error("Permission denied accessing aliases file: '%s'", path, exc_info=True)
        raise
    except tomllib.TOMLDecodeError as e:
        logger.error("Failed to parse aliases file '%s': %s", path, e, exc_info=True)
        raise
    except Exception as e:
        logger.error("Failed to load aliases from '%s': %s", path, e, exc_info=True)
        raise

    aliases = {}
    for phrase, canonical in data.get('aliases', {}).items():
        # Mirror normalize_headline's pre-alias text form: lowercase, then
        # strip punctuation (not space-replace), then collapse whitespace.
        phrase = str(phrase).strip().lower()
        for char in string.punctuation:
            phrase = phrase.replace(char, '')
        phrase = ' '.join(phrase.split())
        canonical = str(canonical).strip().lower()
        if phrase and canonical:
            aliases[phrase] = canonical
    logger.info("Successfully loaded %d aliases from '%s'", len(aliases), path)
    return aliases


def get_stopwords(path):
    logger.info("Attempting to load stopwords from file: '%s'", path)
    try:
        with open(path, 'r', encoding='utf-8-sig') as stopwords_file:
            logger.debug("Opened stopwords file '%s'", path)
            lines = stopwords_file.read().splitlines()
            stopwords = set(lines)
            logger.info("Successfully loaded %d unique stopwords from '%s' (%d total lines)", len(stopwords), path, len(lines))
            return stopwords
    except FileNotFoundError:
        logger.error("Stopwords file not found at path: '%s'", path, exc_info=True)
        raise
    except PermissionError:
        logger.error("Permission denied accessing stopwords file: '%s'", path, exc_info=True)
        raise
    except Exception as e:
        logger.error("Failed to load stopwords from '%s': %s", path, e, exc_info=True)
        raise


def get_excluded_phrases(path):
    """Load boilerplate link phrases from a file (one per line, '#' comments).

    These are strings that appear in navigation/footer links ("skip to content",
    "your privacy choices", "terms of service") and should never be treated as
    headlines. Returns a list of stripped, non-empty phrases in file order.
    """
    logger.info("Attempting to load excluded phrases from file: '%s'", path)
    try:
        with open(path, 'r', encoding='utf-8-sig') as phrases_file:
            lines = phrases_file.read().splitlines()
    except FileNotFoundError:
        logger.error("Excluded phrases file not found at path: '%s'", path, exc_info=True)
        raise
    except PermissionError:
        logger.error("Permission denied accessing excluded phrases file: '%s'", path, exc_info=True)
        raise
    except Exception as e:
        logger.error("Failed to load excluded phrases from '%s': %s", path, e, exc_info=True)
        raise

    phrases = []
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith('#'):
            continue
        phrases.append(stripped)
    logger.info("Successfully loaded %d excluded phrases from '%s'", len(phrases), path)
    return phrases


def canonicalize(text):
    """Lowercase, turn punctuation into spaces, and collapse whitespace.

    Produces a comparable form of free text for phrase matching — e.g.
    ``"Terms-of-Service."`` becomes ``"terms of service"``.
    """
    text = (text or '').lower()
    for char in string.punctuation:
        text = text.replace(char, ' ')
    return ' '.join(text.split())


def is_excluded(text, excluded_phrases):
    """Return True if ``text`` contains an excluded phrase as a contiguous
    token subsequence (case-insensitive, punctuation-insensitive).

    Matching is sub-phrase aware, so the entry ``privacy choices`` also matches
    the link text "Your Privacy Choices" without needing a separate exact row.
    """
    if not excluded_phrases:
        return False
    tokens = canonicalize(text).split()
    if not tokens:
        return False
    for phrase in excluded_phrases:
        phrase_tokens = canonicalize(phrase).split()
        if not phrase_tokens:
            continue
        for start in range(len(tokens) - len(phrase_tokens) + 1):
            if tokens[start:start + len(phrase_tokens)] == phrase_tokens:
                logger.debug("Text excluded by phrase %r: %r", phrase, text)
                return True
    return False


def remove_stopwords(text, stopwords):
    logger.debug("Removing stopwords from text (%d chars): %r (available stopwords: %d)", len(text), text, len(stopwords))
    # split() without arguments handles all whitespace (spaces, tabs, newlines)
    words = text.split()
    sentence_words = []

    for word in words:
        # Strip surrounding punctuation and lowercase for comparison
        cleaned_word = word.strip(string.punctuation).lower()
        if cleaned_word and cleaned_word not in stopwords:
            sentence_words.append(word)
        elif cleaned_word in stopwords:
            logger.debug("Word filtered out as stopword: %r (original: %r)", cleaned_word, word)
        else:
            logger.debug("Word dropped (empty after stripping punctuation): %r", word)

    logger.debug("Filtered words: input=%d words, output=%d words -> %s", len(words), len(sentence_words), sentence_words)
    return sentence_words


def replace_aliases(text, aliases):
    """Swap full phrases for canonical forms, longest first, whole-word only.

    Longer phrases are applied before shorter ones so ``president of the united
    states`` becomes ``potus`` rather than leaving ``united states`` for the
    ``us`` rule to snag. ``\b`` boundaries prevent substring matches such as
    collapsing ``united statesman`` to ``usman``.
    """
    if not aliases:
        return text
    text = ' '.join(text.split())
    for phrase in sorted(aliases, key=len, reverse=True):
        pattern = re.compile(r'\b' + re.escape(phrase) + r'\b')
        text = pattern.sub(aliases[phrase], text)
    return text


def normalize_headline(text, stopwords, aliases=None):
    logger.debug("Starting normalization of headline: %r", text)
    headline = text.strip().lower()
    punctuation = string.punctuation
    for char in punctuation:
        headline = headline.replace(char, '')
    logger.debug("Headline after lowercasing and punctuation removal: %r", headline)
    headline = replace_aliases(headline, aliases)
    logger.debug("Headline after replacing aliases: %r", headline)
    normalized = remove_stopwords(headline, stopwords)
    logger.debug("Completed normalization for %r -> %s", text, normalized)
    return normalized