News/RSS connector — aggregate headlines from RSS and Atom feeds.
Uses stdlib xml.etree.ElementTree for parsing (no extra dependencies).
Config file lists feeds to follow. All HTTP calls are in module-level
functions for easy mocking in tests.
NewsRSSConnector(*, config_path: str = _DEFAULT_CONFIG_PATH)
Bases: BaseConnector
Aggregate headlines from configured RSS/Atom feeds.
Source code in src/openjarvis/connectors/news_rss.py
| def __init__(self, *, config_path: str = _DEFAULT_CONFIG_PATH) -> None:
self._config_path = Path(config_path)
self._status = SyncStatus()
|
configure(feeds: List[Dict[str, Any]]) -> None
Validate and persist RSS feed configuration from the connect UI.
Source code in src/openjarvis/connectors/news_rss.py
| def configure(self, feeds: List[Dict[str, Any]]) -> None:
"""Validate and persist RSS feed configuration from the connect UI."""
normalized: List[Dict[str, str]] = []
for feed in feeds:
if not isinstance(feed, dict):
continue
url = str(feed.get("url", "")).strip()
_validate_feed_url(url)
parsed = urlparse(url)
name = str(feed.get("name", "")).strip() or parsed.netloc
normalized.append({"name": name, "url": url})
if not normalized:
raise ValueError("At least one RSS feed URL is required")
from openjarvis.security.file_utils import secure_write_json
secure_write_json(self._config_path, {"feeds": normalized})
|
sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]
Yield Documents for recent items across all configured feeds.
Source code in src/openjarvis/connectors/news_rss.py
| def sync(
self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
"""Yield Documents for recent items across all configured feeds."""
feeds = self._load_config()
for feed in feeds:
feed_name = feed.get("name", "Unknown Feed")
feed_url = feed.get("url", "")
if not feed_url:
continue
try:
xml_text = _fetch_feed(feed_url)
except (httpx.HTTPError, ValueError):
continue
items = _parse_rss_items(xml_text)
for item in items:
pub_dt = _parse_pub_date(item["pubDate"])
# Filter by since if the date is parseable
if since and pub_dt and pub_dt.replace(tzinfo=None) < since:
continue
title = item["title"] or "Untitled"
doc_id = f"rss-{feed_name}-{title[:40]}"
yield Document(
doc_id=doc_id,
source="news_rss",
doc_type="article",
content=item["description"],
title=title,
timestamp=pub_dt or datetime.now(),
url=item["link"] or None,
metadata={"feed_name": feed_name},
)
self._status.state = "idle"
self._status.last_sync = datetime.now()
|