app.py

#  Import necessary libraries from FastAPI for creating the API, handling uploads, forms, and HTTP exceptions


from fastapi import FastAPI, UploadFile, Form, HTTPException
from pydantic import BaseModel
import uvicorn
from fastapi.responses import JSONResponse
from typing import Dict
import hashlib
from openai import OpenAI
from dotenv import load_dotenv
from fastapi.middleware.cors import CORSMiddleware
from firebase_admin import firestore
import json
import re
import pandas as pd
import google.generativeai as genai
from google.generativeai import GenerativeModel
import os
load_dotenv()
client = OpenAI(api_key=os.getenv('DEEPSEEK_API_KEY'), base_url="https://api.deepseek.com",)

# Initialize Gemini LLM

genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
model = genai.GenerativeModel("gemini-2.0-flash")

#  Firebase Admin SDK Initialization
# Get the Firebase credentials (as a JSON string) from environment variables
import firebase_admin
from firebase_admin import credentials
cred_dic = os.getenv("Firebase_cred")


# Parse the JSON string into a Python dictionary
cred_dict = json.loads(cred_dic)


# Initialize the Firebase Admin app with the credentials
# This needs to be done only once
cred = credentials.Certificate(cred_dict)
firebase_admin.initialize_app(cred)


# Create an instance of the FastAPI application
app = FastAPI()


# Add CORSMiddleware to the application
# This allows requests from any origin ("*"), supports credentials,
# allows all HTTP methods ("*"), and allows all headers ("*").
# Be cautious with "*" in production environments; specify origins if possible.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
def generate_df():
    data = [] # Initialize an empty list to store document data
    db = firestore.client() # Get a Firestore client instance
    docs = db.collection("test_results").get()# Retrieve all documents from the "test_results" collection
    for doc in docs:
         doc_data = doc.to_dict()  # Convert Firestore document to a dictionary
         doc_data['id'] = doc.id # Add the document ID to the dictionary
         data.append(doc_data) # Append the document data to the list
    df = pd.DataFrame(data)
    return df

def generate_feedback(email, test_id):

    """
    Generates feedback for a specific test taken by a student.
    It filters the test results by email and test_id, then uses the Gemini model
    to generate constructive feedback based on the student's responses for that test.

    Args:
        email (str): The email of the student.
        test_id (str): The ID of the test.

    Returns:
        str: The generated feedback text, or None if no test results are found.
    """
    df = generate_df()
    df_email = df[df['email'] == email] # Get the DataFrame of all test results
    df_test_id = df_email[df_email['id'] == test_id] # Filter by student's email
    if not df_test_id.empty: # Check if any matching test result was found
        response = df_test_id['responses'].values[0] # Get the 'responses' field from the first (and only) matching row
        # Prepare the prompt for the Gemini model

        
        feedback = model.generate_content(f"""You are an experienced tutor analyzing a student's test responses to provide constructive feedback. Below is the student's test history in JSON format. Your task is to:

Identify Strengths: Highlight areas where the student performed well, demonstrating a strong understanding of the concepts.

Identify Weaknesses: Point out areas where the student struggled or made consistent errors, indicating gaps in understanding.

Provide Actionable Suggestions: Offer specific advice on how the student can improve their performance in future tests.

Encourage and Motivate: End with positive reinforcement to keep the student motivated.
Test History:{str(response)} """)
        return feedback.text  # Return the text part of the response
    else:
        print("No test results found for this id")
def generate_overall_feedback(email):
    """
    Generates overall feedback for a student based on all their test results.
    It filters test results by email and uses the Gemini model to provide
    a holistic view of the student's performance.

    Args:
        email (str): The email of the student.

    Returns:
        str: The generated overall feedback text, or None if no test results are found.
    """
    
    df = generate_df()
    df_email = df[df['email'] == email]
    if not df_email.empty:
        response = df_email['responses'].values 
        feedback = model.generate_content(f"""You are an experienced tutor analyzing a student's test responses to provide constructive feedback. Below is the student's test history in list format. Your task is to:
                                          Identify Strengths: Highlight areas where the student performed well, demonstrating a strong understanding of the concepts.

Identify Weaknesses: Point out areas where the student struggled or made consistent errors, indicating gaps in understanding.

Provide Actionable Suggestions: Offer specific advice on how the student can improve their performance in future tests.

Encourage and Motivate: End with positive reinforcement to keep the student motivated.

Test History:{str(response)} """)
        return feedback.text
    else:
        print("Please try again with a valid email")
    




