anya/services/headlines.py

import logging
import re

import requests

from services.dates import parse_datetime
from services.feeds import looks_like_feed, parse_feed
from services.normalization import is_excluded, normalize_headline
from services.ssr import extract_headlines
from structs.headline import Headline

logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 10
MIN_HEADLINE_WORDS = 3

# Common browser-like request headers. News sites frequently reject requests
# that look like minimal bots, so these make Anya look like a regular browser.
# Accept-Encoding is intentionally omitted so requests/urllib3 negotiates and
# decompresses a response it can actually handle (avoids brotli-only responses
# arriving as undecodable bytes).
DEFAULT_HEADERS = {
    'User-Agent': (
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
        'AppleWebKit/537.36 (KHTML, like Gecko) '
        'Chrome/124.0.0.0 Safari/537.36'
    ),
    'Accept': (
        'text/html,application/xhtml+xml,application/xml;q=0.9,'
        'image/avif,image/webp,image/apng,*/*;q=0.8'
    ),
    'Accept-Language': 'en-US,en;q=0.9',
    'Cache-Control': 'max-age=0',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1',
    'sec-ch-ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
}

# Single streaming pass over the HTML: captures <time> markers (with an optional
# datetime/title attribute or inner text) and <a>/<span> headline text, so every
# headline can be associated with the most recent publication timestamp seen
# before it in document order.
_CANDIDATE_RE = re.compile(
    r"<time\b([^>]*)>(.*?)</time>"                                  # group 1 attrs, group 2 text
    r"|(?:(?:datetime|dateTime)\s*=\s*[\"']([^\"']+)[\"'])"          # group 3 datetime attr
    r"|<(?:article|li)\b[^>]*>"                                    # container boundary (no group)
    r"|<(?:a|span)\b[^>]*>\s*([^<]*?)\s*</(?:a|span)>",             # group 4 headline text
    re.IGNORECASE | re.DOTALL,
)


def _extract_candidates(source_content):
    """Yield ``(text, published_at)`` tuples in document order.

    Each headline is associated with the most recent publication timestamp seen
    before it. Timestamps reset at each ``<article>``/``<li>`` boundary so a
    headline with no date of its own does not inherit another story's timestamp.
    """
    candidates = []
    last_published = None
    for match in _CANDIDATE_RE.finditer(source_content):
        if match.group(1) is not None:
            # A <time ...>...</time> element: prefer an explicit datetime/title
            # attribute, otherwise fall back to its inner text.
            attrs = match.group(1)
            attr_match = re.search(r'(?:datetime|dateTime)\s*=\s*[\"\']([^\"\']+)[\"\']', attrs, re.IGNORECASE) \
                or re.search(r'title\s*=\s*[\"\']([^\"\']+)[\"\']', attrs, re.IGNORECASE)
            raw = attr_match.group(1) if attr_match else match.group(2)
            parsed = parse_datetime(raw)
            if parsed is not None:
                last_published = parsed
        elif match.group(3) is not None:
            # A datetime attribute on some non-<time> element.
            parsed = parse_datetime(match.group(3))
            if parsed is not None:
                last_published = parsed
        elif match.group(4) is not None:
            candidates.append((match.group(4), last_published))
        else:
            # <article>/<li> boundary: a headline without its own timestamp
            # should not inherit the previous article's date.
            last_published = None
    return candidates


def _build_headlines(items, source_url, stopwords, excluded_phrases, aliases=None):
    """Turn ``(title, published_at)`` tuples into headline objects, applying filters."""
    collected = []
    for title, published_at in items:
        cleaned_text = title.strip()
        if not cleaned_text or not is_headline(cleaned_text, stopwords, excluded_phrases, aliases):
            continue
        try:
            normalized = normalize_headline(cleaned_text, stopwords, aliases)
            collected.append(Headline(cleaned_text, normalized, source_url, published_at))
        except Exception as e:
            logger.error("Failed to normalize title %r from '%s': %s", cleaned_text, source_url, e, exc_info=True)
    return collected


def is_headline(text, stopwords=None, excluded_phrases=None, aliases=None):
    if not text or not isinstance(text, str):
        return False
    cleaned_text = text.strip()
    if not cleaned_text:
        return False
    words = cleaned_text.split()
    if len(words) < MIN_HEADLINE_WORDS:
        logger.debug("Text rejected as headline (fewer than %d words): %r", MIN_HEADLINE_WORDS, cleaned_text)
        return False
    if not any(c.isalnum() for c in cleaned_text):
        logger.debug("Text rejected as headline (no alphanumeric characters): %r", cleaned_text)
        return False
    if excluded_phrases and is_excluded(cleaned_text, excluded_phrases):
        logger.debug("Text rejected as non-headline (matches excluded phrase): %r", cleaned_text)
        return False
    if stopwords is not None:
        normalized = normalize_headline(cleaned_text, stopwords, aliases)
        if not normalized:
            logger.debug("Text rejected as headline (no meaningful tokens after stopword removal): %r", cleaned_text)
            return False
    return True


