Update rss_processor.py

rss_processor.py

@@ -1,3 +1,4 @@
+# rss_processor.py
 import os
 import feedparser
 from langchain.vectorstores import Chroma
@@ -10,18 +11,17 @@ import rss_feeds
 from datetime import datetime
 import dateutil.parser
 import hashlib
-import re  # For cleaning HTML and whitespace
+import re
 
 # Setup logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 # Constants
-MAX_ARTICLES_PER_FEED = 10
+MAX_ARTICLES_PER_FEED = 1000
 LOCAL_DB_DIR = "chroma_db"
 RSS_FEEDS = rss_feeds.RSS_FEEDS
 COLLECTION_NAME = "news_articles"
-
 HF_API_TOKEN = os.getenv("DEMO_HF_API_TOKEN", "YOUR_HF_API_TOKEN")
 REPO_ID = "broadfield-dev/news-rag-db"
 
@@ -43,9 +43,7 @@ def clean_text(text):
     """Clean text by removing HTML tags and extra whitespace."""
     if not text or not isinstance(text, str):
         return ""
-    # Remove HTML tags
     text = re.sub(r'<.*?>', '', text)
-    # Normalize whitespace (remove extra spaces, newlines, tabs)
     text = ' '.join(text.split())
     return text.strip().lower()
 
@@ -67,14 +65,12 @@ def fetch_rss_feeds():
                 link = entry.get("link", "")
                 description = entry.get("summary", entry.get("description", ""))
                 
-                # Clean and normalize all text fields
                 title = clean_text(title)
                 link = clean_text(link)
                 description = clean_text(description)
 
-                # Try multiple date fields and parse flexibly
                 published = "Unknown Date"
-                for date_field in ["published", "updated", "created", "pubDate"]:  # Added "pubDate" for broader compatibility
+                for date_field in ["published", "updated", "created", "pubDate"]:
                     if date_field in entry:
                         try:
                             parsed_date = dateutil.parser.parse(entry[date_field])
@@ -84,13 +80,11 @@ def fetch_rss_feeds():
                             logger.debug(f"Failed to parse {date_field} '{entry[date_field]}': {e}")
                             continue
 
-                # Use a robust key for deduplication, including cleaned fields
-                description_hash = hashlib.sha256(description.encode('utf-8')).hexdigest()  # Switched to SHA256 for better uniqueness
+                description_hash = hashlib.sha256(description.encode('utf-8')).hexdigest()
                 key = f"{title}|{link}|{published}|{description_hash}"
                 if key not in seen_keys:
                     seen_keys.add(key)
-                    # Try multiple image sources
-                    image = "svg"  # Default fallback
+                    image = "svg"
                     for img_source in [
                         lambda e: clean_text(e.get("media_content", [{}])[0].get("url")) if e.get("media_content") else "",
                         lambda e: clean_text(e.get("media_thumbnail", [{}])[0].get("url")) if e.get("media_thumbnail") else "",
@@ -114,50 +108,58 @@ def fetch_rss_feeds():
                         "image": image,
                     })
                     article_count += 1
-                else:
-                    logger.debug(f"Duplicate article skipped in feed {feed_url}: {key}")
         except Exception as e:
             logger.error(f"Error fetching {feed_url}: {e}")
     logger.info(f"Total articles fetched: {len(articles)}")
     return articles
 
 def categorize_feed(url):
-    if "nature" in url.lower() or "science.org" in url.lower() or "arxiv.org" in url.lower() or "plos.org" in url.lower() or "annualreviews.org" in url.lower() or "journals.uchicago.edu" in url.lower() or "jneurosci.org" in url.lower() or "cell.com" in url.lower() or "nejm.org" in url.lower() or "lancet.com" in url.lower():
+    """Categorize an RSS feed based on its URL."""
+    if not url or not isinstance(url, str):
+        logger.warning(f"Invalid URL provided for categorization: {url}")
+        return "Uncategorized"
+
+    url = url.lower().strip()  # Normalize the URL
+
+    logger.debug(f"Categorizing URL: {url}")  # Add debugging for visibility
+
+    if any(keyword in url for keyword in ["nature", "science.org", "arxiv.org", "plos.org", "annualreviews.org", "journals.uchicago.edu", "jneurosci.org", "cell.com", "nejm.org", "lancet.com"]):
         return "Academic Papers"
