Constantine, [11/17/2024 11:25 AM]
import os
import re
import hashlib
import uuid
import logging
import asyncio
from functools import lru_cache
from io import BytesIO
from dotenv import load_dotenv
from collections import defaultdict

import aiohttp
import openai
import sympy as sp
import matplotlib.pyplot as plt
from cachetools import TTLCache, cached

from telegram import (
    Update,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    InputMediaPhoto
)
from telegram.ext import (
    ApplicationBuilder,
    CommandHandler,
    ContextTypes,
    CallbackQueryHandler,
    MessageHandler,
    filters
)

# Firebase Imports
import firebase_admin
from firebase_admin import credentials, firestore, storage

# LangChain Imports (if needed for advanced retrieval)
# from langchain.embeddings import OpenAIEmbeddings
# from langchain.vectorstores import FAISS
# from langchain.chains import RetrievalQA
# from langchain.llms import OpenAI

# Load Environment Variables
load_dotenv()

# Configuration
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
FIREBASE_SERVICE_ACCOUNT_PATH = os.getenv('FIREBASE_SERVICE_ACCOUNT_PATH')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
GOOGLE_API_KEY = os.getenv('GOOGLE_API_KEY')
GOOGLE_CX = os.getenv('GOOGLE_CX')
YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY')

# Validate Configuration
missing_configs = [
    var for var in [
        'TELEGRAM_BOT_TOKEN',
        'FIREBASE_SERVICE_ACCOUNT_PATH',
        'OPENAI_API_KEY',
        'GOOGLE_API_KEY',
        'GOOGLE_CX',
        'YOUTUBE_API_KEY'
    ] if not os.getenv(var)
]
if missing_configs:
    raise ValueError(f"The following environment variables are missing: {', '.join(missing_configs)}")

# Initialize OpenAI
openai.api_key = OPENAI_API_KEY

# Initialize Firebase
cred = credentials.Certificate(FIREBASE_SERVICE_ACCOUNT_PATH)
firebase_admin.initialize_app(cred, {
    'storageBucket': 'your-firebase-storage-bucket-url'  # Replace with your actual Firebase Storage bucket URL
})
db = firestore.client()
bucket = storage.bucket()

# Logging Configuration
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(name)

# Define JAMB Subjects
JAMB_SUBJECTS = [
    "Biology",
    "Chemistry",
    "Economics",
    "English Language",
    "Government",
    "Literature in English",
    "Mathematics",
    "Physics",
    "Geography",
    "Commerce",
    "Accounting",
    "Agriculture",
    # Add more subjects as needed
]

# Initialize Cache (TTLCache: maxsize, ttl in seconds)
cache = TTLCache(maxsize=1024, ttl=300)  # Cache up to 1024 items for 5 minutes

# Decorator for caching Firestore fetches
def cache_fetch(subject, year, question_id):
    """
    Generates a cache key for question details.
    """
    return f"{subject}:{year}:{question_id}"

@cached(cache, key=lambda subject, year, question_id: cache_fetch(subject, year, question_id))
async def fetch_question_details(subject: str, year: str, question_id: str) -> dict:
    """
    Retrieves the details of a specific question from Firestore.
    Caches the result to minimize database reads.
    """
    try:
        question_ref = db.collection('jamb_questions') \
                        .document(subject) \
                        .collection('years') \
                        .document(year) \
                        .collection('questions') \
                        .document(question_id)
        doc = question_ref.get()
        if doc.exists:
            return doc.to_dict()
        else:
            logger.warning(f"Question ID '{question_id}' not found under {subject} {year}.")
            return {}
    except Exception as e:
        logger.error(f"Error fetching details for question ID '{question_id}': {e}")
        return {}

async def fetch_available_years(subject: str) -> list:

Constantine, [11/17/2024 11:25 AM]
"""
    Fetches available years for a given subject from Firestore.
    """
    try:
        years_ref = db.collection('jamb_questions').document(subject).collection('years')
        docs = years_ref.stream()
        years = [doc.id for doc in docs]
        return sorted(years, reverse=True)  # Descending order
    except Exception as e:
        logger.error(f"Error fetching years for subject '{subject}': {e}")
        return []

async def fetch_questions(subject: str, year: str) -> list:
    """
    Fetches available questions for a given subject and year from Firestore.
    """
    try:
        questions_ref = db.collection('jamb_questions') \
                          .document(subject) \
                          .collection('years') \
                          .document(year) \
                          .collection('questions')
        docs = questions_ref.stream()
        questions = [{'id': doc.id, 'question': doc.to_dict().get('question', 'No question text')} for doc in docs]
        return questions
    except Exception as e:
        logger.error(f"Error fetching questions for subject '{subject}' and year '{year}': {e}")
        return []

