anya/structs/headline.py

import logging
from urllib.parse import urlparse

from services.similarity import cosine_lists

logger = logging.getLogger(__name__)


class Headline:
    display_text: str
    normalized_text: list[str]
    source_url: str
    domain: str
    published_at: object

    def __init__(self, display_text: str, normalized_text: list[str], source_url: str = "", published_at=None):
        self.display_text = display_text
        self.normalized_text = normalized_text
        self.source_url = source_url
        self.domain = self._extract_domain(source_url)
        self.published_at = published_at
        logger.debug(
            "Initialized Headline instance (display_text=%r, token_count=%d, source=%r, domain=%r, published_at=%r)",
            self.display_text, len(self.normalized_text), self.source_url, self.domain, self.published_at,
        )

    @staticmethod
    def _extract_domain(url: str):
        """Return a normalized outlet domain for a source URL, or None if absent.

        Strips scheme, whitespace and a leading ``www.``/feed/redirect subdomain
        (``www.``, ``feeds.``, ``rss.``, ``moxie.``) so that ``https://www.cnn.com/us``
        and ``https://feeds.npr.org/1001/rss.xml`` both resolve to their outlet
        (``cnn.com``, ``npr.org``) — one outlet counts once regardless of section
        URL or feed host.
        """
        if not url or not isinstance(url, str) or not url.strip():
            return None
        host = urlparse(url.strip()).netloc.lower()
        if not host:
            return None
        for prefix in ("www.", "feeds.", "rss.", "moxie."):
            if host.startswith(prefix):
                host = host[len(prefix):]
                break
        return host

    def compare_headlines(self, other_headline):
        if not isinstance(other_headline, Headline):
            logger.warning("Comparing Headline %r with incompatible object of type %s: %r",
                           self.display_text, type(other_headline).__name__, other_headline)
        logger.debug("Comparing Headline %r against %r",
                     self.display_text, getattr(other_headline, 'display_text', repr(other_headline)))

        try:
            score = cosine_lists(self.normalized_text, other_headline.normalized_text)
            logger.debug("Headline comparison score: %.4f between %r and %r",
                         score, self.display_text, getattr(other_headline, 'display_text', repr(other_headline)))
            return score
        except Exception as e:
            logger.error("Error comparing headlines (%r vs %r): %s",
                         self.display_text, getattr(other_headline, 'display_text', repr(other_headline)), e, exc_info=True)
            raise