feat: RSS service — feedparser fetch, cache, prune

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-11 19:46:01 -04:00
parent bbdcea57a7
commit 9bb054f3d5
2 changed files with 190 additions and 0 deletions
@@ -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