Create workflows/rss_poller.py

workflows/rss_poller.py

@@ -0,0 +1,67 @@
+from prefect import flow, task
+import feedparser
+from typing import List, Dict
+from datetime import datetime
+from database.operations import is_entry_processed, save_rss_entry, SessionLocal
+from config.settings import settings
+import hashlib
+
+@task(retries=3, retry_delay_seconds=60)
+def fetch_rss_feed(feed_url: str) -> List[Dict]:
+    """Fetch and parse RSS feed"""
+    feed = feedparser.parse(feed_url)
+    entries = []
+    
+    for entry in feed.entries:
+        # Create unique ID for entry
+        entry_id = hashlib.md5(
+            f"{feed_url}{entry.get('link', '')}".encode()
+        ).hexdigest()
+        
+        entries.append({
+            "entry_id": entry_id,
+            "feed_url": feed_url,
+            "title": entry.get("title", ""),
+            "link": entry.get("link", ""),
+            "published": datetime(*entry.published_parsed[:6]) if hasattr(entry, 'published_parsed') else datetime.utcnow(),
+            "content": entry.get("summary", "") or entry.get("description", "")
+        })
+    
+    return entries
+
+@task
+def filter_new_entries(entries: List[Dict]) -> List[Dict]:
+    """Filter out already processed entries"""
+    db = SessionLocal()
+    new_entries = []
+    
+    try:
+        for entry in entries:
+            if not is_entry_processed(db, entry["entry_id"]):
+                # Save to database
+                save_rss_entry(
+                    db,
+                    entry_id=entry["entry_id"],
+                    feed_url=entry["feed_url"],
+                    title=entry["title"],
+                    link=entry["link"],
+                    published=entry["published"],
+                    content=entry["content"]
+                )
+                new_entries.append(entry)
+    finally:
+        db.close()
+    
+    return new_entries
+
+@flow(name="RSS Feed Polling")
+def poll_rss_feeds():
+    """Poll all RSS feeds and return new entries"""
+    all_new_entries = []
+    
+    for feed_url in settings.rss_feed_list:
+        entries = fetch_rss_feed(feed_url)
+        new_entries = filter_new_entries(entries)
+        all_new_entries.extend(new_entries)
+    
+    return all_new_entries