EditionPN/app/forms.py
2021-06-14 18:44:53 +02:00

265 lines
12 KiB
Python

from flask_wtf import FlaskForm
from flask_wtf.file import FileAllowed
from wtforms import StringField, SubmitField, FileField, RadioField, widgets, FieldList, FormField, HiddenField
from wtforms.validators import DataRequired, Optional
from wtforms_alchemy import model_form_factory, ClassMap
from wtforms_alchemy.fields import QuerySelectMultipleField
from app import db
import app.models as models
import yaml
import json
import os
import re
REPERTOIRE_EXPORT = "./export/"
if not os.path.exists(REPERTOIRE_EXPORT):
os.makedirs(REPERTOIRE_EXPORT)
categorie_liste = ["saes","ressources","acs","ues","semestres","competences"]
categorie_to_model = {"saes": "SAE", "ressources": "Ressource", "acs": "AC", "coefsaes": "CoefSAE", "coefressources": "CoefRessource", "pns": "PN", "semestres": "Semestre", "competences": "Competence"}
separateur = None
BaseModelForm = model_form_factory(FlaskForm, include_primary_keys=True)
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, models.Semestre):
return obj.num
if isinstance(obj, models.Competence) or isinstance(obj, models.SAE) or isinstance(obj, models.Ressource) or isinstance(obj, models.AC):
return obj.code
return json.JSONEncoder.default(self, obj)
class ModelForm(BaseModelForm):
@classmethod
def get_session(self):
return db.session
class RefListField(QuerySelectMultipleField):
option_widget=widgets.CheckboxInput()
class AccueilForm(FlaskForm):
ajouter = SubmitField("Ajouter")
reset = SubmitField("Reset")
exporterJSON = SubmitField("ExporterJSON")
exporterYAML = SubmitField("ExporterYAML")
fichier = FileField("Choisir fichier", validators=[FileAllowed(["yml","json"], "Fichier YAML/JSON seulement!")])
importer = SubmitField("Importer")
def ListRef(self):
result = {"semestres": [], "competences": [], "acs": [], "pns": [], "saes": [], "coefsaes": [], "ressources": [], "coefressources": []}
associations = {"semestres_competences": models.Semestres_Competences, "acs_competences": models.ACs_Competences, "ressources_acs": models.Ressources_ACs, "ressources_saes": models.Ressources_SAEs, "saes_acs": models.SAEs_ACs}
for key in result.keys():
model = getattr(models, categorie_to_model[key])
for ref in model.query.all():
result[key].append(ref.export())
for assoName, association in associations.items():
result[assoName] = db.session.query(association).all()
return result
def exportJSON(self):
result = self.ListRef()
fichier = REPERTOIRE_EXPORT + "referentiels" + ".json"
with open(fichier, "w", encoding="utf8") as fid:
json.dump(result, fid, cls=CustomEncoder, indent=4)
def exportYAML(self):
result = self.ListRef()
fichier = REPERTOIRE_EXPORT + "referentiels" + ".yml"
with open(fichier, "w", encoding="utf8") as fid:
yaml.dump(yaml.safe_load(json.dumps(result, cls=CustomEncoder)), fid, indent=4, sort_keys=False)
def importRef(self):
fichier = self.fichier.data
filename, extension = fichier.filename.split(".")
if extension == "yml":
filedata = yaml.safe_load(fichier.read())
elif extension == "json":
filedata = json.load(fichier)
else:
self.fichier.errors.append("Une erreur a été produite.")
return
for categorie, liste in filedata.items():
if "_" not in categorie:
model = getattr(models, categorie_to_model[categorie])
for ref in liste:
print("Adding:", ref)
if categorie == "semestres":
semestre = model.query.filter_by(num=ref["num"]).first()
if not semestre:
ref["ues"] = []
db.session.add(model(**ref))
else:
print("Not added: Already in database")
elif "coef" in categorie:
competence = models.Competence.query.filter_by(code=ref["competence"]).first()
if categorie == "coefsaes":
sae = models.SAE.query.filter_by(code=ref["sae"]).first()
if not models.CoefSAE.query.filter_by(sae=sae,competence=competence).first():
db.session.add(models.CoefSAE(sae=sae,competence=competence,coef=ref["coef"]))
else:
print("Not added: Already in database")
elif categorie == "coefressources":
ressource = models.Ressource.query.filter_by(code=ref["ressource"]).first()
if not models.CoefRessource.query.filter_by(ressource=ressource, competence=competence).first():
db.session.add(models.CoefRessource(ressource=ressource, competence=competence,coef=ref["coef"]))
else:
print("Not added: Already in database")
else:
refBDD = model.query.filter_by(code=ref["code"]).first()
if not refBDD:
for categorieRef, valeur in ref.items():
if categorieRef in categorie_liste:
ref[categorieRef] = []
db.session.add(model(**ref))
else:
print("Not added: Already in database")
else:
for assoc in liste:
print("Adding association:", assoc)
firstModelname, secondModelname = categorie.split("_")
firstVal, secondVal = assoc
firstModel = getattr(models, categorie_to_model[firstModelname])
secondModel = getattr(models, categorie_to_model[secondModelname])
if firstModelname == "semestres":
firstRef = firstModel.query.filter_by(num=firstVal).first()
secondRef = secondModel.query.filter_by(code=secondVal).first()
if secondRef not in firstRef.ues: firstRef.ues.append(secondRef)
else: print("Not added: Association already exist")
else:
firstRef = firstModel.query.filter_by(code=firstVal).first()
secondRef = secondModel.query.filter_by(code=secondVal).first()
if secondRef not in firstRef.__dict__[secondModelname]: firstRef.__dict__[secondModelname].append(secondRef)
else: print("Not added: Association already exist")
db.session.commit()
class CoefForm(FlaskForm):
objetformation = HiddenField("Objet de formation")
coef = StringField("Coef")
class UEForm(FlaskForm):
ue = HiddenField("Competence")
acs = RefListField("Liste des ACs", query_factory=lambda: models.AC.query.all(), get_label=lambda ref: ref.getLabel())
ressources = RefListField("Liste des Ressources inclus dans les ACs", query_factory=lambda: [], get_label=lambda ref: ref.getLabel())
saes = RefListField("Liste des SAEs inclus dans les ACs", query_factory=lambda: [], get_label=lambda ref: ref.getLabel())
coef = FieldList(FormField(CoefForm))
class SemestreForm(FlaskForm):
ues = RefListField("Liste des UEs", query_factory=lambda: models.Competence.query.all(), get_label=lambda ref: ref.getLabel())
ueform = FieldList(FormField(UEForm))
update = SubmitField("Update")
class Form(ModelForm):
class Meta:
not_null_validator_type_map = ClassMap({db.String: [DataRequired()]})
sauvegarder = SubmitField("Sauvegarder")
charger = SubmitField("Charger")
supprimer = SubmitField("Supprimer")
exporter = SubmitField("Exporter")
fichier = FileField("Choisir fichier", validators=[FileAllowed(["yml"], "Fichier Yaml seulement!")])
importer = SubmitField("Importer")
referentiel = RadioField("Liste des référentiels", validators=[Optional()])
def chargerRef(self):
if self.referentiel.data == None:
self.referentiel.errors.append("Aucun référentiel n'a été selectionné!")
return False
else:
return True
def chargerBDD(self, referentiel):
for categorie in list(self.data.keys())[7:-1]:
self[categorie].process_data(referentiel.__dict__[categorie])
self.referentiel.process_data(referentiel)
def importerRef(self):
if len(self.fichier.errors) == 0:
if self.fichier.data.filename == "":
self.fichier.errors.append("Aucun fichier selectioné.")
return
if re.match(self.regex, self.fichier.data.filename[:-4]) == None:
self.fichier.errors.append("Mauvais type de référentiel!")
return
fichier_Yaml = yaml.safe_load(self.fichier.data.read())
for categorie, valeur in fichier_Yaml.items():
if categorie in categorie_liste:
model = getattr(models, categorie_to_model[categorie])
self[categorie].process_data([model.query.filter_by(code=ref).first() for ref in valeur])
else:
self[categorie].process_data(valeur)
self.validate()
def exporterRef(self):
""" Exporte dans un fichier yaml les informations du formulaire rempli """
output = {}
for categorie, valeur in list(self.data.items())[7:-1]:
if categorie in categorie_liste:
output[categorie] = [ referentiel.code for referentiel in valeur ]
else:
output[categorie] = valeur
fichier = REPERTOIRE_EXPORT + self.code.data + ".yml"
with open(fichier, "w", encoding="utf8") as fid:
fid.write(yaml.dump(output))
def supprimerRef(self):
if self.referentiel.data == None:
self.referentiel.erros.append("Aucun référentiel n'a été selectionné!")
return False
else:
temp = self.referentiel.data[1:-1].split()
model = getattr(models, temp[0])
referentiel = model.query.filter_by(code=temp[1]).first()
db.session.delete(referentiel)
db.session.commit()
return True
def sauvegarderRef(self):
model = getattr(models, self.__class__.__name__[:-4])
referentiel = model.query.filter_by(code=self.code.data).first()
if referentiel == None: referentiel = model()
self.populate_obj(referentiel)
db.session.add(referentiel)
db.session.commit()
class PNForm(Form):
regex = "^PN\d$"
class Meta:
model = models.PN
type = RadioField("Type de parcours", choices=["1","2","3"])
class ACForm(Form):
regex = "^AC\d{4}$"
class Meta:
model = models.AC
saes = RefListField("Liste des SAEs", query_factory=lambda: models.SAE.query.all(), get_label=lambda ref: ref.getLabel())
ressources = RefListField("Liste des Ressources", query_factory=lambda: models.Ressource.query.all(), get_label=lambda ref: ref.getLabel())
class SAEForm(Form):
regex = "^SAE\d{2}$"
class Meta:
model = models.SAE
acs = RefListField("Liste des ACs", query_factory=lambda: models.AC.query.all(), get_label=lambda ref: ref.getLabel())
ressources = RefListField("Liste des Ressources", query_factory=lambda: models.Ressource.query.all(), get_label=lambda ref: ref.getLabel())
class RessourceForm(Form):
regex = "^R\d{3}$"
class Meta:
model = models.Ressource
acs = RefListField("Liste des ACs", query_factory=lambda: models.AC.query.all(), get_label=lambda ref: ref.getLabel())
saes = RefListField("Liste des SAEs", query_factory=lambda: models.SAE.query.all(), get_label=lambda ref: ref.getLabel())
class CompetenceForm(Form):
regex = "^RT\d$"
class Meta:
model = models.Competence