generateur_v3/backend/api_old/apis/exercices/route_exercices.py

import csv
import io
import os
import sys
from typing import List
from fastapi import APIRouter, Depends, Form, UploadFile, status
from api.schemas.exercices import ExerciceSchema
from database.exercices.validators import get_support_compatibility_for_exo_source_from_data
from database.auth.models import UserModel
from generateur.generateur_csv import Csv_generator
from database.decorators import as_form
from database.exercices.models import Exercice
from fastapi.exceptions import HTTPException
from database.exercices.crud import add_tag_db, create_exo_db, delete_tag_db, get_exo_source_path, update_exo_db, delete_exo_db, clone_exo_db
from config import Exercice_schema, ExerciceIn_form, ExerciceIn_schema, Exo_schema, TagIn_schema, User_schema
from services.auth import check_author_exo,  get_current_clientId, get_current_user, jwt_optional, jwt_required
from services.io import get_abs_path_from_relative_to_root, get_filename_from_path
from fastapi.responses import FileResponse, Response, StreamingResponse
from pydantic import BaseModel
from fastapi_jwt_auth import AuthJWT
from fastapi_pagination import paginate, Page
router = APIRouter()

# Exercices


@router.get("/exercices", response_model=Page[Exo_schema])
async def get_exercices():
    exo_list = await Exo_schema.from_queryset(Exercice.all())
    return paginate(exo_list)
 

@router.get('/exercices/user', response_model=Page[Exo_schema])
async def get_exercices(Authorize: AuthJWT=Depends(jwt_required)):
    username = Authorize.get_jwt_subject()
    user = await UserModel.get(username=username)
    
    exo_list = await Exo_schema.from_queryset(Exercice.filter(author_id=user.id))
    return paginate(exo_list)

@router.get('/exercices/public', response_model=Page[Exo_schema])
async def get_exercices(Authorize: AuthJWT=Depends(jwt_optional)):
    username = Authorize.get_jwt_subject()
    is_authenticated = username != None
    if is_authenticated:
        user = await UserModel.get(username=username)
        exo_list = Exercice.filter(author_id__not = user.id)
        return paginate(exo_list)
    exo_list = await Exo_schema.from_queryset(Exercice.all())
    return paginate(exo_list)


@router.get('/exercice/{id_code}', response_model=ExerciceSchema)
async def get_exercice(id_code: str, Authorize: AuthJWT= Depends(jwt_optional)):
    username = Authorize.get_jwt_subject()
    is_authenticated = username != None
    
    exo = await Exercice.get(id_code=id_code)
    if is_authenticated:
        user = await UserModel.get(username=username)
        author = await exo.author
        is_author = author.id == user.id
        exo_obj = await Exercice_schema.from_tortoise_orm(exo)
        exo_dict = exo_obj.dict()
        exo_dict['is_author'] = is_author
        return exo_dict
    
    exo_obj = await Exercice_schema.from_tortoise_orm(exo)
    exo_dict =exo_obj.dict()
    exo_dict['is_author'] = False
    return exo_dict


async def validate_file(file: UploadFile):
    data = await file.read()
    try:
        exo_supports_compatibility = get_support_compatibility_for_exo_source_from_data(
            data)
        if not exo_supports_compatibility['isPdf'] and not exo_supports_compatibility['isCsv'] and not exo_supports_compatibility['isWeb']:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"exo_source":
                                                                                          '[Error] : Exercice non valide (compatible avec aucun support)'})

    except Exception as e:
        msg = e.args[0]
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"exo_source": msg})
    await file.seek(0)
    return file

@router.post("/exercices", response_model=Exercice_schema)
async def create_exercice(file: UploadFile = Depends(validate_file), exo: ExerciceIn_schema = Depends(ExerciceIn_form.as_form), current_user: UserModel = Depends(get_current_user)):
    file_obj = file.file._file
    file_obj.name = file.filename
    exo_obj = await create_exo_db(**{**exo.dict(exclude_unset=True)}, exo_source=file_obj, author_id=current_user.id)
    return await Exercice_schema.from_tortoise_orm(exo_obj)


@router.delete("/exercices/{id_code}", response_model=str)
async def delete_exercice(id_code: str, author: User_schema = Depends(check_author_exo)):
    await delete_exo_db(id_code)
    return "success"


@router.put("/exercices/{id_code}", response_model=Exercice_schema)
async def update_exercice(id_code: str, file: UploadFile, exo: ExerciceIn_form = Depends(ExerciceIn_form.as_form), author: User_schema = Depends(check_author_exo)):
    file_obj = file.file._file
    file_obj.name = file.filename
    exo_obj = await update_exo_db(id_code, **{**exo.dict(exclude_unset=True)}, exo_source=file_obj)
    return await Exercice_schema.from_tortoise_orm(exo_obj)


@router.post('/exercices/{id_code}/clone', response_model=Exercice_schema)
async def clone_exercice(id_code: str, user: User_schema = Depends(get_current_user)):
    exo_obj = await clone_exo_db(id_code, user.id)
    return await Exercice_schema.from_tortoise_orm(exo_obj)


@router.get('/exercices/{id_code}/exo_source')
async def get_exo_source(id_code: str, author: User_schema = Depends(check_author_exo)):
    path = await get_exo_source_path(id_code)
    filename = get_filename_from_path(path)
    return FileResponse(path, filename=filename)

# Tags


@router.post('/exercices/{id_code}/tags', response_model=Exercice_schema)
async def update_tag(id_code: str, tags_data: List[TagIn_schema],  current_user: User_schema = Depends(get_current_user)):
    exo_obj = await add_tag_db(id_code, tags_data, current_user.id)
    return await Exercice_schema.from_tortoise_orm(exo_obj)


@router.delete('/exercices/{exo_id}/tags/{tags_id}', response_model=Exercice_schema)
async def remove_tag(exo_id: str, tag_id: str, owner: User_schema = Depends(check_author_exo)):
    exo_obj = await delete_tag_db(exo_id, tag_id)
    return await Exercice_schema.from_tortoise_orm(exo_obj)


@router.get('/generator/csv/{exo_id}')
async def generate_csv(exo_id: str, filename: str):
    exo = await Exercice.get(id_code=exo_id)

    if exo.csvSupport == False:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                            detail='Impossible de générer cet exercice sur le support csv')

    source_path = get_abs_path_from_relative_to_root(exo.exo_source)
    consigne = exo.consigne

    buffer = io.StringIO()
    writer = csv.writer(buffer, delimiter=',',
                        quotechar=',', quoting=csv.QUOTE_MINIMAL, dialect='excel')  # mettre | comme sep un jour
    Csv_generator(source_path, 10, 10, 12, consigne, writer)

    return StreamingResponse(iter([buffer.getvalue()]), headers={"Content-Disposition": f'attachment;filename="{filename}'}, media_type='text/csv')


class ExoOption(BaseModel):
    id: str
    nbInExo: int
    nbOfExo: int


@router.post('/generator/pdf')
async def generate_pdf(exos_list: List[ExoOption]):
    return