async def store_interaction(user_id: int, command: str, response: str, sources: list = None):
    """
    Stores user interaction in Firestore.
    """
    try:
        user_ref = db.collection('users').document(str(user_id))
        interactions_ref = user_ref.collection('interactions')
        interactions_ref.add({
            'command': command,
            'response': response,
            'sources': sources or [],
            'timestamp': firestore.SERVER_TIMESTAMP
        })
        logger.info(f"Stored interaction for user {user_id}: {command}")
    except Exception as e:
        logger.error(f"Error storing interaction for user {user_id}: {e}")

# Helper Functions
def detect_diagram_in_question(question: str) -> bool:
    """
    Determines if the question involves a diagram based on keywords.
    """
    diagram_keywords = [
        r"\bdraw\b",
        r"\bdiagram\b",
        r"\bgraph\b",
        r"\bfigure\b",
        r"\billustrate\b",
        r"\bsketch\b",
        r"\bchart\b",
        r"\bshow me\b"
    ]
    pattern = re.compile("|".join(diagram_keywords), re.IGNORECASE)
    return bool(pattern.search(question))

def extract_function_from_question(question: str) -> str:
    """
    Extracts the mathematical function from the question.
    """
    try:
        match = re.search(r"derivative of\s+(.*)\?", question, re.IGNORECASE)
        if match:
            return match.group(1).strip()
        match = re.search(r"plot\s+the function\s+(.*)", question, re.IGNORECASE)
        if match:
            return match.group(1).strip()
        return ""
    except Exception as e:
        logger.error(f"Error extracting function from question '{question}': {e}")
        return ""

def generate_prompt(question: str) -> str:
    """
    Generates a refined prompt for OpenAI based on the user's question.
    """
    prompt = (
        f"You are an academic assistant specializing in JAMB exam questions.\n"
        f"Provide a clear and concise answer to the following question.\n\n"
        f"Question: {question}\n\n"
        f"Answer:"
    )
    return prompt

def generate_circuit_diagram(topic: str) -> BytesIO:
    """
    Generates a circuit diagram based on the given topic.
    """
    try:
        if "series circuit" in topic.lower():
            fig, ax = plt.subplots(figsize=(6, 4))
            # Placeholder for actual circuit diagram drawing logic
            ax.text(0.5, 0.5, 'Series Circuit Diagram', horizontalalignment='center', verticalalignment='center')
            ax.axis('off')
            
            buf = BytesIO()
            plt.savefig(buf, format='png')
            plt.close(fig)
            buf.seek(0)
            return buf
        else:
            return None
    except Exception as e:
        logger.error(f"Error generating circuit diagram for topic '{topic}': {e}")
        return None

Constantine, [11/17/2024 11:25 AM]
def generate_calculus_diagram(function: str) -> BytesIO:
    """
    Generates a plot for a given mathematical function using SymPy and Matplotlib.
    """
    try:
        x = sp.symbols('x')
        func = sp.sympify(function)
        f_lambdified = sp.lambdify(x, func, 'numpy')
        
        x_vals = np.linspace(-10, 10, 400)
        y_vals = f_lambdified(x_vals)
        
        plt.figure(figsize=(6, 4))
        plt.plot(x_vals, y_vals, label=f'f(x) = {function}')
        plt.title(f'Plot of f(x) = {function}')
        plt.xlabel('x')
        plt.ylabel('f(x)')
        plt.legend()
        plt.grid(True)
        
        buf = BytesIO()
        plt.savefig(buf, format='png')
        plt.close()
        buf.seek(0)
        return buf
    except Exception as e:
        logger.error(f"Error generating calculus diagram for function '{function}': {e}")
        return None

async def save_diagram(diagram_buf: BytesIO) -> str:
    """
    Saves the diagram image to Firebase Storage and returns its public URL.
    """
    try:
        unique_id = str(uuid.uuid4())
        blob = bucket.blob(f"diagrams/{unique_id}.png")
        
        # Upload the diagram
        blob.upload_from_file(diagram_buf, content_type='image/png')
        
        # Make the blob publicly accessible
        blob.make_public()
        
        return blob.public_url
    except Exception as e:
        logger.error(f"Error saving diagram to Firebase Storage: {e}")
        return None

@cached(cache, key=lambda subject, year, question_id: generate_query_hash(subject, year, question_id))
async def get_question_details(subject: str, year: str, question_id: str) -> dict:
    """
    Wrapper to fetch question details with caching.
    """
    return await fetch_question_details(subject, year, question_id)

