#!/usr/bin/env python3 """ AI News Digest Bot - Discord/Telegram implementation Daily curation of AI news from RSS sources """ import discord import asyncio import feedparser import requests from datetime import datetime, timezone import json import re from typing import List, Dict, Any from dataclasses import dataclass import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class NewsItem: title: str url: str source: str summary: str published: datetime category: str = "general" class AINewsBot: def __init__(self, discord_token: str = None, telegram_token: str = None): self.discord_token = discord_token self.telegram_token = telegram_token # RSS Sources for AI news self.sources = { "TechCrunch AI": "https://techcrunch.com/category/artificial-intelligence/feed/", "VentureBeat AI": "https://venturebeat.com/ai/feed/", "The Verge AI": "https://www.theverge.com/ai-artificial-intelligence/rss/index.xml", "Ars Technica AI": "http://feeds.arstechnica.com/arstechnica/technology-lab", "MIT Tech Review": "https://www.technologyreview.com/feed/", } # Keywords for categorization self.categories = { "breaking": ["breakthrough", "launches", "releases", "announces", "unveils"], "business": ["funding", "investment", "revenue", "acquisition", "IPO", "market"], "research": ["paper", "study", "research", "university", "arxiv", "nature"], "tools": ["API", "tool", "platform", "service", "app", "software"] } self.discord_client = None if discord_token: intents = discord.Intents.default() intents.message_content = True self.discord_client = discord.Client(intents=intents) async def fetch_rss_feed(self, url: str, source_name: str) -> List[NewsItem]: """Fetch and parse RSS feed from source""" try: # Use requests to fetch the feed response = requests.get(url, timeout=10) response.raise_for_status() feed = feedparser.parse(response.content) items = [] for entry in feed.entries[:10]: # Limit to 10 most recent # Parse published date published = datetime.now(timezone.utc) if hasattr(entry, 'published_parsed') and entry.published_parsed: published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc) # Extract summary summary = "" if hasattr(entry, 'summary'): # Clean HTML tags from summary summary = re.sub(r'<[^>]+>', '', entry.summary)[:200] + "..." # Categorize based on title/summary content category = self._categorize_item(entry.title + " " + summary) item = NewsItem( title=entry.title, url=entry.link, source=source_name, summary=summary, published=published, category=category ) items.append(item) logger.info(f"Fetched {len(items)} items from {source_name}") return items except Exception as e: logger.error(f"Error fetching RSS from {source_name}: {e}") return [] def _categorize_item(self, text: str) -> str: """Categorize news item based on content""" text_lower = text.lower() for category, keywords in self.categories.items(): if any(keyword in text_lower for keyword in keywords): return category return "general" async def curate_daily_digest(self) -> Dict[str, List[NewsItem]]: """Aggregate and curate news from all sources""" all_items = [] # Fetch from all sources for source_name, url in self.sources.items(): items = await self.fetch_rss_feed(url, source_name) all_items.extend(items) # Filter for AI-relevant content ai_keywords = ["ai", "artificial intelligence", "machine learning", "neural", "gpt", "llm", "chatbot", "automation", "algorithm", "deep learning"] ai_items = [] for item in all_items: item_text = (item.title + " " + item.summary).lower() if any(keyword in item_text for keyword in ai_keywords): ai_items.append(item) # Remove duplicates by title similarity deduplicated = self._deduplicate_items(ai_items) # Sort by category and recency categorized = {} for item in sorted(deduplicated, key=lambda x: x.published, reverse=True): if item.category not in categorized: categorized[item.category] = [] categorized[item.category].append(item) # Limit items per category for category in categorized: categorized[category] = categorized[category][:5] logger.info(f"Curated {len([item for items in categorized.values() for item in items])} AI news items") return categorized def _deduplicate_items(self, items: List[NewsItem]) -> List[NewsItem]: """Remove duplicate items based on title similarity""" unique_items = [] seen_titles = set() for item in items: # Simple deduplication by first 50 chars of title title_key = item.title[:50].lower() if title_key not in seen_titles: seen_titles.add(title_key) unique_items.append(item) return unique_items def format_digest_discord(self, categorized_items: Dict[str, List[NewsItem]]) -> str: """Format digest for Discord posting""" today = datetime.now().strftime("%B %d, %Y") digest = f"🤖 **AI Daily Digest** — {today}\n\n" category_emojis = { "breaking": "🔥", "business": "📊", "research": "💡", "tools": "🛠️", "general": "📰" } for category, items in categorized_items.items(): if not items: continue emoji = category_emojis.get(category, "📰") digest += f"**{emoji} {category.title()}:**\n" for item in items: digest += f"- [{item.title}]({item.url}) ({item.source})\n" digest += "\n" digest += "---\n*Want personalized AI news? Try premium for $5/mo*" return digest async def post_to_discord(self, channel_id: int, content: str): """Post digest to Discord channel""" if not self.discord_client: logger.error("Discord client not initialized") return try: channel = self.discord_client.get_channel(channel_id) if channel: await channel.send(content) logger.info(f"Posted digest to Discord channel {channel_id}") else: logger.error(f"Discord channel {channel_id} not found") except Exception as e: logger.error(f"Error posting to Discord: {e}") async def run_daily_digest(self, discord_channel_id: int = None): """Main method to run daily digest generation and posting""" logger.info("Starting daily AI news digest generation...") # Curate news categorized_items = await self.curate_daily_digest() if not any(categorized_items.values()): logger.warning("No AI news items found") return # Format for Discord discord_content = self.format_digest_discord(categorized_items) # Post to Discord if configured if self.discord_client and discord_channel_id: await self.post_to_discord(discord_channel_id, discord_content) # Return formatted content for other use return { "discord": discord_content, "items": categorized_items, "timestamp": datetime.now(timezone.utc).isoformat() } # Test function async def test_digest(): """Test the digest generation locally""" bot = AINewsBot() result = await bot.run_daily_digest() if result: print("=== AI NEWS DIGEST TEST ===") print(result["discord"]) print(f"\nTotal items: {len([item for items in result['items'].values() for item in items])}") else: print("No digest generated") if __name__ == "__main__": # Test locally asyncio.run(test_digest())