def prepare_headlines(sources, stopwords, timeout=DEFAULT_TIMEOUT, headers=None, excluded_phrases=None, aliases=None):
    request_headers = headers if headers is not None else DEFAULT_HEADERS
    logger.info("Starting preparation of headlines for %d sources", len(sources) if sources else 0)
    headlines = []
    if not sources:
        logger.warning("No sources provided to prepare_headlines.")
        return headlines

    for idx, source in enumerate(sources, start=1):
        if not source or not source.strip():
            logger.warning("Skipping empty source at index %d", idx)
            continue

        source_url = source.strip()
        logger.info("Fetching source [%d/%d]: '%s'", idx, len(sources), source_url)
        try:
            response = requests.get(source_url, allow_redirects=True, timeout=timeout, headers=request_headers)
            logger.debug("Received HTTP response %d for '%s' (content length: %d bytes)",
                         response.status_code, source_url, len(response.content))
            if response.status_code != 200:
                logger.warning("Source '%s' returned non-200 status code: %d", source_url, response.status_code)
            source_content = response.text
        except requests.exceptions.Timeout as e:
            logger.error("Request timed out for source '%s': %s", source_url, e, exc_info=True)
            continue
        except requests.exceptions.RequestException as e:
            logger.error("HTTP request failed for source '%s': %s", source_url, e, exc_info=True)
            continue
        except Exception as e:
            logger.error("Unexpected error fetching source '%s': %s", source_url, e, exc_info=True)
            continue

        if looks_like_feed(source_content):
            logger.info("Detected RSS/Atom feed for source '%s'", source_url)
            try:
                feed_headlines = _build_headlines(parse_feed(source_content), source_url, stopwords, excluded_phrases, aliases)
            except Exception as e:
                logger.error("Failed to parse feed from '%s': %s", source_url, e, exc_info=True)
                continue
            headlines.extend(feed_headlines)
            logger.info("Successfully extracted %d headlines from source '%s'", len(feed_headlines), source_url)
            continue

        try:
            ssr_items = extract_headlines(source_content)
        except Exception as e:
            logger.error("Failed to extract embedded JSON from '%s': %s", source_url, e, exc_info=True)
            ssr_items = []
        if ssr_items:
            ssr_headlines = _build_headlines(ssr_items, source_url, stopwords, excluded_phrases, aliases)
            headlines.extend(ssr_headlines)
            logger.info("Successfully extracted %d headlines via embedded JSON from source '%s'", len(ssr_headlines), source_url)
            continue

        logger.debug("Parsing HTML content from '%s' for headline candidates", source_url)
        try:
            candidates = _extract_candidates(source_content)
            logger.info("Found %d candidate tags in source '%s'", len(candidates), source_url)
        except Exception as e:
            logger.error("Regex extraction failed on content from '%s': %s", source_url, e, exc_info=True)
            continue

        source_headlines_count = 0
        for tag_idx, (link_text, published_at) in enumerate(candidates, start=1):
            cleaned_text = link_text.strip()
            if not cleaned_text:
                logger.debug("Skipping empty tag text at position %d from '%s'", tag_idx, source_url)
                continue
            if not is_headline(cleaned_text, stopwords, excluded_phrases, aliases):
                logger.debug("Skipping non-headline tag text at position %d from '%s': %r", tag_idx, source_url, cleaned_text)
                continue
            logger.debug("Processing tag [%d/%d] from '%s': %r", tag_idx, len(candidates), source_url, cleaned_text)
            try:
                normalized_headline = normalize_headline(cleaned_text, stopwords, aliases)
                headline = Headline(cleaned_text, normalized_headline, source_url, published_at)
                headlines.append(headline)
                source_headlines_count += 1
            except Exception as e:
                logger.error("Failed to normalize/create headline for text %r from '%s': %s", cleaned_text, source_url, e, exc_info=True)

        logger.info("Successfully extracted %d headlines from source '%s'", source_headlines_count, source_url)

    logger.info("Finished preparing headlines. Total headlines collected across all sources: %d", len(headlines))
    return headlines