anya/services/ssr.py

import html
import json
import logging
import re

from services.dates import parse_datetime

logger = logging.getLogger(__name__)

# JSON-LD structured data blocks (schema.org) that most news sites emit for SEO —
# and Next.js's serialized page state. In both cases the headlines (and dates)
# are already present in the HTML, so no JavaScript execution is required.
_JSON_LD_RE = re.compile(
    r"""<script\b[^>]*?\btype=(["'])application/ld\+json\1[^>]*>(.*?)</script>""",
    re.IGNORECASE | re.DOTALL,
)
_NEXT_DATA_RE = re.compile(
    r"""<script\b[^>]*?\bid=(["'])__NEXT_DATA__\1[^>]*>(.*?)</script>""",
    re.IGNORECASE | re.DOTALL,
)

_ARTICLE_TYPES = {
    'newsarticle', 'article', 'report', 'analysisnewsarticle',
    'opinionnewsarticle', 'reviewnewsarticle', 'blogposting',
    'liveblogposting', 'backgroundnewsarticle', 'reportagenewsarticle',
}

_URL_KEYS = ('url', 'href', 'link', 'canonicalUrl', 'slug', 'uri')


def _has_url(obj):
    for key in _URL_KEYS:
        value = obj.get(key)
        if isinstance(value, str) and value.strip():
            return True
    return False


def _schema_type(obj):
    """Last path segment of an object's ``@type``, lowercased (handles URLs)."""
    if not isinstance(obj, dict):
        return ''
    t = obj.get('@type')
    if isinstance(t, list):
        t = t[0] if t else None
    if not isinstance(t, str):
        return ''
    return t.strip().rstrip('/').rsplit('/', 1)[-1].lower()


def _iter_json_scripts(content):
    """Yield unescaped JSON bodies of JSON-LD and Next.js SSR ``<script>`` blocks."""
    if not content or not isinstance(content, str):
        return
    for pattern in (_JSON_LD_RE, _NEXT_DATA_RE):
        for match in pattern.finditer(content):
            body = match.group(2)
            if body and body.strip():
                yield html.unescape(body.strip())


def _collect(obj, out):
    """Recursively append ``(headline, published_at)`` for article-like objects.

    Headlines are recognized by three signals:
      * ``headline`` — article-specific in schema.org, trusted directly.
      * ``title`` + a url-ish sibling — the common Next.js SSR shape, where
        ``title`` alone is too ambiguous (section labels use it too).
      * ``name`` — trusted only with an article-type/date/author signal.
    """
    if isinstance(obj, dict):
        stype = _schema_type(obj)
        headline = None

        raw = obj.get('headline')
        if isinstance(raw, str) and raw.strip():
            headline = raw.strip()
        else:
            raw_title = obj.get('title')
            if isinstance(raw_title, str) and raw_title.strip() and _has_url(obj):
                headline = raw_title.strip()
            else:
                raw_name = obj.get('name')
                if isinstance(raw_name, str) and raw_name.strip() and (
                    stype in _ARTICLE_TYPES or obj.get('datePublished') or obj.get('author')
                ):
                    headline = raw_name.strip()

        if headline:
            published = obj.get('datePublished') or obj.get('dateModified') or obj.get('date')
            published_at = parse_datetime(published) if isinstance(published, str) else None
            out.append((headline, published_at))

        for value in obj.values():
            _collect(value, out)
    elif isinstance(obj, list):
        for value in obj:
            _collect(value, out)


def extract_headlines(content):
    """Best-effort extraction of ``(headline, published_at)`` from embedded JSON.

    Walks JSON-LD and Next.js SSR data for article-like objects and returns a
    de-duplicated list of tuples, where ``published_at`` is a ``datetime`` or
    ``None``.
    """
    results = []
    seen = set()
    for body in _iter_json_scripts(content):
        try:
            data = json.loads(body)
        except (json.JSONDecodeError, ValueError, TypeError):
            continue
        collected = []
        _collect(data, collected)
        for headline, published in collected:
            key = headline.casefold()
            if key in seen:
                continue
            seen.add(key)
            results.append((headline, published))

    logger.info("Extracted %d headline(s) from embedded JSON", len(results))
    return results