import os import random import shutil import string from typing import List from config import Exercice_schema, TagIn_schema from services.io import get_abs_path_from_relative_to_root, get_ancestor, get_parent_dir, remove_fastapi_root, remove_if_exists, get_or_create_dir from tortoise import Model from generateur.generateur_main import Generateur from .models import Exercice, Tag async def create_exo_db(*args, **kwargs) -> Exercice: code = await generate_unique_code(Exercice) return await Exercice.create(*args, **{**kwargs, 'id_code': code}) async def delete_exo_db(id_code: str): exo = await Exercice.get(id_code=id_code) path = get_abs_path_from_relative_to_root( exo.exo_source) parent = get_parent_dir(path) shutil.rmtree(parent) return await exo.delete() async def update_exo_db(id_code: str, **kwargs) -> Exercice: exo = await Exercice.get(id_code=id_code) path = get_abs_path_from_relative_to_root(exo.exo_source) remove_if_exists(path) await exo.update_from_dict({**kwargs, 'origin_id': exo.origin_id, 'isOriginal': exo.isOriginal}).save() return exo #flag async def get_or_create_tag(id_code: str, data: List[TagIn_schema]): tag = await Tag.get_or_none(id_code=id_code) if tag == None: code = await generate_unique_code(Tag) return await Tag.create(**{**data,'id_code': code}) return tag async def add_or_remove_tag(exo_id_code: str, tag: Tag): exo = await Exercice.get(id_code = exo_id_code) is_present = await exo.tags.all().get_or_none(id_code=tag.id_code) if is_present == None: await exo.tags.add(tag) else: await exo.tags.remove(tag) return exo async def add_tag_db(id_code: str, tags_data: List[TagIn_schema], user_id:int) -> Exercice: exo = await Exercice.get(id_code = id_code) for t in tags_data: tag = await get_or_create_tag(t.id_code, {**t.dict(exclude_unset=True), 'owner_id': user_id}) await exo.tags.add(tag) return exo async def delete_tag_db(exo_id: str, tag_id: str) -> Exercice: exo = await Exercice.get(id_code=exo_id) tag = await exo.tags.all().get(id_code=tag_id) await exo.tags.remove(tag) return exo def clone_exo_source(path, id_code): upload_root = get_ancestor(path, 2) path = get_abs_path_from_relative_to_root(path) new_path = get_abs_path_from_relative_to_root(os.path.join(upload_root, id_code)) get_or_create_dir((new_path)) return remove_fastapi_root(shutil.copy(path, new_path)) async def clone_exo_db(id_code:str, user_id): exo = await Exercice.get(id_code=id_code) new_id_code = await generate_unique_code(Exercice) exo_obj = await Exercice_schema.from_tortoise_orm(exo) exo_obj = exo_obj.dict(exclude_unset=True) exo_obj.pop('tags') exo_obj.pop('exercices') path = clone_exo_source(exo.exo_source, new_id_code) new_exo = Exercice(**{**exo_obj, 'id_code': new_id_code, 'exo_source': path, "isOriginal": False, 'author_id': user_id}, origin_id=exo.id) await new_exo.save() return new_exo async def get_exo_source_path(id_code: str): exo = await Exercice.get(id_code=id_code) path = get_abs_path_from_relative_to_root(exo.exo_source) return path