spb/ai-atlas

1"""RSS / Atom feeds (feedparser) → normalized items."""2from __future__ import annotations34from dataclasses import dataclass, field5from datetime import UTC, datetime6from typing import Any78import feedparser910from aiatlas.sdk.extract.html import clean_text111213@dataclass14class FeedItem:15    id: str16    url: str17    title: str18    summary: str | None19    published_at: datetime | None20    updated_at: datetime | None21    authors: list[str] = field(default_factory=list)22    categories: list[str] = field(default_factory=list)23    content_html: str | None = None24    raw: dict[str, Any] = field(default_factory=dict, repr=False)252627def _dt(struct: Any) -> datetime | None:28    if not struct:29        return None30    try:31        return datetime(*struct[:6], tzinfo=UTC)32    except Exception:  # noqa: BLE00133        return None343536def parse_feed(content: bytes | str) -> tuple[dict[str, Any], list[FeedItem]]:37    parsed = feedparser.parse(content)38    feed_meta = {"title": parsed.feed.get("title"), "link": parsed.feed.get("link"), "updated": parsed.feed.get("updated"),39                 "subtitle": parsed.feed.get("subtitle"), "bozo": bool(parsed.get("bozo"))}40    items: list[FeedItem] = []41    for e in parsed.entries:42        url = e.get("link") or ""43        if not url:44            for link in e.get("links", []):45                if link.get("rel") in (None, "alternate") and link.get("href"):46                    url = link["href"]47                    break48        content_html = None49        if e.get("content"):50            content_html = e["content"][0].get("value")51        summary = e.get("summary") or e.get("description")52        items.append(FeedItem(53            id=e.get("id") or e.get("guid") or url,54            url=url,55            title=clean_text(e.get("title") or "")[:500],56            summary=clean_text(_strip_tags(summary))[:4000] if summary else None,57            published_at=_dt(e.get("published_parsed")) or _dt(e.get("created_parsed")),58            updated_at=_dt(e.get("updated_parsed")),59            authors=[a.get("name") for a in e.get("authors", []) if a.get("name")] or ([e["author"]] if e.get("author") else []),60            categories=[t.get("term") for t in e.get("tags", []) if t.get("term")],61            content_html=content_html,62            raw={k: v for k, v in e.items() if isinstance(v, (str, int, float))},63        ))64    return feed_meta, items656667def _strip_tags(s: str) -> str:68    from selectolax.parser import HTMLParser6970    try:71        return HTMLParser(s).text(separator=" ")72    except Exception:  # noqa: BLE00173        return s747576__all__ = ["FeedItem", "parse_feed"]77