102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
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 |