From 9bb054f3d5052daf642380386df30cd78d434ae7 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 11 Mar 2026 19:46:01 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20RSS=20service=20=E2=80=94=20feedparser?= =?UTF-8?q?=20fetch,=20cache,=20prune?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- src/fabledassistant/services/rss.py | 148 ++++++++++++++++++++++++++++ tests/test_rss_service.py | 42 ++++++++ 2 files changed, 190 insertions(+) create mode 100644 src/fabledassistant/services/rss.py create mode 100644 tests/test_rss_service.py diff --git a/src/fabledassistant/services/rss.py b/src/fabledassistant/services/rss.py new file mode 100644 index 0000000..1c02f16 --- /dev/null +++ b/src/fabledassistant/services/rss.py @@ -0,0 +1,148 @@ +"""RSS feed service: fetch, parse with feedparser, and cache items to DB.""" + +import asyncio +import logging +from datetime import datetime, timezone + +import feedparser +from sqlalchemy import select, text + +from fabledassistant.models import async_session +from fabledassistant.models.rss_feed import RssFeed, RssItem + +logger = logging.getLogger(__name__) + +CONTENT_MAX_CHARS = 2000 +# Keep only items from the last N days to avoid unbounded growth +ITEM_MAX_AGE_DAYS = 14 + + +def extract_item(entry) -> dict: + """Extract a clean item dict from a feedparser entry object.""" + # Prefer full content over summary (feedparser uses a list of Content objects) + content = "" + raw_content = getattr(entry, "content", None) + if isinstance(raw_content, list) and raw_content: + content = raw_content[0].value + else: + content = entry.get("summary", "") + content = content[:CONTENT_MAX_CHARS] + + pub = None + if entry.published_parsed: + try: + pub = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc) + except Exception: + pass + + return { + "guid": entry.get("id", entry.get("link", "")), + "title": entry.get("title", ""), + "url": entry.get("link", ""), + "content": content, + "published_at": pub, + } + + +async def _parse_feed(url: str) -> feedparser.FeedParserDict: + """Run feedparser in a thread executor so we don't block the event loop.""" + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, feedparser.parse, url) + + +async def fetch_and_cache_feed(feed_id: int, url: str) -> int: + """ + Fetch a feed URL, parse it, and upsert new items into rss_items. + Returns the number of new items stored. + """ + try: + parsed = await _parse_feed(url) + except Exception: + logger.warning("Failed to fetch RSS feed %s", url, exc_info=True) + return 0 + + if parsed.bozo and not parsed.entries: + logger.warning("Malformed RSS feed %s: %s", url, parsed.bozo_exception) + return 0 + + new_count = 0 + async with async_session() as session: + for entry in parsed.entries: + item_data = extract_item(entry) + if not item_data["guid"]: + continue + # Check if already stored + existing = await session.execute( + select(RssItem).where( + RssItem.feed_id == feed_id, + RssItem.guid == item_data["guid"], + ) + ) + if existing.scalars().first() is not None: + continue + item = RssItem( + feed_id=feed_id, + **item_data, + ) + session.add(item) + new_count += 1 + + # Update last_fetched_at on the feed + feed_row = await session.get(RssFeed, feed_id) + if feed_row: + feed_row.last_fetched_at = datetime.now(timezone.utc) + # Auto-populate title from feed metadata if blank + if not feed_row.title and parsed.feed.get("title"): + feed_row.title = parsed.feed.title[:200] + + await session.commit() + + # Prune old items to keep DB tidy + await _prune_old_items(feed_id) + return new_count + + +async def _prune_old_items(feed_id: int) -> None: + """Delete items older than ITEM_MAX_AGE_DAYS from a feed.""" + async with async_session() as session: + await session.execute( + text(""" + DELETE FROM rss_items + WHERE feed_id = :feed_id + AND published_at < NOW() - INTERVAL '14 days' + """).bindparams(feed_id=feed_id) + ) + await session.commit() + + +async def get_recent_items(user_id: int, limit: int = 20) -> list[dict]: + """Return the most recent RSS items across all of a user's feeds.""" + async with async_session() as session: + result = await session.execute( + select(RssItem, RssFeed.title.label("feed_title")) + .join(RssFeed, RssItem.feed_id == RssFeed.id) + .where(RssFeed.user_id == user_id) + .order_by(RssItem.published_at.desc().nullslast()) + .limit(limit) + ) + rows = result.all() + + return [ + {**item.to_dict(), "feed_title": feed_title} + for item, feed_title in rows + ] + + +async def refresh_all_feeds(user_id: int) -> dict[int, int]: + """Fetch all feeds for a user. Returns {feed_id: new_items_count}.""" + async with async_session() as session: + result = await session.execute( + select(RssFeed).where(RssFeed.user_id == user_id) + ) + feeds = list(result.scalars().all()) + + results = {} + for feed in feeds: + count = await fetch_and_cache_feed(feed.id, feed.url) + results[feed.id] = count + return results diff --git a/tests/test_rss_service.py b/tests/test_rss_service.py new file mode 100644 index 0000000..9ba0dad --- /dev/null +++ b/tests/test_rss_service.py @@ -0,0 +1,42 @@ +import pytest +from unittest.mock import patch, MagicMock + + +def test_extract_item_fields(): + """extract_item() should pull title, link, id, summary from a feedparser entry.""" + from fabledassistant.services.rss import extract_item + entry = MagicMock() + entry.get = lambda k, d="": {"title": "Test Post", "link": "https://example.com/1", + "id": "guid-1", "summary": "A summary"}.get(k, d) + entry.published_parsed = None + item = extract_item(entry) + assert item["title"] == "Test Post" + assert item["url"] == "https://example.com/1" + assert item["guid"] == "guid-1" + assert item["content"] == "A summary" + + +def test_extract_item_truncates_content(): + """extract_item() should truncate content to 2000 chars.""" + from fabledassistant.services.rss import extract_item + entry = MagicMock() + entry.get = lambda k, d="": {"summary": "x" * 3000, "title": "", "link": "", "id": "g"}.get(k, d) + entry.published_parsed = None + item = extract_item(entry) + assert len(item["content"]) == 2000 + + +def test_extract_item_prefers_content_over_summary(): + """extract_item() should prefer 'content' field over 'summary' when present.""" + from fabledassistant.services.rss import extract_item + entry = MagicMock() + content_obj = MagicMock() + content_obj.value = "Full content here" + entry.get = lambda k, d="": { + "title": "T", "link": "http://x.com", "id": "g", + "summary": "Short summary", + }.get(k, d) + entry.content = [content_obj] + entry.published_parsed = None + item = extract_item(entry) + assert item["content"] == "Full content here"