Update app.py

app.py

@@ -1,31 +1,25 @@
 import os
-import feedparser
-from langchain.vectorstores import Chroma
-from langchain.embeddings import HuggingFaceEmbeddings
-from langchain.docstore.document import Document
+import threading
+from flask import Flask, render_template, request, jsonify
+from rss_processor import fetch_rss_feeds, process_and_store_articles, download_from_hf_hub, upload_to_hf_hub, clean_text
 import logging
-from huggingface_hub import HfApi, login, snapshot_download
-import shutil
-import rss_feeds
-from datetime import datetime, date
-import dateutil.parser
+import time
+from datetime import datetime
 import hashlib
-import re
+import glob
+from langchain.vectorstores import Chroma
+from langchain.embeddings import HuggingFaceEmbeddings
+
+app = Flask(__name__)
 
 # Setup logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# Constants
-MAX_ARTICLES_PER_FEED = 10
-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"
-
-# Initialize Hugging Face API
-login(token=HF_API_TOKEN)
-hf_api = HfApi()
+# Global flag to track background loading
+loading_complete = True  # Start as True to allow initial rendering
+last_update_time = time.time()
+last_data_hash = None  # Track the hash of the last data to detect changes
 
 def get_embedding_model():
     """Returns a singleton instance of the embedding model to avoid reloading."""
@@ -33,213 +27,363 @@ def get_embedding_model():
         get_embedding_model.model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
     return get_embedding_model.model
 
-def get_daily_db_dir():
-    """Returns the path for today's Chroma DB."""
-    return f"chroma_db_{date.today().isoformat()}"
+def load_feeds_in_background():
+    global loading_complete, last_update_time
+    try:
+        logger.info("Starting background RSS feed fetch")
+        articles = fetch_rss_feeds()
+        logger.info(f"Fetched {len(articles)} articles")
+        process_and_store_articles(articles)
+        last_update_time = time.time()
+        logger.info("Background feed processing complete")
+        upload_to_hf_hub()
+    except Exception as e:
+        logger.error(f"Error in background feed loading: {e}")
+    finally:
+        loading_complete = True
 
-def clean_text(text):
-    """Clean text by removing HTML tags and extra whitespace."""
-    if not text or not isinstance(text, str):
-        return ""
-    text = re.sub(r'<.*?>', '', text)
-    text = ' '.join(text.split())
-    return text.strip().lower()
-
-def fetch_rss_feeds():
-    articles = []
-    seen_keys = set()
-    for feed_url in RSS_FEEDS:
+def get_all_docs_from_dbs():
+    """Aggregate documents and metadata from all Chroma DB folders."""
+    all_docs = {'documents': [], 'metadatas': []}
+    seen_ids = set()
+    embedding_function = get_embedding_model()
+
+    for db_path in glob.glob("chroma_db_*"):
+        if not os.path.isdir(db_path):
+            continue
         try:
-            logger.info(f"Fetching {feed_url}")
-            feed = feedparser.parse(feed_url)
-            if feed.bozo:
-                logger.warning(f"Parse error for {feed_url}: {feed.bozo_exception}")
+            temp_vector_db = Chroma(
+                persist_directory=db_path,
+                embedding_function=embedding_function,
+                collection_name="news_articles"
+            )
+            # Skip empty databases
+            if temp_vector_db._collection.count() == 0:
                 continue
