#!/usr/bin/env python3 """ RSS Aggregator for AI News Digest Bot Fetches articles from multiple RSS sources and consolidates them. """ import feedparser import json import requests import logging from datetime import datetime, timedelta from typing import List, Dict, Any import hashlib import time import os from urllib.parse import urljoin, urlparse # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class RSSAggregator: def __init__(self, sources_file: str = "references/sources.json", cache_dir: str = "cache"): """Initialize the RSS aggregator with sources configuration.""" self.sources_file = sources_file self.cache_dir = cache_dir self.sources = self.load_sources() self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'AI-News-Digest/1.0 (Automated RSS Aggregator)' }) # Create cache directory if it doesn't exist os.makedirs(cache_dir, exist_ok=True) def load_sources(self) -> List[Dict[str, Any]]: """Load RSS sources from JSON file.""" try: with open(self.sources_file, 'r') as f: data = json.load(f) return data.get('sources', []) except FileNotFoundError: logger.error(f"Sources file not found: {self.sources_file}") return [] except json.JSONDecodeError as e: logger.error(f"Error parsing sources file: {e}") return [] def fetch_feed(self, source: Dict[str, Any]) -> List[Dict[str, Any]]: """Fetch and parse a single RSS feed.""" feed_url = source.get('rss') if not feed_url: logger.warning(f"No RSS URL for source: {source.get('name', 'Unknown')}") return [] try: logger.info(f"Fetching: {source.get('name', 'Unknown')} - {feed_url}") # Add timeout and retries response = self.session.get(feed_url, timeout=30) response.raise_for_status() # Parse RSS feed feed = feedparser.parse(response.content) if feed.bozo: logger.warning(f"Feed parsing warning for {source.get('name')}: {feed.bozo_exception}") articles = [] for entry in feed.entries: article = self.parse_entry(entry, source) if article: articles.append(article) logger.info(f"Fetched {len(articles)} articles from {source.get('name')}") return articles except requests.exceptions.RequestException as e: logger.error(f"Error fetching {source.get('name')}: {e}") return [] except Exception as e: logger.error(f"Unexpected error with {source.get('name')}: {e}") return [] def parse_entry(self, entry: Any, source: Dict[str, Any]) -> Dict[str, Any]: """Parse a single RSS entry into our standard format.""" try: # Extract publication date published = None if hasattr(entry, 'published_parsed') and entry.published_parsed: published = datetime(*entry.published_parsed[:6]) elif hasattr(entry, 'updated_parsed') and entry.updated_parsed: published = datetime(*entry.updated_parsed[:6]) # Create unique ID for deduplication article_id = hashlib.md5( f"{entry.get('title', '')}{entry.get('link', '')}".encode() ).hexdigest() article = { 'id': article_id, 'title': entry.get('title', '').strip(), 'link': entry.get('link', ''), 'description': entry.get('summary', '').strip(), 'published': published.isoformat() if published else None, 'source': { 'name': source.get('name'), 'category': source.get('category'), 'weight': source.get('weight', 0.5), 'url': source.get('url') }, 'fetched_at': datetime.now().isoformat() } # Clean up description HTML tags if present if article['description']: import re article['description'] = re.sub(r'<[^>]+>', '', article['description']) article['description'] = article['description'][:500] # Limit length return article except Exception as e: logger.error(f"Error parsing entry: {e}") return None def fetch_all_feeds(self, max_age_hours: int = 24) -> List[Dict[str, Any]]: """Fetch articles from all configured RSS sources.""" all_articles = [] cutoff_time = datetime.now() - timedelta(hours=max_age_hours) for source in self.sources: try: articles = self.fetch_feed(source) # Filter by age if publication date is available filtered_articles = [] for article in articles: if article.get('published'): pub_date = datetime.fromisoformat(article['published']) if pub_date >= cutoff_time: filtered_articles.append(article) else: # Include articles without publication date filtered_articles.append(article) all_articles.extend(filtered_articles) # Small delay to be respectful to servers time.sleep(1) except Exception as e: logger.error(f"Error processing source {source.get('name', 'Unknown')}: {e}") continue # Deduplicate articles by ID seen_ids = set() unique_articles = [] for article in all_articles: if article['id'] not in seen_ids: seen_ids.add(article['id']) unique_articles.append(article) logger.info(f"Aggregated {len(unique_articles)} unique articles from {len(self.sources)} sources") return unique_articles def save_cache(self, articles: List[Dict[str, Any]], filename: str = None): """Save articles to cache file.""" if filename is None: filename = f"articles_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" cache_path = os.path.join(self.cache_dir, filename) try: with open(cache_path, 'w') as f: json.dump({ 'fetched_at': datetime.now().isoformat(), 'count': len(articles), 'articles': articles }, f, indent=2) logger.info(f"Saved {len(articles)} articles to {cache_path}") return cache_path except Exception as e: logger.error(f"Error saving cache: {e}") return None def load_cache(self, filename: str) -> List[Dict[str, Any]]: """Load articles from cache file.""" cache_path = os.path.join(self.cache_dir, filename) try: with open(cache_path, 'r') as f: data = json.load(f) return data.get('articles', []) except FileNotFoundError: logger.error(f"Cache file not found: {cache_path}") return [] except json.JSONDecodeError as e: logger.error(f"Error parsing cache file: {e}") return [] def main(): """Main function for command-line usage.""" import argparse parser = argparse.ArgumentParser(description='Aggregate RSS feeds for AI news') parser.add_argument('--sources', default='references/sources.json', help='Sources JSON file') parser.add_argument('--output', help='Output file path') parser.add_argument('--max-age', type=int, default=24, help='Maximum age of articles in hours') parser.add_argument('--cache-dir', default='cache', help='Cache directory') args = parser.parse_args() aggregator = RSSAggregator(args.sources, args.cache_dir) articles = aggregator.fetch_all_feeds(args.max_age) # Save to cache cache_file = aggregator.save_cache(articles) # Also save to specified output if provided if args.output: try: with open(args.output, 'w') as f: json.dump({ 'fetched_at': datetime.now().isoformat(), 'count': len(articles), 'articles': articles }, f, indent=2) print(f"Saved {len(articles)} articles to {args.output}") except Exception as e: print(f"Error saving to output file: {e}") print(f"Aggregation complete. Found {len(articles)} unique articles.") return articles if __name__ == "__main__": main()