-    elif "reuters.com/business" in url.lower() or "bloomberg.com" in url.lower() or "ft.com" in url.lower() or "marketwatch.com" in url.lower() or "cnbc.com" in url.lower() or "foxbusiness.com" in url.lower() or "wsj.com" in url.lower() or "bworldonline.com" in url.lower() or "economist.com" in url.lower() or "forbes.com" in url.lower():
+    elif any(keyword in url for keyword in ["reuters.com/business", "bloomberg.com", "ft.com", "marketwatch.com", "cnbc.com", "foxbusiness.com", "wsj.com", "bworldonline.com", "economist.com", "forbes.com"]):
         return "Business"
-    elif "investing.com" in url.lower() or "cnbc.com/market" in url.lower() or "marketwatch.com/market" in url.lower() or "fool.co.uk" in url.lower() or "zacks.com" in url.lower() or "seekingalpha.com" in url.lower() or "barrons.com" in url.lower() or "yahoofinance.com" in url.lower():
+    elif any(keyword in url for keyword in ["investing.com", "cnbc.com/market", "marketwatch.com/market", "fool.co.uk", "zacks.com", "seekingalpha.com", "barrons.com", "yahoofinance.com"]):
         return "Stocks & Markets"
-    elif "whitehouse.gov" in url.lower() or "state.gov" in url.lower() or "commerce.gov" in url.lower() or "transportation.gov" in url.lower() or "ed.gov" in url.lower() or "dol.gov" in url.lower() or "justice.gov" in url.lower() or "federalreserve.gov" in url.lower() or "occ.gov" in url.lower() or "sec.gov" in url.lower() or "bls.gov" in url.lower() or "usda.gov" in url.lower() or "gao.gov" in url.lower() or "cbo.gov" in url.lower() or "fema.gov" in url.lower() or "defense.gov" in url.lower() or "hhs.gov" in url.lower() or "energy.gov" in url.lower() or "interior.gov" in url.lower():
+    elif any(keyword in url for keyword in ["whitehouse.gov", "state.gov", "commerce.gov", "transportation.gov", "ed.gov", "dol.gov", "justice.gov", "federalreserve.gov", "occ.gov", "sec.gov", "bls.gov", "usda.gov", "gao.gov", "cbo.gov", "fema.gov", "defense.gov", "hhs.gov", "energy.gov", "interior.gov"]):
         return "Federal Government"
-    elif "weather.gov" in url.lower() or "metoffice.gov.uk" in url.lower() or "accuweather.com" in url.lower() or "weatherunderground.com" in url.lower() or "noaa.gov" in url.lower() or "wunderground.com" in url.lower() or "climate.gov" in url.lower() or "ecmwf.int" in url.lower() or "bom.gov.au" in url.lower():
+    elif any(keyword in url for keyword in ["weather.gov", "metoffice.gov.uk", "accuweather.com", "weatherunderground.com", "noaa.gov", "wunderground.com", "climate.gov", "ecmwf.int", "bom.gov.au"]):
         return "Weather"
-    elif "data.worldbank.org" in url.lower() or "imf.org" in url.lower() or "un.org" in url.lower() or "oecd.org" in url.lower() or "statista.com" in url.lower() or "kff.org" in url.lower() or "who.int" in url.lower() or "cdc.gov" in url.lower() or "bea.gov" in url.lower() or "census.gov" in url.lower() or "fdic.gov" in url.lower():
+    elif any(keyword in url for keyword in ["data.worldbank.org", "imf.org", "un.org", "oecd.org", "statista.com", "kff.org", "who.int", "cdc.gov", "bea.gov", "census.gov", "fdic.gov"]):
         return "Data & Statistics"
-    elif "nasa" in url.lower() or "spaceweatherlive" in url.lower() or "space" in url.lower() or "universetoday" in url.lower() or "skyandtelescope" in url.lower() or "esa" in url.lower():
+    elif any(keyword in url for keyword in ["nasa", "spaceweatherlive", "space", "universetoday", "skyandtelescope", "esa"]):
         return "Space"
-    elif "sciencedaily" in url.lower() or "quantamagazine" in url.lower() or "smithsonianmag" in url.lower() or "popsci" in url.lower() or "discovermagazine" in url.lower() or "scientificamerican" in url.lower() or "newscientist" in url.lower() or "livescience" in url.lower() or "atlasobscura" in url.lower():
+    elif any(keyword in url for keyword in ["sciencedaily", "quantamagazine", "smithsonianmag", "popsci", "discovermagazine", "scientificamerican", "newscientist", "livescience", "atlasobscura"]):
         return "Science"
