bvandeusen
e613485474
- Add trafilatura + html2text to dependencies - Replace custom HTMLStripper with html2text for RSS feed content - Fetch full article text via httpx + trafilatura after each new item is stored; falls back to RSS-provided content if fetch/extraction fails - Raise CONTENT_MAX_CHARS from 2000 to 50000 (TEXT column, no migration needed) - Re-embed items with full article content once enrichment completes - Startup backfill enriches existing items with short content (<1000 chars) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
287 lines
11 KiB
Python
287 lines
11 KiB
Python
"""RSS feed service: fetch, parse with feedparser, and cache items to DB."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
import feedparser
|
|
import html2text
|
|
import httpx
|
|
from sqlalchemy import select, text
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.rss_feed import RssFeed, RssItem
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Safety cap on stored content — effectively unlimited for typical articles
|
|
CONTENT_MAX_CHARS = 50_000
|
|
# Keep only items from the last N days to avoid unbounded growth
|
|
ITEM_MAX_AGE_DAYS = 90
|
|
|
|
_h2t = html2text.HTML2Text()
|
|
_h2t.ignore_links = True
|
|
_h2t.ignore_images = True
|
|
_h2t.ignore_emphasis = False
|
|
_h2t.body_width = 0 # No line-wrapping
|
|
|
|
|
|
def _html_to_text(html: str) -> str:
|
|
"""Convert HTML to clean plain text via html2text."""
|
|
if not html or "<" not in html:
|
|
return html
|
|
try:
|
|
return _h2t.handle(html).strip()
|
|
except Exception:
|
|
return html
|
|
|
|
|
|
async def _fetch_full_article(url: str) -> str | None:
|
|
"""Fetch a URL and extract its main article text via trafilatura.
|
|
|
|
Returns clean plain text, or None if extraction fails or yields nothing useful.
|
|
Runs trafilatura in a thread executor since it does synchronous HTML parsing.
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True, headers={
|
|
"User-Agent": "Mozilla/5.0 (compatible; FabledAssistant/1.0; +https://fabledsword.com)",
|
|
}) as client:
|
|
resp = await client.get(url)
|
|
resp.raise_for_status()
|
|
raw_html = resp.text
|
|
except Exception:
|
|
logger.debug("Failed to fetch article URL %s", url)
|
|
return None
|
|
|
|
loop = asyncio.get_event_loop()
|
|
try:
|
|
import trafilatura
|
|
text = await loop.run_in_executor(
|
|
None,
|
|
lambda: trafilatura.extract(
|
|
raw_html,
|
|
include_comments=False,
|
|
include_tables=True,
|
|
favor_recall=True,
|
|
),
|
|
)
|
|
return text or None
|
|
except Exception:
|
|
logger.debug("trafilatura extraction failed for %s", url, exc_info=True)
|
|
return None
|
|
|
|
|
|
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 = _html_to_text(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.
|
|
"""
|
|
scheme = url.split("://")[0].lower() if "://" in url else ""
|
|
if scheme not in ("http", "https"):
|
|
logger.warning("Blocked RSS fetch with non-http(s) scheme: %s", url[:80])
|
|
return 0
|
|
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
|
|
feed_user_id: int | None = None
|
|
|
|
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)
|
|
feed_user_id = feed_row.user_id
|
|
# 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()
|
|
|
|
# Collect IDs of unclassified items after commit.
|
|
# We query classified_at IS NULL (not just the items inserted above) because
|
|
# classification is best-effort and may have failed on previous fetches.
|
|
# Re-queuing all unclassified items for this feed on each fetch is intentional:
|
|
# it provides automatic retry without a separate retry loop. The classifier
|
|
# only writes to items it successfully classifies, so already-classified items
|
|
# are not re-processed (they have classified_at set).
|
|
unclassified_ids: list[int] = []
|
|
new_item_data: list[tuple[int, str, str]] = [] # (id, title, content) for embedding
|
|
if new_count > 0:
|
|
result = await session.execute(
|
|
select(RssItem.id, RssItem.title, RssItem.content, RssItem.classified_at).where(
|
|
RssItem.feed_id == feed_id,
|
|
)
|
|
)
|
|
for row in result.fetchall():
|
|
item_id, title, content, classified_at = row
|
|
if classified_at is None:
|
|
unclassified_ids.append(item_id)
|
|
new_item_data.append((item_id, title or "", content or ""))
|
|
|
|
# Prune old items to keep DB tidy
|
|
await _prune_old_items(feed_id)
|
|
|
|
# Fire-and-forget classification for unclassified items
|
|
if unclassified_ids and feed_user_id is not None:
|
|
from fabledassistant.services.rss_classifier import classify_and_store
|
|
asyncio.create_task(classify_and_store(unclassified_ids, feed_user_id))
|
|
|
|
# Collect (id, url) for newly inserted items to enrich with full article text
|
|
new_items_for_enrichment: list[tuple[int, str]] = []
|
|
if new_count > 0:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(RssItem.id, RssItem.url).where(
|
|
RssItem.feed_id == feed_id,
|
|
RssItem.url != "",
|
|
).order_by(RssItem.fetched_at.desc()).limit(new_count)
|
|
)
|
|
new_items_for_enrichment = list(result.fetchall())
|
|
|
|
# Fire-and-forget: fetch full article text, then re-embed with richer content
|
|
if new_items_for_enrichment and feed_user_id is not None:
|
|
from fabledassistant.services.embeddings import upsert_rss_item_embedding
|
|
|
|
async def _enrich_and_embed() -> None:
|
|
for item_id, article_url in new_items_for_enrichment:
|
|
full_text = await _fetch_full_article(article_url)
|
|
if full_text and len(full_text) > 200:
|
|
async with async_session() as session:
|
|
item = await session.get(RssItem, item_id)
|
|
if item:
|
|
item.content = full_text[:CONTENT_MAX_CHARS]
|
|
await session.commit()
|
|
await upsert_rss_item_embedding(
|
|
item_id, feed_user_id, item.title or "", item.content
|
|
)
|
|
else:
|
|
# Enrich failed — still embed with RSS-provided content
|
|
async with async_session() as session:
|
|
item = await session.get(RssItem, item_id)
|
|
if item:
|
|
await upsert_rss_item_embedding(
|
|
item_id, feed_user_id, item.title or "", item.content or ""
|
|
)
|
|
await asyncio.sleep(0.5) # Polite pacing between article fetches
|
|
|
|
asyncio.create_task(_enrich_and_embed())
|
|
elif new_item_data and feed_user_id is not None:
|
|
# No URLs to enrich — embed with RSS content only
|
|
from fabledassistant.services.embeddings import upsert_rss_item_embedding
|
|
|
|
async def _embed_only() -> None:
|
|
for item_id, title, content in new_item_data:
|
|
await upsert_rss_item_embedding(item_id, feed_user_id, title, content)
|
|
await asyncio.sleep(0.05)
|
|
|
|
asyncio.create_task(_embed_only())
|
|
|
|
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 '90 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
|