anya/services/normalization.py

import logging
import string

logger = logging.getLogger(__name__)

ALIASES = {
    'federal reserve': 'fed',
    'united states': 'us',
    'united kingdom': 'uk',
    'european union': 'eu',
    'north atlantic treaty organization': 'nato',
    'white house': 'wh',
    'united nations': 'un',
    'international monetary fund': 'imf',
    'world trade organization': 'wto',
    'central intelligence agency': 'cia',
    'federal bureau of investigation': 'fbi',
    'department of justice': 'doj',
    'department of defense': 'dod',
    'environmental protection agency': 'epa',
    'securities and exchange commission': 'sec',
    'internal revenue service': 'irs',
    'social security administration': 'ssa',
    'centers for disease control': 'cdc',
    'national security agency': 'nsa',
    'department of homeland security': 'dhs',
    'supreme court': 'scotus',
    'republican party': 'gop',
    'democratic party': 'democrats',
    'president of the united states': 'potus',
    'vice president': 'vp',
    'prime minister': 'pm',
    'secretary of state': 'secstate',
    'attorney general': 'ag',
    'gross domestic product': 'gdp',
    'consumer price index': 'cpi',
    'unemployment rate': 'unemployment',
    'inflation rate': 'inflation',
    'national debt': 'debt',
    'budget deficit': 'deficit',
    'middle east': 'mideast',
    'asia pacific': 'apac',
    'latin america': 'latam'
}


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 normalize_headline(text, stopwords):
    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_common_aliases(headline)
    logger.debug("Headline after replacing common aliases: %r", headline)
    normalized = remove_stopwords(headline, stopwords)
    logger.debug("Completed normalization for %r -> %s", text, normalized)
    return normalized


def replace_common_aliases(headline):
    for alias in ALIASES:
        headline = headline.replace(alias, ALIASES[alias])

    return headline