"""RSS / Atom feeds (feedparser) → normalized items.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any import feedparser from aiatlas.sdk.extract.html import clean_text @dataclass class FeedItem: id: str url: str title: str summary: str | None published_at: datetime | None updated_at: datetime | None authors: list[str] = field(default_factory=list) categories: list[str] = field(default_factory=list) content_html: str | None = None raw: dict[str, Any] = field(default_factory=dict, repr=False) def _dt(struct: Any) -> datetime | None: if not struct: return None try: return datetime(*struct[:6], tzinfo=UTC) except Exception: # noqa: BLE001 return None def parse_feed(content: bytes | str) -> tuple[dict[str, Any], list[FeedItem]]: parsed = feedparser.parse(content) feed_meta = {"title": parsed.feed.get("title"), "link": parsed.feed.get("link"), "updated": parsed.feed.get("updated"), "subtitle": parsed.feed.get("subtitle"), "bozo": bool(parsed.get("bozo"))} items: list[FeedItem] = [] for e in parsed.entries: url = e.get("link") or "" if not url: for link in e.get("links", []): if link.get("rel") in (None, "alternate") and link.get("href"): url = link["href"] break content_html = None if e.get("content"): content_html = e["content"][0].get("value") summary = e.get("summary") or e.get("description") items.append(FeedItem( id=e.get("id") or e.get("guid") or url, url=url, title=clean_text(e.get("title") or "")[:500], summary=clean_text(_strip_tags(summary))[:4000] if summary else None, published_at=_dt(e.get("published_parsed")) or _dt(e.get("created_parsed")), updated_at=_dt(e.get("updated_parsed")), authors=[a.get("name") for a in e.get("authors", []) if a.get("name")] or ([e["author"]] if e.get("author") else []), categories=[t.get("term") for t in e.get("tags", []) if t.get("term")], content_html=content_html, raw={k: v for k, v in e.items() if isinstance(v, (str, int, float))}, )) return feed_meta, items def _strip_tags(s: str) -> str: from selectolax.parser import HTMLParser try: return HTMLParser(s).text(separator=" ") except Exception: # noqa: BLE001 return s __all__ = ["FeedItem", "parse_feed"]