# Inline Keyboard Generators
def generate_subject_keyboard() -> InlineKeyboardMarkup:
    """
    Generates an inline keyboard with JAMB subjects.
    """
    keyboard = []
    # Organize subjects into rows of 2 buttons each for better UI
    row = []
    for idx, subject in enumerate(JAMB_SUBJECTS, start=1):
        row.append(InlineKeyboardButton(subject, callback_data=f"subject|{subject}"))
        if idx % 2 == 0:
            keyboard.append(row)
            row = []
    if row:
        keyboard.append(row)
    return InlineKeyboardMarkup(keyboard)

def generate_year_keyboard(years: list, subject: str) -> InlineKeyboardMarkup:
    """
    Generates an inline keyboard with available years for the selected subject.
    """
    keyboard = []
    for year in years:
        keyboard.append([InlineKeyboardButton(year, callback_data=f"year|{subject}|{year}")])
    return InlineKeyboardMarkup(keyboard)

def generate_question_keyboard(questions: list, subject: str, year: str) -> InlineKeyboardMarkup:
    """
    Generates an inline keyboard with available questions for the selected subject and year.
    """
    keyboard = []
    # Organize questions into rows of 3 buttons each
    row = []
    for idx, question in enumerate(questions, start=1):
        row.append(InlineKeyboardButton(f"Q{idx}", callback_data=f"question|{subject}|{year}|{question['id']}"))
        if idx % 3 == 0:
            keyboard.append(row)
            row = []
    if row:
        keyboard.append(row)
    return InlineKeyboardMarkup(keyboard)

def generate_get_answer_keyboard(subject: str, year: str, question_id: str) -> InlineKeyboardMarkup:
    """
    Generates an inline keyboard button to get the answer.
    """
    keyboard = [
        [InlineKeyboardButton("🔍 Get Answer", callback_data=f"get_answer|{subject}|{year}|{question_id}")]
    ]
    return InlineKeyboardMarkup(keyboard)

def generate_post_answer_keyboard() -> InlineKeyboardMarkup:
    """
    Generates an inline keyboard with buttons for sources, videos, and related questions.

Constantine, [11/17/2024 11:25 AM]
"""
    keyboard = [
        [
            InlineKeyboardButton("📄 Sources", callback_data='sources'),
            InlineKeyboardButton("🎥 Videos", callback_data='videos')
        ],
        [
            InlineKeyboardButton("🔗 Related Questions", callback_data='related_questions')
        ]
    ]
    return InlineKeyboardMarkup(keyboard)

# Command Handlers