-            article_count = 0
-            for entry in feed.entries:
-                if article_count >= MAX_ARTICLES_PER_FEED:
-                    break
-                title = entry.get("title", "No Title")
-                link = entry.get("link", "")
-                description = entry.get("summary", entry.get("description", ""))
-                
-                title = clean_text(title)
-                link = clean_text(link)
-                description = clean_text(description)
-
-                published = "Unknown Date"
-                for date_field in ["published", "updated", "created", "pubDate"]:
-                    if date_field in entry:
-                        try:
-                            parsed_date = dateutil.parser.parse(entry[date_field])
-                            published = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
-                            break
-                        except (ValueError, TypeError) as e:
-                            logger.debug(f"Failed to parse {date_field} '{entry[date_field]}': {e}")
-                            continue
-
-                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)
-                    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 "",
-                        lambda e: clean_text(e.get("enclosure", {}).get("url")) if e.get("enclosure") else "",
-                        lambda e: clean_text(next((lnk.get("href") for lnk in e.get("links", []) if lnk.get("type", "").startswith("image")), "")),
-                    ]:
-                        try:
-                            img = img_source(entry)
-                            if img and img.strip():
-                                image = img
-                                break
-                        except (IndexError, AttributeError, TypeError):
-                            continue
-
-                    articles.append({
-                        "title": title,
-                        "link": link,
-                        "description": description,
-                        "published": published,
-                        "category": categorize_feed(feed_url),
-                        "image": image,
-                    })
-                    article_count += 1
+
+            db_data = temp_vector_db.get(include=['documents', 'metadatas'])
+            if db_data.get('documents') and db_data.get('metadatas'):
+                for doc, meta in zip(db_data['documents'], db_data['metadatas']):
+                    # Use a more robust unique identifier
+                    doc_id = f"{meta.get('title', 'No Title')}|{meta.get('link', '')}|{meta.get('published', 'Unknown Date')}"
+                    if doc_id not in seen_ids:
+                        seen_ids.add(doc_id)
+                        all_docs['documents'].append(doc)
+                        all_docs['metadatas'].append(meta)
         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):
