anya/services/feeds.py

import logging
import xml.etree.ElementTree as ET

from services.dates import parse_datetime

logger = logging.getLogger(__name__)

# Distinctive root tags used to sniff feed (XML) content apart from HTML.
FEED_MARKERS = ('<rss', '<feed', '<rdf:rdf')


def looks_like_feed(content):
    """Best-effort sniff for RSS/Atom XML content (vs. HTML pages)."""
    if not content or not isinstance(content, str):
        return False
    head = content[:4000].lower()
    return any(marker in head for marker in FEED_MARKERS)


def _local(tag):
    """Return an element's local name, ignoring any XML namespace prefix."""
    return tag.rsplit('}', 1)[-1] if '}' in tag else tag


def _child_text(elem, names):
    """First direct child whose local name is in ``names``, with text content."""
    for child in elem:
        if _local(child.tag) in names:
            text = (child.text or '').strip()
            if text:
                return text
    return None


def parse_feed(content):
    """Extract ``(title, published_at)`` tuples from an RSS or Atom feed.

    Understands RSS 2.0 (``<item>``), RSS 1.0 (``<rdf:RDF>`` items) and Atom
    (``<entry>``). Publication timestamps are taken from ``pubDate``,
    ``published``, ``updated`` or ``dc:date`` (``None`` when absent or
    unparseable).
    """
    items = []
    try:
        root = ET.fromstring(content)
    except ET.ParseError as e:
        logger.error("Failed to parse feed XML: %s", e)
        return items
    except Exception as e:
        logger.error("Unexpected error parsing feed: %s", e, exc_info=True)
        return items

    # Work over every <item>/<entry> element regardless of nesting, which covers
    # RSS 2.0, RSS 1.0 (RDF) and Atom in one pass.
    entries = [el for el in root.iter() if _local(el.tag) in ('item', 'entry')]

    for entry in entries:
        title = _child_text(entry, ('title',))
        if not title:
            continue
        published = _child_text(entry, ('pubDate', 'published', 'updated', 'date'))
        items.append((title, parse_datetime(published)))

    logger.info("Parsed %d entries from feed", len(items))
    return items