@app.post("/get_single_feedback")
async def get_single_feedback(email: str, test_id: str):

    """
    API endpoint to get feedback for a single test.
    Expects 'email' and 'test_id' as form data.
    """

    
    feedback = generate_feedback(email, test_id)
    return JSONResponse(content={"feedback": feedback})

@app.post("/get_overall_feedback")
async def get_overall_feedback(email: str):

    #  """
    # API endpoint to get overall feedback for a student.
    # Expects 'email' as form data.
    # """
    
    feedback = generate_overall_feedback(email)  # Call the helper function to generate overall feedback
    return JSONResponse(content={"feedback": feedback})

@app.post("/get_strong_weak_topics")
async def get_strong_weak_topics(email: str):

    # """
    # API endpoint to identify strong and weak topics for a student.
    # Requires at least 10 test attempts. Uses DeepSeek API for analysis.
    # Expects 'email' as form data.
    # """

    
    df = generate_df()
    df_email = df[df['email'] == email]

    # Check if the student has attempted at least 10 tests
    if len(df_email)<10:

        # The original condition was `len(df)>=10`, which seems to check the total number of tests in the DB,
    # not for the specific user. Changed to `len(df_email) >= 10`.
        return JSONResponse(content={"message": "Please attempt atleast 10 tests to enable this feature"})

    elif len(df)>=10:
         # Get responses from the latest 10 tests (or all if fewer than 10, but the check above ensures at least 10)
        response = df_email['responses'].values[:10]
        # Assuming response is a list of responses
        formatted_data = str(response)  # Convert response to a string format suitable for the API call
        section_info = {
            'filename': 'student_performance',
            'schema': {
                'weak_topics': ['Topic#1', 'Topic#2', '...'],
                'strong_topics': ['Topic#1', 'Topic#2', '...']
            }
        }
        
        # Generate response using the client
        completion = client.chat.completions.create(
            model="deepseek-chat",
            response_format={"type": "json_object"},
            messages=[
                {
                    "role": "system",
                    "content": f"""You are an Educational Performance Analyst focusing on {section_info['filename'].replace('_', ' ')}.
                    Analyze the provided student responses to identify and categorize topics into 'weak' and 'strong' based on their performance. Try to give
                    high level topics like algebra, trignometry, geometry etc in your response.
                    Do not add any explanations, introduction, or comments - return ONLY valid JSON.
                    """
                },
                {
                    "role": "user",
                    "content": f"""
                    Here is the raw data for {section_info['filename']}:
                    
                    {formatted_data}
                    
                    Convert this data into JSON that matches this schema:
                    {json.dumps(section_info['schema'], indent=2)}
                    """
                }
            ],
            temperature=0.0  # Set temperature to 0 for deterministic output
        )
        
        # Extract the JSON content from the completion object
        strong_weak_topics = completion.choices[0].message.content  # Access the content attribute directly
        
        return JSONResponse(content=json.loads(strong_weak_topics))
    else:
        return JSONResponse(content={"error": "No test results found for this email"})