-    elif "wired" in url.lower() or "techcrunch" in url.lower() or "arstechnica" in url.lower() or "gizmodo" in url.lower() or "theverge" in url.lower():
+    elif any(keyword in url for keyword in ["wired", "techcrunch", "arstechnica", "gizmodo", "theverge"]):
         return "Tech"
-    elif "horoscope" in url.lower() or "astrostyle" in url.lower():
+    elif any(keyword in url for keyword in ["horoscope", "astrostyle"]):
         return "Astrology"
-    elif "cnn_allpolitics" in url.lower() or "bbci.co.uk/news/politics" in url.lower() or "reuters.com/arc/outboundfeeds/newsletter-politics" in url.lower() or "politico.com/rss/politics" in url.lower() or "thehill" in url.lower():
+    elif any(keyword in url for keyword in ["cnn_allpolitics", "bbci.co.uk/news/politics", "reuters.com/arc/outboundfeeds/newsletter-politics", "politico.com/rss/politics", "thehill"]):
         return "Politics"
-    elif "weather" in url.lower() or "swpc.noaa.gov" in url.lower() or "foxweather" in url.lower():
+    elif any(keyword in url for keyword in ["weather", "swpc.noaa.gov", "foxweather"]):
         return "Earth Weather"
-    elif "vogue" in url.lower():
+    elif "vogue" in url:
         return "Lifestyle"
-    elif "phys.org" in url.lower() or "aps.org" in url.lower() or "physicsworld" in url.lower():
+    elif any(keyword in url for keyword in ["phys.org", "aps.org", "physicsworld"]):
         return "Physics"
-    return "Uncategorized"
+    else:
+        logger.warning(f"No matching category found for URL: {url}")
+        return "Uncategorized"
 
 def process_and_store_articles(articles):
     documents = []
-    existing_ids = set(vector_db.get()["ids"])  # Get existing document IDs to avoid duplicates
+    existing_ids = set(vector_db.get()["ids"])  # Load existing IDs once
     for article in articles:
         try:
-            # Clean and normalize all fields
             title = clean_text(article["title"])
             link = clean_text(article["link"])
             description = clean_text(article["description"])
@@ -177,29 +179,28 @@ def process_and_store_articles(articles):
             }
             doc = Document(page_content=description, metadata=metadata, id=doc_id)
             documents.append(doc)
+            existing_ids.add(doc_id)  # Update in-memory set to avoid duplicates within this batch
         except Exception as e:
             logger.error(f"Error processing article {article['title']}: {e}")
     
     if documents:
         try:
             vector_db.add_documents(documents)
-            vector_db.persist()  # Explicitly persist changes
-            logger.info(f"Added {len(documents)} new articles to DB")
+            vector_db.persist()
+            logger.info(f"Added {len(documents)} new articles to DB. Total documents: {len(vector_db.get()['ids'])}")
         except Exception as e:
             logger.error(f"Error storing articles: {e}")
 
 def download_from_hf_hub():
-    # Only download if the local DB doesn’t exist (initial setup)
     if not os.path.exists(LOCAL_DB_DIR):
         try:
             hf_api.create_repo(repo_id=REPO_ID, repo_type="dataset", exist_ok=True, token=HF_API_TOKEN)
             logger.info(f"Downloading Chroma DB from {REPO_ID}...")
-            hf_api.download_repo(repo_id=REPO_ID, repo_type="dataset", local_dir=LOCAL_DB_DIR, token=HF_API_TOKEN)
+            hf_api.hf_hub_download(repo_id=REPO_ID, filename="chroma_db", local_dir=LOCAL_DB_DIR, repo_type="dataset", token=HF_API_TOKEN)
         except Exception as e:
             logger.error(f"Error downloading from Hugging Face Hub: {e}")
-            raise
     else:
-        logger.info("Local Chroma DB already exists, skipping download.")
+        logger.info("Local Chroma DB exists, loading existing data.")
 
 def upload_to_hf_hub():
     if os.path.exists(LOCAL_DB_DIR):
@@ -219,9 +220,9 @@ def upload_to_hf_hub():
             logger.info(f"Database uploaded to: {REPO_ID}")
         except Exception as e:
             logger.error(f"Error uploading to Hugging Face Hub: {e}")
-            raise
 
 if __name__ == "__main__":
+    download_from_hf_hub()  # Ensure DB is initialized
     articles = fetch_rss_feeds()
     process_and_store_articles(articles)
     upload_to_hf_hub()