async def jamb_command(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the /jamb command. Prompts the user to select a subject.
    """
    user = update.effective_user
    user_id = user.id

    # Update user interaction in Firestore
    await store_interaction(user_id, "/jamb", "Initiated /jamb command")

    welcome_text = "📚 *JAMB Past Questions*\n\nSelect a subject from the list below to begin:"
    await update.message.reply_text(welcome_text, parse_mode='Markdown', reply_markup=generate_subject_keyboard())

async def subject_selection_callback(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles subject selection. Prompts the user to select a year.
    """
    query = update.callback_query
    await query.answer()

    data = query.data  # Expected format: "subject|Subject Name"
    _, subject = data.split("|", 1)

    user_id = query.from_user.id

    # Fetch available years for the selected subject from Firestore
    years = await fetch_available_years(subject)

    if not years:
        await query.edit_message_text("🔎 *No available years found for this subject.*", parse_mode='Markdown')
        return

    message = f"📅 *Select the year for {subject}:*"
    await query.edit_message_text(message, parse_mode='Markdown', reply_markup=generate_year_keyboard(years, subject))

async def year_selection_callback(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles year selection. Prompts the user to select a question.
    """
    query = update.callback_query
    await query.answer()

    data = query.data  # Expected format: "year|Subject Name|Year"
    _, subject, year = data.split("|", 2)

    user_id = query.from_user.id

    # Fetch available questions for the selected subject and year from Firestore
    questions = await fetch_questions(subject, year)

    if not questions:
        await query.edit_message_text("❌ *No questions found for this selection.*", parse_mode='Markdown')
        return

    message = f"📝 *Select a question from {year} for {subject}:*"
    await query.edit_message_text(message, parse_mode='Markdown', reply_markup=generate_question_keyboard(questions, subject, year))

async def question_selection_callback(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles question selection. Retrieves and presents the selected question along with diagrams if any.
    """
    query = update.callback_query
    await query.answer()

    data = query.data  # Expected format: "question|Subject Name|Year|Question ID"
    _, subject, year, question_id = data.split("|", 3)

    user_id = query.from_user.id

    # Retrieve the specific question from Firestore
    question_doc = await get_question_details(subject, year, question_id)

    if not question_doc:
        await query.edit_message_text("❌ *Failed to retrieve the question.*", parse_mode='Markdown')
        return

    # Structured response
    response_text = (
        f"📚 *Subject:* {subject}\n"
        f"📅 *Year:* {year}\n"
        f"📝 *Question Number:* {question_id}\n\n"
        f"*Question:*\n{question_doc.get('question', 'No question text found.')}"
    )

    # Check if the question involves a diagram
    if question_doc.get('involves_diagram', False):
        diagram_urls = question_doc.get('diagram_urls', [])
        if diagram_urls:
            await query.message.reply_text(response_text, parse_mode='Markdown')

Constantine, [11/17/2024 11:25 AM]
for url in diagram_urls:
                await query.message.reply_photo(photo=url, caption="📄 Diagram")
        else:
            await query.message.reply_text(response_text, parse_mode='Markdown')
    else:
        await query.message.reply_text(response_text, parse_mode='Markdown')

    # Provide an 'Get Answer' button
    keyboard = generate_get_answer_keyboard(subject, year, question_id)
    await query.message.reply_text("Would you like to get the answer?", reply_markup=keyboard)

async def get_answer_callback(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the 'Get Answer' button. Retrieves and presents the answer to the selected question.
    """
    query = update.callback_query
    await query.answer()

    data = query.data  # Expected format: "get_answer|Subject Name|Year|Question ID"
    _, subject, year, question_id = data.split("|", 3)

    user_id = query.from_user.id

    # Retrieve the specific question's answer from Firestore
    question_doc = await get_question_details(subject, year, question_id)

    if not question_doc:
        await query.edit_message_text("❌ *Failed to retrieve the answer.*", parse_mode='Markdown')
        return

    # Structured answer response
    answer_text = (
        f"✅ *Answer:*\n{question_doc.get('answer', 'No answer provided.')}"
    )

    await query.message.reply_text(answer_text, parse_mode='Markdown')

    # Optionally, provide additional options after the answer
    keyboard = generate_post_answer_keyboard()
    await query.message.reply_text("What would you like to do next?", reply_markup=keyboard)

async def ask_command(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the /ask command. Sends the user's query to GPT-4 and returns the response.
    Usage: /ask Your question here
    """
    user = update.effective_user
    user_id = user.id
    query = ' '.join(context.args)
    
    if not query:
        await update.message.reply_text(
            "❌ Please enter a question after /ask.\n\nExample: /ask What is the capital of France?"
        )
        return

    # Update user interaction in Firestore
    await store_interaction(user_id, f"/ask {query}", "Processing /ask command")

    await update.message.reply_text("🔍 Generating your answer using GPT-4...")

    prompt = generate_prompt(query)
    
    try:
        response = openai.Completion.create(
            engine="gpt-4",
            prompt=prompt,
            max_tokens=500,
            temperature=0.2,
            n=1,
            stop=None,
        )
        answer = response.choices[0].text.strip()
        
        # Store interaction in Firestore
        await store_interaction(user_id, f"/ask {query}", answer)
    
        await update.message.reply_text(answer, parse_mode='Markdown')
    except Exception as e:
        logger.error(f"Error in /ask command: {e}")
        await update.message.reply_text("🔴 An error occurred while processing your request. Please try again later.")

async def search_command(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the /search command. Performs a Google search and uses OpenAI to synthesize the results.
    Usage: /search Your search query here
    """
    user = update.effective_user
    user_id = user.id
    query = ' '.join(context.args)
    
    if not query:
        await update.message.reply_text(
            "❌ Please enter a search query after /search.\n\nExample: /search JAMB registration process"
        )
        return

    # Update user interaction in Firestore
    await store_interaction(user_id, f"/search {query}", "Processing /search command")

    await update.message.reply_text("🔍 Performing Google search...")

    # Perform asynchronous image search as an example
    search_results = await google_image_search_async(query, num_images=3)
    
    if not search_results:
        await update.message.reply_text("🔎 No results found for your query.")
        return

Constantine, [11/17/2024 11:25 AM]
# Synthesize the search results using OpenAI
    summary_prompt = f"You have performed a Google search for the following query:\n\nQuery: {query}\n\nThe top results are:\n"
    for idx, link in enumerate(search_results, start=1):
        summary_prompt += f"{idx}. {link}\n"
    summary_prompt += "\nProvide a concise summary based on the above links."

    try:
        summary_response = openai.Completion.create(
            engine="gpt-4",
            prompt=summary_prompt,
            max_tokens=300,
            temperature=0.3,
            n=1,
            stop=None,
        )
        summary = summary_response.choices[0].text.strip()
        
        # Store interaction in Firestore
        await store_interaction(user_id, f"/search {query}", summary, sources=search_results)
    
        # Send the synthesized summary
        await update.message.reply_text(f"📄 *Summary of Search Results:*\n{summary}", parse_mode='Markdown')
        
        # Optionally, send the top images
        for idx, img_url in enumerate(search_results, start=1):
            await update.message.reply_photo(photo=img_url, caption=f"🎥 Search Result {idx}")
    except Exception as e:
        logger.error(f"Error in /search command: {e}")
        await update.message.reply_text("🔴 An error occurred while processing your search. Please try again later.")

async def help_command(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the /help command. Provides guidance on how to use the bot.
    """
    help_text = (
        "📚 *Welcome to the JAMB Assistant Bot!*\n\n"
        "Here are the commands you can use:\n\n"
        "/jamb - Navigate through JAMB past questions by subject and year.\n"
        "/ask [your question] - Ask any question and receive an answer powered by GPT-4.\n"
        "/search [your query] - Perform a Google search and get a synthesized summary of the results.\n"
        "/help - Show this help message.\n\n"
        "🔍 *Examples:*\n"
        "/jamb\n"
        "/ask What is the capital of Nigeria?\n"
        "/search Best JAMB preparation resources\n\n"
        "If you have any issues or suggestions, feel free to reach out!"
    )
    await update.message.reply_text(help_text, parse_mode='Markdown')

# Callback Query Handlers for /jamb
async def get_answer_callback(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Handles the 'Get Answer' button. Retrieves and presents the answer to the selected question.
    """
    query = update.callback_query
    await query.answer()

    data = query.data  # Expected format: "get_answer|Subject Name|Year|Question ID"
    _, subject, year, question_id = data.split("|", 3)

    user_id = query.from_user.id

    # Retrieve the specific question's answer from Firestore
    question_doc = await get_question_details(subject, year, question_id)

    if not question_doc:
        await query.edit_message_text("❌ *Failed to retrieve the answer.*", parse_mode='Markdown')
        return

    # Structured answer response
    answer_text = (
        f"✅ *Answer:*\n{question_doc.get('answer', 'No answer provided.')}"
    )

    await query.message.reply_text(answer_text, parse_mode='Markdown')

    # Optionally, provide additional options after the answer
    keyboard = generate_post_answer_keyboard()
    await query.message.reply_text("What would you like to do next?", reply_markup=keyboard)

# Echo Handler for Unrecognized Messages
async def echo(update: Update, context: ContextTypes.DEFAULT.TYPE) -> None:
    """
    Replies to unrecognized messages.
    """
    user = update.effective_user
    user_id = user.id
    message_text = update.message.text

    # Store interaction in Firestore
    await store_interaction(user_id, "unknown", message_text)

    response = "🤖 Sorry, I didn't understand that command. Please use /jamb, /ask, /search, or /help to interact with the bot."
    await update.message.reply_text(response)

Constantine, [11/17/2024 11:25 AM]
# Main Function
def main():
    """
    Initializes and starts the Telegram bot application.
    """
    application = (
        ApplicationBuilder()
        .token(TELEGRAM_BOT_TOKEN)
        .concurrent_updates(True)  # Enable concurrent updates for better performance
        .build()
    )
    
    # Register Command Handlers
    application.add_handler(CommandHandler("jamb", jamb_command))
    application.add_handler(CommandHandler("ask", ask_command))
    application.add_handler(CommandHandler("search", search_command))
    application.add_handler(CommandHandler("help", help_command))
    
    # Register Callback Query Handlers with specific patterns
    application.add_handler(CallbackQueryHandler(subject_selection_callback, pattern=r"^subject\|"))
    application.add_handler(CallbackQueryHandler(year_selection_callback, pattern=r"^year\|"))
    application.add_handler(CallbackQueryHandler(question_selection_callback, pattern=r"^question\|"))
    application.add_handler(CallbackQueryHandler(get_answer_callback, pattern=r"^get_answer\|"))
    
    # Register Echo Handler for Unrecognized Messages
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    
    # Start the Bot
    try:
        logger.info("Bot is starting...")
        application.run_polling()
    except Exception as e:
        logger.error(f"Error running bot: {e}", exc_info=True)

if name == "main":
    main()

give me the breakdown on how to implement this code as a coding beginner and deploy and test the project