@app.post("/generate_flashcards")
async def generate_flashcards(email: str):

     """
    API endpoint to generate flashcards for a student's weak topics.
    Requires at least 10 test attempts.
    First, it identifies weak topics using DeepSeek.
    Then, it generates flashcards for these topics using Gemini.
    Expects 'email' as form data.
    """
    
    df = generate_df() # Get all test results
    df_email = df[df['email'] == email] 

    if len(df_email) < 10:

        return JSONResponse(content={"message": "Please attempt at least 10 tests to enable flashcard generation."})
        # Step 1: Get the weak topics via DeepSeek
    # Get responses from the latest 10 tests
    response = df_email['responses'].values[:10]
    formatted_data = str(response)

    schema = {
        'weak_topics': ['Topic#1', 'Topic#2', '...'],
        'strong_topics': ['Topic#1', 'Topic#2', '...']
    }

    completion = client.chat.completions.create(
        model="deepseek-chat",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": f"""You are an Educational Performance Analyst focusing on student performance.
                Analyze the provided student responses to identify and categorize topics into 'weak' and 'strong' based on their performance.
                Do not add any explanations - return ONLY valid JSON."""
            },
            {
                "role": "user",
                "content": f"""
                Here is the raw data:
                {formatted_data}

                Convert this data into JSON that matches this schema:
                {json.dumps(schema, indent=2)}
                """
            }
        ],
        temperature=0.0
    )

    # Extract weak topics
    strong_weak_json = json.loads(completion.choices[0].message.content)
    weak_topics = strong_weak_json.get("weak_topics", [])

    if not weak_topics:
        return JSONResponse(content={"message": "Could not extract weak topics."})

     # Step 2: Generate flashcards using Gemini for the identified weak topics
    topic_str = ", ".join(weak_topics)
    flashcard_prompt = f"""Create 5 concise, simple, straightforward and distinct Anki cards to study the following topic, each with a front and back.
Avoid repeating the content in the front on the back of the card. Avoid explicitly referring to the author or the article.
Use the following format:
Front: [front section of card 1]
Back: [back section of card 1]
...
The topics: {topic_str}
"""

    flashcard_response = model.generate_content(flashcard_prompt)

    # Step 3: Parse Gemini response into JSON format
    flashcards_raw = flashcard_response.text.strip()
    flashcard_pattern = re.findall(r"Front:\s*(.*?)\nBack:\s*(.*?)(?=\nFront:|\Z)", flashcards_raw, re.DOTALL)
 # Use regex to find all "Front:" and "Back:" pairs
    flashcards = [{"Front": front.strip(), "Back": back.strip()} for front, back in flashcard_pattern]

    return JSONResponse(content=flashcards)

@app.post("/generate_detailed_summary")
async def generate_detailed_summary(email: str):
    """
    API endpoint to generate detailed summaries for a student's weak topics.
    Requires at least 10 test attempts.
    First, it identifies weak topics using DeepSeek.
    Then, it generates summaries for these topics using Gemini.
    Expects 'email' as form data.
    """

    
    df = generate_df()
    df_email = df[df['email'] == email]

    if len(df_email) < 10:
        return JSONResponse(content={"message": "Please attempt at least 10 tests to enable detailed summary generation."})

    # Step 1: Get the weak topics via DeepSeek
    response = df_email['responses'].values[:10]
    formatted_data = str(response)

    schema = {
        'weak_topics': ['Topic#1', 'Topic#2', '...'],
        'strong_topics': ['Topic#1', 'Topic#2', '...']
    }

    completion = client.chat.completions.create(
        model="deepseek-chat",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": f"""You are an Educational Performance Analyst focusing on student performance.
                Analyze the provided student responses to identify and categorize topics into 'weak' and 'strong' based on their performance.
                Do not add any explanations - return ONLY valid JSON."""
            },
            {
                "role": "user",
                "content": f"""
                Here is the raw data:
                {formatted_data}

                Convert this data into JSON that matches this schema:
                {json.dumps(schema, indent=2)}
                """
            }
        ],
        temperature=0.0
    )

    # Extract weak topics
    strong_weak_json = json.loads(completion.choices[0].message.content)
    weak_topics = strong_weak_json.get("weak_topics", [])

    if not weak_topics:
        return JSONResponse(content={"message": "Could not extract weak topics."})

    # Step 2: Generate flashcards using Gemini
    topic_str = ", ".join(weak_topics)
#     flashcard_prompt = f"""Create 5 concise, simple, straightforward and distinct Anki cards to study the following topic, each with a front and back.
# Avoid repeating the content in the front on the back of the card. Avoid explicitly referring to the author or the article.
# Use the following format:
# Front: [front section of card 1]
# Back: [back section of card 1]
# ...
# The topics: {topic_str}
# """

#     flashcard_response = model.generate_content(flashcard_prompt)

#     # Step 3: Parse Gemini response into JSON format
#     flashcards_raw = flashcard_response.text.strip()
#     flashcard_pattern = re.findall(r"Front:\s*(.*?)\nBack:\s*(.*?)(?=\nFront:|\Z)", flashcards_raw, re.DOTALL)

#     flashcards = [{"Front": front.strip(), "Back": back.strip()} for front, back in flashcard_pattern]

    summarization_prompt = f"""
Write an informative and concise summary (approximately 200 words) for each of the following topics. 
Do not mention the author or source. Use clear and academic language.

List of topics:
{topic_str}

Use the following format:
Topic: [topic name]
Summary: [200-word summary]
"""

    summary_response = model.generate_content(summarization_prompt)
    
    # Step 3: Parse response into JSON format
    summary_raw = summary_response.text.strip()
    summary_pattern = re.findall(r"Topic:\s*(.*?)\nSummary:\s*(.*?)(?=\nTopic:|\Z)", summary_raw, re.DOTALL)
    
    summaries = [{topic.strip(): summary.strip() for topic, summary in summary_pattern}]

    return JSONResponse(content=summaries)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)