#!/usr/bin/env python3 """ Telegram Bot implementation for AI News Digest """ import asyncio import logging import os from telegram import Update from telegram.ext import Application, CommandHandler, ContextTypes from bot import AINewsBot from datetime import datetime # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TelegramNewsBot: """Telegram wrapper for AI News Bot""" def __init__(self, telegram_token: str): self.telegram_token = telegram_token self.news_bot = AINewsBot() self.application = Application.builder().token(telegram_token).build() # Add command handlers self.application.add_handler(CommandHandler("start", self.start_command)) self.application.add_handler(CommandHandler("digest", self.digest_command)) self.application.add_handler(CommandHandler("help", self.help_command)) self.application.add_handler(CommandHandler("subscribe", self.subscribe_command)) async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /start command""" welcome_text = """ 🤖 **Welcome to AI News Digest Bot!** Get the latest AI news curated daily from top tech sources. **Commands:** /digest - Get today's AI news digest /help - Show this help message /subscribe - Get daily updates (coming soon) **Features:** • Curated news from TechCrunch, MIT Tech Review, and more • Categorized by Breaking, Business, Research, and Tools • Updated daily at 8 AM UTC Ready to stay informed about AI? Try /digest now! """ await update.message.reply_text(welcome_text, parse_mode='Markdown') async def digest_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /digest command""" try: await update.message.reply_text("🤖 Generating your AI news digest... ⚡") # Generate digest result = await self.news_bot.run_daily_digest() if not result: await update.message.reply_text("❌ No AI news available right now. Try again later!") return # Format for Telegram telegram_content = self.format_for_telegram(result['items']) # Send digest (split if too long) await self.send_long_message(update, telegram_content) logger.info(f"Sent digest to user {update.effective_user.id}") except Exception as e: logger.error(f"Error sending digest: {e}") await update.message.reply_text("❌ Error generating digest. Please try again later.") async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /help command""" help_text = """ 🤖 **AI News Digest Bot Help** **Commands:** /start - Welcome message and overview /digest - Get today's curated AI news /help - Show this help message /subscribe - Daily updates (premium feature) **About:** This bot curates the most important AI news from trusted sources like TechCrunch, MIT Technology Review, and Ars Technica. News is categorized by: 🔥 **Breaking** - Major announcements and launches 📊 **Business** - Funding, acquisitions, market news 💡 **Research** - Academic papers and studies 🛠️ **Tools** - New AI tools and platforms **Premium Features (coming soon):** • Personalized topics (Web3, robotics, enterprise AI) • Trend analysis and predictions • Early morning delivery (5 AM vs 8 AM) • Weekly deep-dive reports Questions? Contact support or try /digest to get started! """ await update.message.reply_text(help_text, parse_mode='Markdown') async def subscribe_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /subscribe command""" subscribe_text = """ 📬 **Daily Subscriptions - Coming Soon!** We're working on automatic daily delivery of AI news directly to your Telegram. **What you'll get:** • Daily AI digest at 8 AM UTC • No spam - just the important news • Easy unsubscribe anytime **Premium upgrades:** • Personalized topics • Earlier delivery (5 AM) • Weekly trend analysis • Only $5/month For now, use /digest anytime to get the latest news! Want to be notified when subscriptions launch? Save this chat and we'll announce it here first. """ await update.message.reply_text(subscribe_text, parse_mode='Markdown') def format_for_telegram(self, categorized_items): """Format digest for Telegram posting""" today = datetime.now().strftime("%B %d, %Y") digest = f"🤖 **AI News 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: # Use Telegram-style links digest += f"• [{item.title}]({item.url})\n" digest += f" _{item.source}_\n\n" digest += "---\n" digest += "*Want personalized AI news? Premium coming soon for $5/mo*\n\n" digest += "Use /help for more commands or /digest for fresh news anytime!" return digest async def send_long_message(self, update: Update, content: str, max_length: int = 4000): """Send long message, splitting if necessary""" if len(content) <= max_length: await update.message.reply_text(content, parse_mode='Markdown') return # Split content parts = self._split_telegram_message(content, max_length) for i, part in enumerate(parts): await update.message.reply_text(part, parse_mode='Markdown') if i < len(parts) - 1: await asyncio.sleep(1) # Brief delay between messages def _split_telegram_message(self, content: str, max_length: int) -> list: """Split message for Telegram (preserves markdown)""" if len(content) <= max_length: return [content] parts = [] lines = content.split('\n') current_part = "" for line in lines: if len(current_part + line + '\n') <= max_length: current_part += line + '\n' else: if current_part: parts.append(current_part.strip()) current_part = line + '\n' else: # Line too long, force split parts.append(line[:max_length]) current_part = line[max_length:] if current_part: parts.append(current_part.strip()) return parts async def run_webhook(self, webhook_url: str, port: int = 8443): """Run bot with webhook (for production)""" await self.application.bot.set_webhook(webhook_url) await self.application.run_webhook( listen="0.0.0.0", port=port, webhook_url=webhook_url ) async def run_polling(self): """Run bot with polling (for testing)""" logger.info("Starting Telegram bot with polling...") await self.application.run_polling() async def main(): """Main function for testing""" import argparse parser = argparse.ArgumentParser(description='AI News Digest Telegram Bot') parser.add_argument('--polling', action='store_true', help='Run with polling (for testing)') parser.add_argument('--webhook', help='Webhook URL for production') parser.add_argument('--port', type=int, default=8443, help='Port for webhook') args = parser.parse_args() # Get Telegram token telegram_token = os.environ.get('TELEGRAM_BOT_TOKEN') if not telegram_token: print("Error: TELEGRAM_BOT_TOKEN environment variable not set") print("\nTo get a token:") print("1. Message @BotFather on Telegram") print("2. Send: /newbot") print("3. Follow instructions to create bot") print("4. Set token: export TELEGRAM_BOT_TOKEN='your_token_here'") return # Create and run bot bot = TelegramNewsBot(telegram_token) try: if args.webhook: await bot.run_webhook(args.webhook, args.port) else: await bot.run_polling() except KeyboardInterrupt: logger.info("Bot stopped by user") except Exception as e: logger.error(f"Bot error: {e}") if __name__ == "__main__": asyncio.run(main())