Optimize RSS feed processing with parallel parsing and improved error handling

scraper/scraper.py

@@ -23,6 +23,8 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 FEED_FILE = os.getenv("FEED_FILE", "./rss_short_feed.json")
+MAX_FEED_WORKERS = int(os.getenv("MAX_FEED_WORKERS", "10"))
+MAX_ARTICLE_WORKERS = int(os.getenv("MAX_ARTICLE_WORKERS", "10"))
 
 # Ensure necessary NLTK resources are downloaded
 d = Downloader()
@@ -69,17 +71,44 @@ def mine_all_articles(rss_feed_sources, limit=None):
 
     sources = rss_feed_sources["rss_feeds"]
 
-    for site, data in sources.items():
+    # Parse RSS feeds in parallel for better performance
-        logger.info(f"Parsing RSS feed: {data['rss_url']}")
+    def parse_single_feed(site, data):
+        """Parse a single RSS feed with timeout and error handling"""
         try:
-            feed = feedparser.parse(data["rss_url"])
+            logger.info(f"Parsing RSS feed: {data['rss_url']}")
+            # Add more aggressive timeout settings
+            feed = feedparser.parse(data["rss_url"], timeout=15)  # 15 second timeout
             feed_entries = feed.entries[:limit] if limit else feed.entries
 
+            entries = []
             for entry in feed_entries:
                 if "link" in entry and "title" in entry:
-                    all_links.append((site, entry.title, entry.link))
+                    entries.append((site, entry.title, entry.link))
+            return entries
         except Exception as e:
             logger.error(f"Error parsing RSS feed: {site} Error: {str(e)}")
+            return []
+
+    # Use ThreadPoolExecutor for parallel RSS feed parsing
+    from concurrent.futures import ThreadPoolExecutor, as_completed
+    max_workers = min(10, len(sources))  # Limit concurrent workers
+    
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        # Submit all feed parsing tasks
+        future_to_site = {
+            executor.submit(parse_single_feed, site, data): site 
+            for site, data in sources.items()
+        }
+        
+        # Collect results as they complete
+        for future in as_completed(future_to_site, timeout=30):  # 30 second overall timeout
+            try:
+                entries = future.result()
+                all_links.extend(entries)
+            except Exception as e:
+                site = future_to_site[future]
+                logger.error(f"Error processing feed for {site}: {str(e)}")
+
     return all_links
 
 
@@ -271,36 +300,41 @@ def pull_article(link, source, title=None, save_to_file=True):
 
 def safe_pull_articles(article_list):
     """
-    Safely pull articles with improved error handling and reduced parallelism.
+    Safely pull articles with improved error handling and increased parallelism.
     """
+    if not article_list:
+        return [], []
+
     results = []
     errors = []
 
-    # Process in smaller batches to reduce resource strain
+    # Use ThreadPoolExecutor for parallel article pulling with higher concurrency
-    batch_size = 5
+    # Use configurable worker setting
+    max_workers = min(MAX_ARTICLE_WORKERS, len(article_list))  # Cap at configured workers, but don't exceed article count
+    batch_size = max(1, min(20, len(article_list) // 4))  # Dynamic batch size
     
-    for i in range(0, len(article_list), batch_size):
+    logger.info(f"Starting parallel article pulling with {max_workers} workers and batch size {batch_size}")
-        batch = article_list[i : i + batch_size]
-        logger.info(f"Processing batch {i // batch_size + 1} with {len(batch)} articles")
 
-        # Use ThreadPoolExecutor instead of ProcessPoolExecutor to avoid
+    # Process all articles in parallel with proper error handling
-        # process termination issues with browser automation
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        with ThreadPoolExecutor(max_workers=3) as executor:  # Reduced workers
+        # Submit all tasks at once for maximum parallelism
-            futures = [
+        futures = [
-                executor.submit(pull_article, link, source, title)
+            executor.submit(pull_article, link, source, title)
-                for source, title, link in batch
+            for source, title, link in article_list
-            ]
+        ]
 
-            for future in as_completed(futures):
+        # Collect results as they complete
-                try:
+        for i, future in enumerate(as_completed(futures, timeout=300)):  # 5 minute timeout total
-                    result = future.result(timeout=120)  # 2 minute timeout
+            try:
+                result = future.result(timeout=120)  # 2 minute timeout per article
+                if result:  # Only count non-empty results
                     results.append(result)
-                except Exception as e:
+                # Log progress every 100 articles
-                    logger.error(f"Error in pull_article: {e}")
+                if (i + 1) % 100 == 0:
-                    errors.append(e)
+                    logger.info(f"Processed {i + 1} articles out of {len(article_list)}")
-
+            except Exception as e:
-        # Add a small delay between batches to reduce system load
+                logger.error(f"Error processing article: {e}")
-        time.sleep(5)
+                errors.append(e)
 
     return results, errors