import os import logging import asyncio import sys from typing import Dict, Optional from dotenv import load_dotenv from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CallbackQueryHandler, ContextTypes from openai import OpenAI import json from prompts import PYTHON_QUIZ_PROMPT, SYSTEM_MESSAGE # Load environment variables load_dotenv() # Configure logging logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) class TelegramQuizBot: def __init__(self): self.telegram_token = os.getenv('TELEGRAM_BOT_TOKEN') self.openai_api_key = os.getenv('OPENAI_API_KEY') self.openai_model = os.getenv('OPENAI_MODEL', 'gpt-4.1-nano') # Configurable OpenAI model self.chat_id = os.getenv('TELEGRAM_CHAT_ID') if not all([self.telegram_token, self.openai_api_key, self.chat_id]): raise ValueError("Missing required environment variables. Please check your .env file.") # Initialize OpenAI client self.openai_client = OpenAI(api_key=self.openai_api_key) # Store current quiz state self.current_quiz: Optional[Dict] = None self.quiz_active = False # Event for proper shutdown self.shutdown_event = asyncio.Event() # Initialize Telegram application self.application = Application.builder().token(self.telegram_token).build() self.setup_handlers() def setup_handlers(self): """Set up command and callback handlers""" self.application.add_handler(CallbackQueryHandler(self.handle_answer)) def generate_quiz_question(self) -> Dict: """Generate a quiz question using OpenAI""" try: response = self.openai_client.chat.completions.create( model=self.openai_model, # Use configurable model messages=[ {"role": "system", "content": SYSTEM_MESSAGE}, {"role": "user", "content": PYTHON_QUIZ_PROMPT} ], response_format={"type": "json_object"}, max_tokens=800, temperature=0.7 ) quiz_data = json.loads(response.choices[0].message.content) logger.info(f"Generated quiz question using OpenAI model: {self.openai_model}") return quiz_data except Exception as e: logger.error(f"Error generating quiz with OpenAI: {e}") raise async def send_quiz(self): """Send a quiz question to the user""" try: # Generate quiz question self.current_quiz = self.generate_quiz_question() self.quiz_active = True # Format question message with explanation if available question_text = "🧠 *Python Quiz Question* 🐍\n\n" # Add explanation if present if 'explanation' in self.current_quiz: question_text += f"📚 *Concept:*\n{self.current_quiz['explanation']}\n\n" # Add the main question question_text += f"❓ *Question:*\n{self.current_quiz['question']}\n\n" # Create inline keyboard with options keyboard = [] for option, text in self.current_quiz['options'].items(): keyboard.append([InlineKeyboardButton(f"{option}: {text}", callback_data=option)]) reply_markup = InlineKeyboardMarkup(keyboard) # Send quiz message with regular Markdown for better compatibility await self.application.bot.send_message( chat_id=self.chat_id, text=question_text, reply_markup=reply_markup, parse_mode='Markdown' ) logger.info("Quiz sent successfully!") except Exception as e: logger.error(f"Error sending quiz: {e}") # Fallback to no parse mode if Markdown fails try: await self.application.bot.send_message( chat_id=self.chat_id, text=question_text, reply_markup=reply_markup ) except Exception as fallback_error: logger.error(f"Fallback send also failed: {fallback_error}") async def handle_answer(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle user's answer selection""" try: if not self.quiz_active or not self.current_quiz: await update.callback_query.answer("No active quiz found!") return query = update.callback_query selected_answer = query.data correct_answer = self.current_quiz['correct_answer'] await query.answer() # Prepare response message using markdown formatting if selected_answer == correct_answer: result_emoji = "✅" result_text = "Correct!" else: result_emoji = "❌" result_text = "Incorrect!" response_message = f"{result_emoji} *{result_text}*\n\n" response_message += f"*Your answer:* {selected_answer}\n" response_message += f"*Correct answer:* {correct_answer}\n\n" # Add the original options response_message += "*Original Options:*\n" for option, text in self.current_quiz['options'].items(): emoji = "✅" if option == correct_answer else "❌" response_message += f"{emoji} *{option}:* {text}\n" response_message += "\n" response_message += "*Explanations:*\n\n" # Add explanations for all options with proper code formatting for option in ['A', 'B', 'C', 'D']: emoji = "✅" if option == correct_answer else "❌" explanation = self.current_quiz['explanations'][option] # Clean up markdown formatting for better display explanation = self._clean_markdown_for_telegram(explanation) response_message += f"{emoji} *{option}:* {explanation}\n\n" # Remove the inline keyboard from the original message but keep the question visible await query.edit_message_reply_markup(reply_markup=None) # Send the results as a new message to keep the original question visible try: await self.application.bot.send_message( chat_id=self.chat_id, text=response_message, parse_mode='Markdown' ) except Exception as markdown_error: logger.warning(f"Markdown parsing failed: {markdown_error}. Trying without formatting.") # Fallback: send without markdown formatting plain_message = response_message.replace('*', '') await self.application.bot.send_message( chat_id=self.chat_id, text=plain_message ) # Reset quiz state self.quiz_active = False self.current_quiz = None logger.info(f"User answered {selected_answer}, correct answer was {correct_answer}") # Terminate the bot after providing the answer logger.info("Quiz completed. Terminating bot...") self.shutdown_event.set() except Exception as e: logger.error(f"Error handling answer: {e}") def _clean_markdown_for_telegram(self, text: str) -> str: """Clean markdown formatting for better Telegram display""" import re # Replace triple backticks with single backticks for inline code # This works better in Telegram markdown text = re.sub(r'```python\n(.*?)\n```', r'`\1`', text, flags=re.DOTALL) text = re.sub(r'```(.*?)```', r'`\1`', text, flags=re.DOTALL) # Clean up multi-line code blocks by replacing newlines with spaces text = re.sub(r'`([^`]*\n[^`]*)`', lambda m: f'`{m.group(1).replace(chr(10), " ")}`', text) # Only escape characters that are outside of code blocks and really cause issues # Split text into parts: code blocks and regular text parts = [] current_pos = 0 # Find all code blocks (text between backticks) for match in re.finditer(r'`[^`]*`', text): # Add text before code block (escape special chars here) before_code = text[current_pos:match.start()] # Only escape underscore and square brackets which commonly cause issues before_code = before_code.replace('_', '\\_').replace('[', '\\[').replace(']', '\\]') parts.append(before_code) # Add code block as-is (no escaping inside code) parts.append(match.group()) current_pos = match.end() # Add remaining text after last code block remaining_text = text[current_pos:] remaining_text = remaining_text.replace('_', '\\_').replace('[', '\\[').replace(']', '\\]') parts.append(remaining_text) return ''.join(parts) async def run(self): """Run the bot""" try: # Start the bot await self.application.initialize() await self.application.start() # Send initial quiz automatically logger.info("Bot started. Sending initial quiz...") await self.send_quiz() # Start polling await self.application.updater.start_polling() # Wait for shutdown event await self.shutdown_event.wait() # Stop the bot gracefully logger.info("Stopping bot...") await self.application.updater.stop() await self.application.stop() await self.application.shutdown() # Force exit sys.exit(0) except Exception as e: logger.error(f"Error running bot: {e}") sys.exit(1) async def main(): """Main function""" try: bot = TelegramQuizBot() await bot.run() except KeyboardInterrupt: logger.info("Bot stopped by user") except Exception as e: logger.error(f"Unexpected error: {e}") sys.exit(1) if __name__ == "__main__": asyncio.run(main())