-    """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()
-
-    logger.debug(f"Categorizing URL: {url}")
-
-    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 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 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 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 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 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 any(keyword in url for keyword in ["nasa", "spaceweatherlive", "space", "universetoday", "skyandtelescope", "esa"]):
-        return "Space"
-    elif any(keyword in url for keyword in ["sciencedaily", "quantamagazine", "smithsonianmag", "popsci", "discovermagazine", "scientificamerican", "newscientist", "livescience", "atlasobscura"]):
-        return "Science"
-    elif any(keyword in url for keyword in ["wired", "techcrunch", "arstechnica", "gizmodo", "theverge"]):
-        return "Tech"
-    elif any(keyword in url for keyword in ["horoscope", "astrostyle"]):
-        return "Astrology"
-    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 any(keyword in url for keyword in ["weather", "swpc.noaa.gov", "foxweather"]):
-        return "Earth Weather"
-    elif "vogue" in url:
-        return "Lifestyle"
-    elif any(keyword in url for keyword in ["phys.org", "aps.org", "physicsworld"]):
-        return "Physics"
-    else:
-        logger.warning(f"No matching category found for URL: {url}")
-        return "Uncategorized"
-
-def process_and_store_articles(articles):
-    db_path = get_daily_db_dir()
-    vector_db = Chroma(
-        persist_directory=db_path,
-        embedding_function=get_embedding_model(),
-        collection_name=COLLECTION_NAME
-    )
+            logger.error(f"Error loading DB {db_path}: {e}")
     
+    return all_docs
+
+def compute_data_hash(categorized_articles):
+    """Compute a hash of the current articles to detect changes."""
+    if not categorized_articles:
+        return ""
+    # Create a sorted string representation of the articles for consistent hashing
+    data_str = ""
+    for cat, articles in sorted(categorized_articles.items()):
+        for article in sorted(articles, key=lambda x: x["published"]):
+            data_str += f"{cat}|{article['title']}|{article['link']}|{article['published']}|"
+    return hashlib.sha256(data_str.encode('utf-8')).hexdigest()
+
+@app.route('/')
+def index():
+    global loading_complete, last_update_time, last_data_hash
+
+    # Check if any DB exists; if not, download from Hugging Face
+    if not glob.glob("chroma_db_*"):
+        logger.info("No Chroma DBs found, downloading from Hugging Face Hub...")
+        download_from_hf_hub()
+
+    # Start background RSS feed update
+    loading_complete = False
+    threading.Thread(target=load_feeds_in_background, daemon=True).start()
+
+    # Load existing data immediately
     try:
-        existing_ids = set(vector_db.get(include=[])["ids"])
-    except Exception:
-        existing_ids = set()
+        all_docs = get_all_docs_from_dbs()
+        total_docs = len(all_docs['documents'])
+        logger.info(f"Total articles across all DBs at startup: {total_docs}")
+        if not all_docs.get('metadatas'):
+            logger.info("No articles in any DB yet")
+            return render_template("index.html", categorized_articles={}, has_articles=False, loading=True)
+
+        # Process and categorize articles with deduplication
+        enriched_articles = []
+        seen_keys = set()
+        for doc, meta in zip(all_docs['documents'], all_docs['metadatas']):
+            if not meta:
+                continue
+            title = meta.get("title", "No Title")
+            link = meta.get("link", "")
+            description = meta.get("original_description", "No Description")
+            published = meta.get("published", "Unknown Date").strip()
+
+            title = clean_text(title)
+            link = clean_text(link)
+            description = clean_text(description)
 
-    docs_to_add = []
-    ids_to_add = []
-    
-    for article in articles:
-        try:
-            title = clean_text(article["title"])
-            link = clean_text(article["link"])
-            description = clean_text(article["description"])
-            published = article["published"]
             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:
+                    published = datetime.strptime(published, "%Y-%m-%d %H:%M:%S").isoformat() if "Unknown" not in published else published
+                except (ValueError, TypeError):
+                    published = "1970-01-01T00:00:00"
+                enriched_articles.append({
+                    "title": title,
+                    "link": link,
+                    "description": description,
+                    "category": meta.get("category", "Uncategorized"),
+                    "published": published,
+                    "image": meta.get("image", "svg"),
+                })
+
+        enriched_articles.sort(key=lambda x: x["published"], reverse=True)
+
+        categorized_articles = {}
+        for article in enriched_articles:
+            cat = article["category"]
+            if cat not in categorized_articles:
+                categorized_articles[cat] = []
+            categorized_articles[cat].append(article)
+
+        categorized_articles = dict(sorted(categorized_articles.items(), key=lambda x: x[0].lower()))
+
+        for cat in categorized_articles:
+            categorized_articles[cat] = sorted(categorized_articles[cat], key=lambda x: x["published"], reverse=True)[:10]
+            if len(categorized_articles[cat]) >= 2:
+                logger.debug(f"Category {cat} top 2: {categorized_articles[cat][0]['title']} | {categorized_articles[cat][1]['title']}")
+
+        # Compute initial data hash
+        last_data_hash = compute_data_hash(categorized_articles)
+
+        logger.info(f"Displaying articles at startup: {sum(len(articles) for articles in categorized_articles.values())} total")
+        return render_template("index.html", 
+                              categorized_articles=categorized_articles, 
+                              has_articles=True, 
+                              loading=True)
+    except Exception as e:
+        logger.error(f"Error retrieving articles at startup: {e}")
+        return render_template("index.html", categorized_articles={}, has_articles=False, loading=True)
+
+@app.route('/search', methods=['POST'])
+def search():
+    query = request.form.get('search')
+    if not query:
+        logger.info("Empty search query received")
+        return jsonify({"categorized_articles": {}, "has_articles": False, "loading": False})
+
+    try:
+        logger.info(f"Performing semantic search for: '{query}'")
+        
+        embedding_function = get_embedding_model()
+        enriched_articles = []
+        seen_keys = set()
+        db_paths = glob.glob("chroma_db_*")
+
+        if not db_paths:
+            logger.warning("No Chroma DBs found for search.")
+            return jsonify({"categorized_articles": {}, "has_articles": False, "loading": False})
+
+        all_search_results = []
+        for db_path in db_paths:
+            if not os.path.isdir(db_path): continue
+            try:
+                vector_db = Chroma(
+                    persist_directory=db_path,
+                    embedding_function=embedding_function,
+                    collection_name="news_articles"
+                )
+                if vector_db._collection.count() > 0:
+                    results = vector_db.similarity_search_with_relevance_scores(query, k=20)
+                    all_search_results.extend(results)
+            except Exception as e:
+                logger.error(f"Error searching in DB {db_path}: {e}")
+
+        # Sort all results by relevance score (higher is better)
+        all_search_results.sort(key=lambda x: x[1], reverse=True)
+
+        # Process and deduplicate top results
+        for doc, score in all_search_results:
+            meta = doc.metadata
+            title = clean_text(meta.get("title", "No Title"))
+            link = clean_text(meta.get("link", ""))
+            description = clean_text(meta.get("original_description", "No Description"))
+            published = meta.get("published", "Unknown Date").strip()
+
+            description_hash = hashlib.sha256(description.encode('utf-8')).hexdigest()
+            key = f"{title}|{link}|{published}|{description_hash}"
             
-            doc_id = f"{title}|{link}|{published}|{description_hash}"
-            
-            if doc_id in existing_ids:
-                logger.debug(f"Skipping duplicate in DB {db_path}: {doc_id}")
+            if key not in seen_keys:
+                seen_keys.add(key)
+                enriched_articles.append({
+                    "title": meta.get("title", "No Title"),
+                    "link": meta.get("link", ""),
+                    "description": meta.get("original_description", "No Description"),
+                    "category": meta.get("category", "Uncategorized"),
+                    "published": published,
+                    "image": meta.get("image", "svg"),
+                })
+
+        logger.info(f"Found {len(enriched_articles)} unique articles from semantic search.")
+        if not enriched_articles:
+            return jsonify({"categorized_articles": {}, "has_articles": False, "loading": False})
+
+        # Categorize the articles
+        categorized_articles = {}
+        for article in enriched_articles:
+            cat = article["category"]
+            categorized_articles.setdefault(cat, []).append(article)
+
+        return jsonify({
+            "categorized_articles": categorized_articles,
+            "has_articles": bool(enriched_articles),
+            "loading": False
+        })
+    except Exception as e:
+        logger.error(f"Semantic search error: {e}", exc_info=True)
+        return jsonify({"categorized_articles": {}, "has_articles": False, "loading": False}), 500
+
+
+@app.route('/check_loading')
+def check_loading():
+    global loading_complete, last_update_time
+    if loading_complete:
+        return jsonify({"status": "complete", "last_update": last_update_time})
+    return jsonify({"status": "loading"}), 202
+
+@app.route('/get_updates')
+def get_updates():
+    global last_update_time, last_data_hash
+    try:
+        all_docs = get_all_docs_from_dbs()
+        if not all_docs.get('metadatas'):
+            return jsonify({"articles": [], "last_update": last_update_time, "has_updates": False})
+
+        enriched_articles = []
+        seen_keys = set()
+        for doc, meta in zip(all_docs['documents'], all_docs['metadatas']):
+            if not meta:
                 continue
+            title = meta.get("title", "No Title")
+            link = meta.get("link", "")
+            description = meta.get("original_description", "No Description")
+            published = meta.get("published", "Unknown Date").strip()
 
-            metadata = {
-                "title": article["title"],
-                "link": article["link"],
-                "original_description": article["description"],
-                "published": article["published"],
-                "category": article["category"],
-                "image": article["image"],
-            }
-            doc = Document(page_content=description, metadata=metadata)
-            docs_to_add.append(doc)
-            ids_to_add.append(doc_id)
-            existing_ids.add(doc_id)
-        except Exception as e:
-            logger.error(f"Error processing article {article.get('title', 'N/A')}: {e}")
-    
-    if docs_to_add:
-        try:
-            vector_db.add_documents(documents=docs_to_add, ids=ids_to_add)
-            vector_db.persist()
-            logger.info(f"Added {len(docs_to_add)} new articles to DB {db_path}. Total in DB: {vector_db._collection.count()}")
-        except Exception as e:
-            logger.error(f"Error storing articles in {db_path}: {e}")
+            title = clean_text(title)
+            link = clean_text(link)
+            description = clean_text(description)
+
+            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:
+                    published = datetime.strptime(published, "%Y-%m-%d %H:%M:%S").isoformat() if "Unknown" not in published else published
+                except (ValueError, TypeError):
+                    published = "1970-01-01T00:00:00"
+                enriched_articles.append({
+                    "title": title,
+                    "link": link,
+                    "description": description,
+                    "category": meta.get("category", "Uncategorized"),
+                    "published": published,
+                    "image": meta.get("image", "svg"),
+                })
+
+        enriched_articles.sort(key=lambda x: x["published"], reverse=True)
+        categorized_articles = {}
+        for article in enriched_articles:
+            cat = article["category"]
+            if cat not in categorized_articles:
+                categorized_articles[cat] = []
+            key = f"{article['title']}|{article['link']}|{article['published']}"
+            if key not in [f"{a['title']}|{a['link']}|{a['published']}" for a in categorized_articles[cat]]:
+                categorized_articles[cat].append(article)
+
+        for cat in categorized_articles:
+            unique_articles = []
+            seen_cat_keys = set()
+            for article in sorted(categorized_articles[cat], key=lambda x: x["published"], reverse=True):
+                key = f"{clean_text(article['title'])}|{clean_text(article['link'])}|{article['published']}"
+                if key not in seen_cat_keys:
+                    seen_cat_keys.add(key)
+                    unique_articles.append(article)
+            categorized_articles[cat] = unique_articles[:10]
 
-def download_from_hf_hub():
+        # Compute hash of new data
+        current_data_hash = compute_data_hash(categorized_articles)
+
+        # Compare with last data hash to determine if there are updates
+        has_updates = last_data_hash != current_data_hash
+        if has_updates:
+            logger.info("New RSS data detected, sending updates to frontend")
+            last_data_hash = current_data_hash
+            return jsonify({
+                "articles": categorized_articles,
+                "last_update": last_update_time,
+                "has_updates": True
+            })
+        else:
+            logger.info("No new RSS data, skipping update")
+            return jsonify({
+                "articles": {},
+                "last_update": last_update_time,
+                "has_updates": False
+            })
+    except Exception as e:
+        logger.error(f"Error fetching updates: {e}")
+        return jsonify({"articles": {}, "last_update": last_update_time, "has_updates": False}), 500
+
+@app.route('/get_all_articles/<category>')
+def get_all_articles(category):
     try:
-        hf_api.create_repo(repo_id=REPO_ID, repo_type="dataset", exist_ok=True, token=HF_API_TOKEN)
-        logger.info(f"Downloading all DBs from {REPO_ID}...")
-        snapshot_download(
-            repo_id=REPO_ID,
-            repo_type="dataset",
-            local_dir=".",
-            local_dir_use_symlinks=False,
-            allow_patterns="chroma_db_*/**",
-            token=HF_API_TOKEN
-        )
-        logger.info("Finished downloading DBs.")
+        all_docs = get_all_docs_from_dbs()
+        if not all_docs.get('metadatas'):
+            return jsonify({"articles": [], "category": category})
+
+        enriched_articles = []
+        seen_keys = set()
+        for doc, meta in zip(all_docs['documents'], all_docs['metadatas']):
+            if not meta or meta.get("category") != category:
+                continue
+            title = meta.get("title", "No Title")
+            link = meta.get("link", "")
+            description = meta.get("original_description", "No Description")
+            published = meta.get("published", "Unknown Date").strip()
+
+            title = clean_text(title)
+            link = clean_text(link)
+            description = clean_text(description)
+
+            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:
+                    published = datetime.strptime(published, "%Y-%m-%d %H:%M:%S").isoformat() if "Unknown" not in published else published
+                except (ValueError, TypeError):
+                    published = "1970-01-01T00:00:00"
+                enriched_articles.append({
+                    "title": title,
+                    "link": link,
+                    "description": description,
+                    "category": meta.get("category", "Uncategorized"),
+                    "published": published,
+                    "image": meta.get("image", "svg"),
+                })
+
+        enriched_articles.sort(key=lambda x: x["published"], reverse=True)
+        return jsonify({"articles": enriched_articles, "category": category})
     except Exception as e:
-        logger.error(f"Error downloading from Hugging Face Hub: {e}")
+        logger.error(f"Error fetching all articles for category {category}: {e}")
+        return jsonify({"articles": [], "category": category}), 500
 
-def upload_to_hf_hub():
-    db_path = get_daily_db_dir()
-    if os.path.exists(db_path):
-        try:
-            logger.info(f"Uploading updated Chroma DB '{db_path}' to {REPO_ID}...")
-            hf_api.upload_folder(
-                folder_path=db_path,
-                path_in_repo=db_path,
-                repo_id=REPO_ID,
-                repo_type="dataset",
-                token=HF_API_TOKEN
-            )
-            logger.info(f"Database folder '{db_path}' uploaded to: {REPO_ID}")
-        except Exception as e:
-            logger.error(f"Error uploading to Hugging Face Hub: {e}")
+@app.route('/card')
+def card_load():
+    return render_template("card.html")
 
 if __name__ == "__main__":
-    download_from_hf_hub()
-    articles = fetch_rss_feeds()
-    process_and_store_articles(articles)
-    upload_to_hf_hub()
+    app.run(host="0.0.0.0", port=7860)