main.py

from fastapi import FastAPI, HTTPException, Header, Depends, Request
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.exceptions import RequestValidationError
from typing import Optional, List
from pydantic import BaseModel, ValidationError
import pandas as pd
import datetime

api = FastAPI()

# Charger les données à partir du fichier CSV
questions_data = pd.read_csv('questions.csv')

# Dictionnaire des identifiants des utilisateurs
users_credentials = {
    "alice": "wonderland",
    "bob": "builder",
    "clementine": "mandarine",
    "admin": "4dm1N"  # Ajout de l'utilisateur admin
}

# Modèle Pydantic pour représenter une question
class Question(BaseModel):
    question: str
    subject: str
    correct: Optional[str] = None # Champ optionnel
    use: str
    responseA: str
    responseB: str
    responseC: Optional[str] = None # Champ optionnel
    responseD: Optional[str] = None # Champ optionnel
    # remark: Optional[str]  = None # Champ optionnel

# Modèle pour représenter une exception personnalisée
class MyException(Exception):
    def __init__(self,
                 status_code: int,
                 name : str,
                 message : str):
        self.status_code = status_code
        self.name = name
        self.message = message
        self.date = str(datetime.datetime.now())

# Gestionnaire d'exception personnalisé
@api.exception_handler(MyException)
def MyExceptionHandler(
    request: Request,
    exception: MyException
    ):
    return JSONResponse(
        status_code=exception.status_code,
        content={
            'url': str(request.url),
            'name': exception.name,
            'message': exception.message,
            'date': exception.date
        }
    )

# Gestionnaire d'exception pour les erreurs de validation de la requête
@api.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            'url': str(request.url),
            'name': "Erreur de validation de la requête (parametre requis)",
            'message': exc.errors(),
            'date': str(datetime.datetime.now())
        },
    )

# Gestionnaire d'exception pour les erreurs Pydantic
@api.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content={
            'url': str(request.url),
            'name': "Erreur de validation Pydantic",
            'message': exc.errors(),
            'date': str(datetime.datetime.now())
        },
    )

# Fonction pour vérifier l'authentification de l'utilisateur
def authenticate(authorisation: str = Header(None)):
    if not authorisation:
        raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
    try:
        scheme, credentials = authorisation.split()
        if scheme != 'Basic':
            raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
        username, password = credentials.split(":")
        if username not in users_credentials or users_credentials[username] != password:
            raise HTTPException(status_code=401, detail="Utilisateur non authorisé ")
    except Exception as e:
        raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
    return username, password

# Endpoint pour vérifier que l'API est fonctionnelle
@api.get('/', name="Vérification que l'API fonctionne")
def check_api():
    return {'message': "L'API fonctionne"}

# Endpoint pour récupérer les questions en fonction du type de test (use) et des catégories (subject) spécifiés
@api.get('/questions', name="Récupération des questions pour un QCM")
def get_questions(use: str, 
                  subject: str, 
                  num_questions: int, 
                  authorisation: str  = Header(None)): 
    """
    Récupère les questions en fonction du type de test (use) et des catégories (subject) spécifiés
    L'application peut produire des QCMs de 5, 10 ou 20 questions (seulement)
    Les questions sont retournées dans un ordre aléatoire
    Seuls les utilisateurs se trouvant dans users_credentials peuvent utiliser cette application
    """

    # Verifier si l'utilsateur existe et a le droit
    authenticate(authorisation)
    
    # Verifier si le nombre de questions demandé correspond au nombre de questions d'un QCM
    if num_questions not in [5,10,20]:
        raise MyException(status_code=422, name="num_questions invalide", \
            message="La requête peut contenir 51,10 ou 20 questions, mais pas "+str(num_questions))
        
    # Filtrer les questions en fonction des paramètres spécifiés
    filtered_questions = questions_data
    if use:
        filtered_questions = filtered_questions[filtered_questions['use'] == use]
    if subject is not None:
        s = subject.split(',')
        filtered_questions = filtered_questions[filtered_questions['subject'].isin(s)]
    print("len(filtered_questions)=",len(filtered_questions))   
    print("num_questions=",num_questions) 
    # Vérifier si des questions sont disponibles dans la catégorie spécifiée
    if len(filtered_questions) == 0:
        raise MyException(status_code=404, name="Aucune question", \
            message="Aucune question ne correspond aux critères sélectionnés")
    # Verifier si le nombre de questions diponibles >= au nombre requis
    elif (len(filtered_questions) < num_questions):
        raise MyException(status_code=400, name="Nb insuffisant de questions ", \
            message="Le nombre de questions correspondantes aux critères sélectionnés est < au nombre requis")

    # Supprimer les valeurs NaN dans les champs correct,responseC, responseD (ils ne sont pas toujours remplis)
    filtered_questions['correct'] = filtered_questions['correct'].apply(lambda x: None if pd.isna(x) else x)
    filtered_questions['responseC'] = filtered_questions['responseC'].apply(lambda x: None if pd.isna(x) else x)
    filtered_questions['responseD'] = filtered_questions['responseD'].apply(lambda x: None if pd.isna(x) else x)

    # Sélectionner un nombre aléatoire de questions
    selected_questions = filtered_questions.sample(n=min(num_questions, len(filtered_questions)))

    # Convertir les données en liste de dictionnaires
    questions_list = selected_questions.to_dict(orient='records')

    # Convertir les dictionnaires en objets Pydantic de type Question
    questions_objects = [Question(**question) for question in questions_list]
    return questions_objects


# Endpoint pour créer une nouvelle question (accessible uniquement par l'utilisateur admin)
@api.post('/questions/create', name="Création d'une nouvelle question")
def create_question(question: Question, 
                    authorisation: str = Header(None)):
    """
    Crée une nouvelle question et l'ajoute à questions.csv
    Seuls l' utilisateur admin a le droit d'utiliser cette fonction
    """
    global questions_data
    
    username, password = authenticate(authorisation)
    if username != 'admin':
        raise HTTPException(status_code=401, detail="Utilisateur non authorisé")

    # Ajouter la nouvelle question au DataFrame
    new_question = question.model_dump()
    questions_data = questions_data.append(new_question, ignore_index=True)

    # Sauvegarder les modifications dans le fichier CSV
    questions_data.to_csv('questions.csv', index=False)
    
    return {'message': 'Question créée avec succès'}