import csv import io from enum import Enum from typing import List from fastapi import APIRouter, Depends, Query, UploadFile, HTTPException, status from fastapi.responses import FileResponse, StreamingResponse from fastapi_pagination.ext.sqlalchemy_future import paginate as p from pydantic import BaseModel from sqlmodel import Session, select, col from database.auth.models import User from database.db import get_session from database.exercices.crud import add_tags_db, check_exercice_author, check_private, check_tag_author, create_exo_db, \ delete_exo_db, get_exo_dependency, clone_exo_db, remove_tag_db, serialize_exo, update_exo_db, get_tags_dependency from database.exercices.models import Exercice, ExerciceCreate, ExerciceEdit, ExerciceReadFull, ExercicesTagLink, Tag, \ TagCreate, TagRead, ExerciceRead from generateur.generateur_csv import Csv_generator from services.auth import get_current_user, get_current_user_optional from services.exoValidation import validate_file, validate_file_optionnal from services.io import add_fast_api_root, get_filename_from_path from services.models import Page router = APIRouter(tags=['exercices']) class ExoType(str, Enum): csv = "csv" pdf = "pdf" web = "web" def filter_exo_by_tags(exos: List[tuple[Exercice, str]], tags: List[Tag]): valid_exos = [exo for exo, tag in exos if all( str(t) in tag.split(',') for t in tags)] return valid_exos def queryFilters_dependency(search: str = "", tags: List[str] | None = Depends(get_tags_dependency), type: ExoType | None = Query(default=None)): return search, tags, type @router.post('/exercices', response_model=ExerciceReadFull, status_code=status.HTTP_201_CREATED) def create_exo(exercice: ExerciceCreate = Depends(ExerciceCreate.as_form), file: UploadFile = Depends(validate_file), user: User = Depends(get_current_user), db: Session = Depends(get_session)): file_obj = file['file'].file._file file_obj.name = file['file'].filename exo_obj = create_exo_db(exercice=exercice, user=user, exo_source=file_obj, supports=file['supports'], db=db) return serialize_exo(exo=exo_obj, user_id=user.id, db=db) @router.post('/clone/{id_code}', response_model=ExerciceReadFull) def clone_exo(exercice: Exercice | None = Depends(check_private), user: User = Depends(get_current_user), db: Session = Depends(get_session)): if not exercice: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail={ "Exercice introuvable"}) exo_obj = clone_exo_db(exercice, user, db) if type(exo_obj) == str: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=exo_obj) return serialize_exo(exo=exo_obj, user_id=user.id, db=db) @router.get('/exercices/user', response_model=Page[ExerciceRead | ExerciceReadFull]) def get_user_exercices(user: User = Depends(get_current_user), queryFilters: tuple[str, List[int] | None, ExoType | None] = Depends(queryFilters_dependency), db: Session = Depends(get_session)): search, tags, type = queryFilters statement = select(Exercice) statement = statement.where(Exercice.author_id == user.id) statement = statement.where(Exercice.name.startswith(search)) if type == ExoType.csv: statement = statement.where(Exercice.csv == True) if type == ExoType.pdf: statement = statement.where(Exercice.pdf == True) if type == ExoType.web: statement = statement.where(Exercice.web == True) for t in tags: sub = select(ExercicesTagLink).where(ExercicesTagLink.exercice_id == Exercice.id).where( ExercicesTagLink.tag_id == t).exists() statement = statement.where(sub) statement = statement.order_by(col(Exercice.id).desc()) page = p(db, statement) exercices = page.items page.items = [ serialize_exo(exo=e, user_id=user.id, db=db) for e in exercices] return page @router.get('/exercices/public', response_model=Page[ExerciceRead | ExerciceReadFull]) def get_public_exercices(user: User | None = Depends(get_current_user_optional), queryFilters: tuple[str, List[int] | None] = Depends(queryFilters_dependency), db: Session = Depends(get_session)): search, tags, type = queryFilters if user is not None: statement = select(Exercice) statement = statement.where(Exercice.author_id != user.id) statement = statement.where(Exercice.private == False) statement = statement.where(Exercice.origin_id == None) statement = statement.where(Exercice.name.startswith(search)) if type == ExoType.csv: statement = statement.where(Exercice.csv == True) if type == ExoType.pdf: statement = statement.where(Exercice.pdf == True) if type == ExoType.web: statement = statement.where(Exercice.web == True) for t in tags: sub = select(ExercicesTagLink).where(ExercicesTagLink.exercice_id == Exercice.id).where( ExercicesTagLink.tag_id == t).exists() statement = statement.where(sub) statement = statement.order_by(col(Exercice.id).desc()) page = p(db, statement) exercices = page.items page.items = [ serialize_exo(exo=e, user_id=user.id, db=db) for e in exercices] return page else: statement = select(Exercice) statement = statement.where(Exercice.private == False) statement = statement.where(Exercice.name.startswith(search)) if type == ExoType.csv: statement = statement.where(Exercice.csv == True) if type == ExoType.pdf: statement = statement.where(Exercice.pdf == True) if type == ExoType.web: statement = statement.where(Exercice.web == True) page = p(db, statement) exercices = page.items page.items = [ serialize_exo(exo=e, user_id=None, db=db) for e in exercices] return page @router.get('/exercice/{id_code}', response_model=ExerciceReadFull) def get_exercice(exo: Exercice = Depends(check_private), user: User | None = Depends(get_current_user_optional), db: Session = Depends(get_session)): return serialize_exo(exo=exo, user_id=getattr(user, 'id', None), db=db) @router.put('/exercice/{id_code}', response_model=ExerciceReadFull) def update_exo(file: UploadFile = Depends(validate_file_optionnal), exo: Exercice = Depends(check_exercice_author), exercice: ExerciceEdit = Depends(ExerciceEdit.as_form), db: Session = Depends(get_session)): if exo is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable') if exo == False: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Cet exercice ne vous appartient pas') file_obj = None if file: file_obj = file["file"].file._file file_obj.name = file['file'].filename exo_obj = update_exo_db(exo, exercice, file['supports'] if file is not None else None, file_obj, db) return serialize_exo(exo=exo_obj, user_id=exo_obj.author_id, db=db) @router.delete('/exercice/{id_code}') def delete_exercice(exercice: Exercice | bool | None = Depends(check_exercice_author), db: Session = Depends(get_session)): if exercice is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Exercice introuvable") if exercice == False: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Cet exercice ne vous appartient pas') delete_exo_db(exercice, db) return {'detail': 'Exercice supprimé avec succès'} class NewTags(BaseModel): exo: ExerciceReadFull tags: list[TagRead] @router.post('/exercice/{id_code}/tags', response_model=NewTags, tags=['tags']) def add_tags(tags: List[TagCreate], exo: Exercice | None = Depends(get_exo_dependency), db: Session = Depends(get_session), user: User = Depends(get_current_user)): if exo is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable') exo_obj, new = add_tags_db(exo, tags, user, db) return {"exo": serialize_exo(exo=exo_obj, user_id=user.id, db=db), "tags": new} @router.delete('/exercice/{id_code}/tags/{tag_id}', response_model=ExerciceReadFull, tags=['tags']) def remove_tag(exo: Exercice | None = Depends(get_exo_dependency), tag: Tag | None | bool = Depends(check_tag_author), db: Session = Depends(get_session)): if exo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable') if tag is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='Tag introuvable') if tag == False: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Vous n'êtes pas le propriétaire du tag") exo_obj = remove_tag_db(exo, tag, db) return serialize_exo(exo=exo_obj, user_id=tag.author_id, db=db) @router.get('/exercice/{id_code}/exo_source') async def get_exo_source(exo: Exercice = Depends(check_exercice_author)): if exo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable') if exo == False: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Cet exercice ne vous appartient pas') path = add_fast_api_root(exo.exo_source) filename = get_filename_from_path(path) return FileResponse(path, headers={'content-disposition': 'attachment;filename=' + filename}) @router.get('/tags', response_model=List[TagRead], tags=['tags']) def get_tags(user: User = Depends(get_current_user), db: Session = Depends(get_session)): return user.tags @router.get('/generator/csv/{id_code}') async def generate_csv(*, exo: Exercice | None = Depends(get_exo_dependency), filename: str): if exo.csv is False: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Impossible de générer cet exercice sur dans ce format') source_path = add_fast_api_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')