nas-tools/info-engine/collector.py

import logging
import os
from datetime import datetime
from email.utils import parsedate_to_datetime
from urllib.parse import urlparse
import xml.etree.ElementTree as ET

import httpx

log = logging.getLogger(__name__)


def _source_name(url: str) -> str:
    host = urlparse(url).netloc
    return host or url


def _first_text(parent, paths: list[str]) -> str:
    for path in paths:
        node = parent.find(path)
        if node is not None and node.text:
            value = node.text.strip()
            if value:
                return value
    return ""


def _atom_link(entry) -> str:
    for link in entry.findall("{http://www.w3.org/2005/Atom}link"):
        href = (link.attrib.get("href") or "").strip()
        rel = (link.attrib.get("rel") or "alternate").strip()
        if href and rel == "alternate":
            return href
    links = entry.findall("{http://www.w3.org/2005/Atom}link")
    if links:
        return (links[0].attrib.get("href") or "").strip()
    return ""


def _parse_datetime(value: str | None) -> str | None:
    if not value:
        return None
    value = value.strip()
    if not value:
        return None

    try:
        return parsedate_to_datetime(value).isoformat()
    except Exception:
        pass

    try:
        iso = value.replace("Z", "+00:00")
        return datetime.fromisoformat(iso).isoformat()
    except Exception:
        return None


def _parse_feed(feed_url: str, xml_text: str) -> list[dict]:
    root = ET.fromstring(xml_text)
    source = _source_name(feed_url)

    items = []

    channel = root.find("channel")
    if channel is not None:
        for item in channel.findall("item"):
            title = _first_text(item, ["title"])
            link = _first_text(item, ["link"])
            summary = _first_text(item, ["description"])
            pub = _first_text(item, ["pubDate", "date"])
            tags = [n.text.strip() for n in item.findall("category") if n.text and n.text.strip()]
            if title and link:
                items.append(
                    {
                        "source": source,
                        "title": title,
                        "summary": summary,
                        "url": link,
                        "tags": tags,
                        "published_at": _parse_datetime(pub),
                        "content_text": summary,
                        "content_html": summary,
                    }
                )
        return items

    for entry in root.findall("{http://www.w3.org/2005/Atom}entry"):
        title = _first_text(entry, ["{http://www.w3.org/2005/Atom}title"])
        link = _atom_link(entry)
        summary = _first_text(
            entry,
            [
                "{http://www.w3.org/2005/Atom}summary",
                "{http://www.w3.org/2005/Atom}content",
            ],
        )
        pub = _first_text(
            entry,
            [
                "{http://www.w3.org/2005/Atom}published",
                "{http://www.w3.org/2005/Atom}updated",
            ],
        )
        tags = [
            (cat.attrib.get("term") or "").strip()
            for cat in entry.findall("{http://www.w3.org/2005/Atom}category")
            if (cat.attrib.get("term") or "").strip()
        ]
        if title and link:
            items.append(
                {
                    "source": source,
                    "title": title,
                    "summary": summary,
                    "url": link,
                    "tags": tags,
                    "published_at": _parse_datetime(pub),
                    "content_text": summary,
                    "content_html": summary,
                }
            )

    return items


async def collect_sources() -> list[dict]:
    sources_env = os.environ.get(
        "INFO_ENGINE_SOURCES",
        "https://hnrss.org/frontpage,https://lobste.rs/rss",
    )
    sources = [s.strip() for s in sources_env.split(",") if s.strip()]
    if not sources:
        log.warning("No INFO_ENGINE_SOURCES configured")
        return []

    timeout = httpx.Timeout(15.0)
    all_items = []

    async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
        for source_url in sources:
            try:
                response = await client.get(source_url)
                response.raise_for_status()
                parsed = _parse_feed(source_url, response.text)
                all_items.extend(parsed)
                log.info("Collected %d items from %s", len(parsed), source_url)
            except Exception as exc:
                log.error("Failed to collect from %s: %s", source_url, exc)

    return all_items