anya/services/normalization.py

import logging
import string

logger = logging.getLogger(__name__)


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)
    normalized = remove_stopwords(headline, stopwords)
    logger.debug("Completed normalization for %r -> %s", text, normalized)
    return normalized