""" Ce fichier permet de gerer les unités d'enseignement (UE) Une unité d'enseignement est definit par : - Code (valeur unique & clé) - Titre - Description - prerequis - Objectif - Programme - Méthode pédagogique - Méthode d'évaluation - Support - Durée (heures, jours, semaines, mois) Rattachement Unité d’enseignement (UE) & formation : Lors du rattachement d’une UE à une formation, il faudra définir : - Le crédit associé (numérique) - Présentiel/ distantiel/ hybride, - Est_note (oui / non) pour savoir cette unité doit être notée - Groupe_evaluation : Dans une formation, on peut rassembler plusieurs UE dans un groupe d’UE """ import bson import pymongo from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import prj_common as mycommon import secrets import inspect import sys, os import csv import pandas as pd from pymongo import ReturnDocument import GlobalVariable as MYSY_GV from math import isnan import GlobalVariable as MYSY_GV import ela_index_bdd_classes as eibdd import email_mgt as email import jinja2 from flask import send_file from xhtml2pdf import pisa from email.message import EmailMessage from email.mime.text import MIMEText from email import encoders import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders """ Ajout d'une unité d'enseignement (UE) """ def Add_Unite_Enseignement(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'code', 'titre', 'description', 'objectif', 'prerequis', 'programme', 'methode_pedagogique', 'methode_evaluation', 'support', 'duration', 'duration_unite', 'domain_id', 'bloc' , 'ects', 'seuil_validation', 'niveau_competence', 'vh_cm', 'vh_tp', 'vh_tpg', 'vh_td', 'semestre', 'note_ue_calculation_rule_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'code', 'titre',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner mydata = {} mydata = diction del mydata['token'] # Initialisation des champs non envoyés à vide for val in field_list: if val not in diction.keys(): mydata[str(val)] = "" # Verifier qu'une UE avec le meme code n'existe pas déjà pour ce partenaire is_ue_exist_count = MYSY_GV.dbname['unite_enseignement'].count_documents({'partner_owner_recid':my_partner['recid'], 'valide':'1', 'code':diction['code']}) if( is_ue_exist_count > 0 ): mycommon.myprint( str(inspect.stack()[0][3]) + " Il existe déjà une UE avec le même code ") return False, " Il existe déjà une UE avec le même code " """ Verifier que l'unite de durée est valide """ if( "duration_unite" in diction.keys() and diction['duration_unite']): if( diction['duration_unite'] not in MYSY_GV.CLASS_DURATION_UNIT): mycommon.myprint( str(inspect.stack()[0][3]) + " L'unité de durée est invalide ") return False, " L'unité de durée est invalide " """ Verifier que la durée est bien un float """ if ("duration" in diction.keys() and diction['duration']): local_status, local_retval = mycommon.IsFloat(str(diction['duration'])) if( local_status is False ): mycommon.myprint( str(inspect.stack()[0][3]) + " La durée est invalide ") return False, " La durée est invalide " if ("note_ue_calculation_rule_id" in diction.keys() and diction['note_ue_calculation_rule_id']): is_valide_calculation_rule_id_count = MYSY_GV.dbname['base_calcul_note_class_and_ue'].count_documents( {'_id': ObjectId(str(diction['note_ue_calculation_rule_id'])), 'valide': '1', 'locked': '0'}) if (is_valide_calculation_rule_id_count != 1): mycommon.myprint(str( inspect.stack()[0][ 3]) + " L'identifiant de la règle de calcul de la note de l'UE est invalide ") return False, " L'identifiant de la règle de calcul de la note de l'UE est invalide " mydata['date_update'] = str(datetime.now()) mydata['update_by'] = str(my_partner['_id']) mydata['partner_owner_recid'] = str(my_partner['recid']) mydata['valide'] = "1" mydata['locked'] = "0" del mydata['token'] inserted_id = MYSY_GV.dbname['unite_enseignement'].insert_one(mydata).inserted_id if (not inserted_id): mycommon.myprint(" Impossible de créer l'unité d'enseignement (2) ") return False, " Impossible de créer l'unité d'enseignement (2) " return True, " L'unité d'enseignement a été correctement ajoutée " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de créer l'UE " """ Mise à jour d'une unité d'enseignement """ def Update_Unite_Enseignement(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', '_id', 'code', 'titre', 'description', 'objectif', 'prerequis', 'programme', 'methode_pedagogique', 'methode_evaluation', 'support', 'duration', 'duration_unite', 'domain_id', 'bloc', 'ects', 'seuil_validation', 'niveau_competence', 'vh_cm', 'vh_tp', 'vh_tpg', 'vh_td', 'semestre', 'note_ue_calculation_rule_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', '_id' ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] ue_id = "" if ("_id" in diction.keys()): if diction['_id']: ue_id = diction['_id'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner # Verifier que l'UE existe et est valide is_valide_opport = MYSY_GV.dbname['unite_enseignement'].count_documents({'_id':ObjectId(str(ue_id)), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( is_valide_opport != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'UE est invalide ") return False, " L'identifiant de l'UE est invalide " if ("note_ue_calculation_rule_id" in diction.keys() and diction['note_ue_calculation_rule_id']): is_valide_calculation_rule_id_count = MYSY_GV.dbname['base_calcul_note_class_and_ue'].count_documents( {'_id': ObjectId(str(diction['note_ue_calculation_rule_id'])), 'valide': '1', 'locked': '0'}) if (is_valide_calculation_rule_id_count != 1): mycommon.myprint(str( inspect.stack()[0][ 3]) + " L'identifiant de la règle de calcul de la note de l'UE est invalide ") return False, " L'identifiant de la règle de calcul de la note de l'UE est invalide " """ Si on veut modidier le code, verifier que le nouveau code n'est pas deja prix """ if( "code" in diction.keys() ): is_new_code_exist_with_orther_id = MYSY_GV.dbname['unite_enseignement'].find_one({'code':str(diction['code']), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( is_new_code_exist_with_orther_id and str(is_new_code_exist_with_orther_id['_id']) != ue_id ): mycommon.myprint( str(inspect.stack()[0][3]) + " Le code de l'UE '"+str(diction['code'])+" est déjà utilisé ") return False, " Le code de l'UE '"+str(diction['code'])+" est déjà utilisé " mydata = {} mydata = diction del mydata['token'] del mydata['_id'] mydata['date_update'] = str(datetime.now()) mydata['update_by'] = str(my_partner['_id']) result = MYSY_GV.dbname['unite_enseignement'].find_one_and_update( {'_id':ObjectId(str(ue_id)), 'partner_owner_recid':str(my_partner['recid'])}, {"$set": mydata}, upsert=False, return_document=ReturnDocument.AFTER ) if ("_id" not in result.keys()): mycommon.myprint( " Impossible de mettre à jour l'UE (2) ") return False, " Impossible de mettre à jour l'UE (2) " return True, " L'UE a été correctement mise à jour " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de mettre à jour l'UE " """ Recuperer la liste des UE, sans filtres """ def Get_List_Unite_Enseignement_no_filter(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token' ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0'} for New_retVal in MYSY_GV.dbname['unite_enseignement'].find(qry).sort([("_id", pymongo.DESCENDING), ]): user = New_retVal user['id'] = str(val_tmp) val_tmp = val_tmp + 1 if (str(New_retVal['duration_unite']) == "heure"): user['duration_concat'] = str(New_retVal['duration']) + " h" elif (str(New_retVal['duration_unite']) == "jour"): user['duration_concat'] = str(New_retVal['duration']) + " j" elif (str(New_retVal['duration_unite']) == "semaine"): user['duration_concat'] = str(New_retVal['duration']) + " s" elif (str(New_retVal['duration_unite']) == "mois"): user['duration_concat'] = str(New_retVal['duration']) + " m" elif (str(New_retVal['duration_unite']) == "annee"): user['duration_concat'] = str(New_retVal['duration']) + " a" else: user['duration_concat'] = str(New_retVal['duration']) + " ?" if( "domain_id" not in user.keys() ): user['domain_id'] = "" if ("bloc" not in user.keys()): user['bloc'] = "" """ Recuperer le nombre de lignes de planification """ count_nb_planif_lines = MYSY_GV.dbname['unite_enseignement_planif'].count_documents( {'ue_id': str(New_retVal['_id']), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) user['planification_line'] = count_nb_planif_lines count_nb_planif_lines_uses = MYSY_GV.dbname['unite_enseignement_planif'].count_documents( {'ue_id': str(New_retVal['_id']), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid']), 'session_formation_sequence_id': {'$exists': True, '$ne': ""}}) user['planification_used_line'] = count_nb_planif_lines_uses RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des unites d'enseignement " """ Recuperer la liste des UE, avec filtres """ def Get_List_Unite_Enseignement_with_filter(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'code', 'titre'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str(inspect.stack()[0][ 3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Toutes les informations fournies ne sont pas valables" """ Verification des champs obligatoires """ field_list_obligatoire = ['token',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner filt_code = {} if ("code" in diction.keys()): filt_code = {'code': {'$regex': str(diction['code']), "$options": "i"}} filt_titre = {} if ("titre" in diction.keys()): filt_titre = {'titre': {'$regex': str(diction['titre']), "$options": "i"}} query = [{'$match': { '$and': [filt_code, filt_titre, {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0'} ]}}, {'$sort': {'_id': -1}}, ] RetObject = [] val_tmp = 0 for New_retVal in MYSY_GV.dbname['unite_enseignement'].aggregate(query): user = New_retVal user['id'] = str(val_tmp) val_tmp = val_tmp + 1 if (str(New_retVal['duration_unite']) == "heure"): user['duration_concat'] = str(New_retVal['duration']) + " h" elif (str(New_retVal['duration_unite']) == "jour"): user['duration_concat'] = str(New_retVal['duration']) + " j" elif (str(New_retVal['duration_unite']) == "semaine"): user['duration_concat'] = str(New_retVal['duration']) + " s" elif (str(New_retVal['duration_unite']) == "mois"): user['duration_concat'] = str(New_retVal['duration']) + " m" elif (str(New_retVal['duration_unite']) == "annee"): user['duration_concat'] = str(New_retVal['duration']) + " a" else: user['duration_concat'] = str(New_retVal['duration']) + " ?" if ("domain_id" not in user.keys()): user['domain_id'] = "" if ("bloc" not in user.keys()): user['bloc'] = "" """ Recuperer le nombre de lignes de planification """ count_nb_planif_lines = MYSY_GV.dbname['unite_enseignement_planif'].count_documents({'ue_id':str(New_retVal['_id']), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) user['planification_line'] = count_nb_planif_lines count_nb_planif_lines_uses = MYSY_GV.dbname['unite_enseignement_planif'].count_documents( {'ue_id': str(New_retVal['_id']), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid']), 'session_formation_sequence_id': { '$exists' : True, '$ne': "" }}) user['planification_used_line'] = count_nb_planif_lines_uses RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des unites d'enseignement " """ Suppression d'une UE Si elle n'est pas utilisée regles : on ne peut supprimer une UE que si elle n'est pas utilisée dans une formation """ def Delete_Given_Unite_Enseignement(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', '_id', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', '_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner # Verifier l'existance de l'ue is_ue_valide = MYSY_GV.dbname['unite_enseignement'].count_documents({'_id':ObjectId(str(diction['_id'])), 'partner_owner_recid':str(my_partner['recid'])}) if (is_ue_valide != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'UE est invalide ") return False, " L'identifiant de l'UE est invalide " """ Verifier si cette formation n'est pas utilisée dans une formation """ is_ue_used_in_class_count = MYSY_GV.dbname['myclass'].count_documents({'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'list_unite_enseignement._id':str(diction['_id'])}) if( is_ue_used_in_class_count > 0 ): mycommon.myprint( str(inspect.stack()[0][3]) + " Cette UE est actuellement utilisée dans "+str(is_ue_used_in_class_count)+" formation. Impossible de la supprimer ") return False, " Cette UE est actuellement utilisée dans "+str(is_ue_used_in_class_count)+" formation. Impossible de la supprimer " qry = {'_id':ObjectId(str(diction['_id'])), 'partner_owner_recid':str(my_partner['recid'])} ret_del_competence = MYSY_GV.dbname['unite_enseignement'].delete_many(qry) """ On va ensuite supprimer les eventuelles planification de cette UE """ ret_del_ue_planif = MYSY_GV.dbname['unite_enseignement_planif'].delete_many({'ue_id':str(diction['_id'])}) return True, "L'UE a été correctement supprimée" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de supprimer l'UE " """ Recuperer une UE données """ def Get_Given_Unite_Enseignement(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', '_id' ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token','_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(diction['_id']))} for New_retVal in MYSY_GV.dbname['unite_enseignement'].find(qry).sort([("_id", pymongo.DESCENDING), ]): user = New_retVal user['id'] = str(val_tmp) val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les données de l'unite d'enseignement " """ Recuperer la liste des UE d'une formation donnée """ def Get_List_Unite_Enseignement_Of_Given_Class(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'class_id', 'class_internal_url'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token',] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 is_qry_data = "0" if( "class_id" in diction.keys() and diction['class_id']) : is_qry_data = "1" if ("class_internal_url" in diction.keys() and diction['class_internal_url']): is_qry_data = "1" if( is_qry_data == "0" ): # L'utilisateur n'a fournir aucun element pour faire la requete. on retourne du vide dans ce cas. user = [] RetObject.append(mycommon.JSONEncoder().encode(user)) # print(" #### RetObject = ", RetObject) return True, RetObject if( "class_id" in diction.keys() and diction['class_id']): qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(diction['class_id']))} elif ( "class_internal_url" in diction.keys() and diction['class_internal_url'] ): qry = {"partner_owner_recid": str(my_partner['recid']), 'valide': '1', 'locked': '0', 'internal_url': str(diction['class_internal_url'])} #print(" #### Get_List_Unite_Enseignement_Of_Given_Class qry = ", qry) for New_retVal in MYSY_GV.dbname['myclass'].find(qry).sort([("_id", pymongo.DESCENDING), ]): if( "list_unite_enseignement" in New_retVal.keys() ): for local_val in New_retVal['list_unite_enseignement'] : eu_class_coef = "" if( "coefficient" in local_val.keys() and local_val['coefficient']): eu_class_coef = local_val['coefficient'] eu_class_seuil_validation = "" if ("seuil_validation" in local_val.keys() and local_val['seuil_validation']): eu_class_seuil_validation = local_val['seuil_validation'] user = local_val user['id'] = str(val_tmp) val_tmp = val_tmp + 1 user['class_id'] = str(New_retVal['_id']) user['internal_url'] = str(New_retVal['internal_url']) """ Recuperation des données associées à cette UE """ ue_data = MYSY_GV.dbname['unite_enseignement'].find_one({'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(local_val['_id']))}) if( ue_data ): for val in ue_data.keys() : user[str(val)] = ue_data[str(val)] if (str(ue_data['duration_unite']) == "heure"): user['duration_concat'] = str(ue_data['duration']) + " h" elif (str(ue_data['duration_unite']) == "jour"): user['duration_concat'] = str(ue_data['duration']) + " j" elif (str(ue_data['duration_unite']) == "semaine"): user['duration_concat'] = str(ue_data['duration']) + " s" elif (str(ue_data['duration_unite']) == "mois"): user['duration_concat'] = str(ue_data['duration']) + " m" elif (str(ue_data['duration_unite']) == "annee"): user['duration_concat'] = str(ue_data['duration']) + " a" else: user['duration_concat'] = str(ue_data['duration']) + " ?" # On ecrase les données si il y a les meme données sur la formation if (eu_class_coef ): user['coefficient'] = str(eu_class_coef) if (eu_class_seuil_validation): user['seuil_validation'] = str(eu_class_seuil_validation) # Si il n'y a pas les infos dans les 2 cas, alors on met une valeur par defaut if( "coefficient" not in user.keys()): user['coefficient'] = "1" if ("seuil_validation" not in user.keys()): user['seuil_validation'] = "99" RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des unites d'enseignement de la formation " """ Recuperer la liste des UE d'une formation donnée en partant de la session_id """ def Get_List_Unite_Enseignement_Of_Given_Class_From_Session_Id(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'session_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'session_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 """ Recuperer la class_id depuis la session_id """ session_data = MYSY_GV.dbname['session_formation'].find_one({'_id':ObjectId(str(diction['session_id'])), 'valide':'1', "partner_owner_recid": str(my_partner['recid'])}) if( session_data is None or "class_internal_url" not in session_data.keys() ): mycommon.myprint( str(inspect.stack()[0][3]) + " La session n'a pas de 'class_internal_url' ") return True, RetObject class_internal_url = session_data['class_internal_url'] qry = {"partner_owner_recid": str(my_partner['recid']), 'valide': '1', 'locked': '0', 'internal_url': str(class_internal_url)} #print(" #### Get_List_Unite_Enseignement_Of_Given_Class qry = ", qry) for New_retVal in MYSY_GV.dbname['myclass'].find(qry).sort([("_id", pymongo.DESCENDING), ]): if( "list_unite_enseignement" in New_retVal.keys() ): for local_val in New_retVal['list_unite_enseignement'] : eu_class_coef = "" if( "coefficient" in local_val.keys() and local_val['coefficient']): eu_class_coef = local_val['coefficient'] eu_class_seuil_validation = "" if ("seuil_validation" in local_val.keys() and local_val['seuil_validation']): eu_class_seuil_validation = local_val['seuil_validation'] user = local_val user['id'] = str(val_tmp) val_tmp = val_tmp + 1 user['class_id'] = str(New_retVal['_id']) user['internal_url'] = str(New_retVal['internal_url']) """ Recuperation des données associées à cette UE """ ue_data = MYSY_GV.dbname['unite_enseignement'].find_one({'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(local_val['_id']))}) if( ue_data ): for val in ue_data.keys() : user[str(val)] = ue_data[str(val)] if (str(ue_data['duration_unite']) == "heure"): user['duration_concat'] = str(ue_data['duration']) + " h" elif (str(ue_data['duration_unite']) == "jour"): user['duration_concat'] = str(ue_data['duration']) + " j" elif (str(ue_data['duration_unite']) == "semaine"): user['duration_concat'] = str(ue_data['duration']) + " s" elif (str(ue_data['duration_unite']) == "mois"): user['duration_concat'] = str(ue_data['duration']) + " m" elif (str(ue_data['duration_unite']) == "annee"): user['duration_concat'] = str(ue_data['duration']) + " a" else: user['duration_concat'] = str(ue_data['duration']) + " ?" # On ecrase les données si il y a les meme données sur la formation if (eu_class_coef ): user['coefficient'] = str(eu_class_coef) if (eu_class_seuil_validation): user['seuil_validation'] = str(eu_class_seuil_validation) # Si il n'y a pas les infos dans les 2 cas, alors on met une valeur par defaut if( "coefficient" not in user.keys()): user['coefficient'] = "1" if ("seuil_validation" not in user.keys()): user['seuil_validation'] = "99" RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des unites d'enseignement de la formation " """ Cette fonction permet de recupérer les données d'une UE dans une formation données c'est a dire les detail : credit, is_noted, etc """ def Get_Givent_Unite_Enseignement_Data_Of_Given_Class(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'class_id', 'ue_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'class_id', 'ue_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(diction['class_id'])), 'list_unite_enseignement._id':str(my_partner['ue_id'])} #print(" ### qry 0147 = ", qry) for New_retVal in MYSY_GV.dbname['myclass'].find(qry).sort([("_id", pymongo.DESCENDING), ]): if( "list_unite_enseignement" in New_retVal.keys() ): for local_val in New_retVal['list_unite_enseignement'] : if( "_id" in local_val.keys() and local_val['_id'] == str(my_partner['ue_id']) ): user = local_val user['id'] = str(val_tmp) val_tmp = val_tmp + 1 """ Recuperation des données associées à cette UE """ ue_data = MYSY_GV.dbname['unite_enseignement'].find_one({'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0', '_id':ObjectId(str(local_val['_id']))}) if( ue_data ): for val in ue_data.keys() : user[str(val)] = ue_data[str(val)] if (str(ue_data['duration_unite']) == "heure"): user['duration_concat'] = str(ue_data['duration']) + " h" elif (str(ue_data['duration_unite']) == "jour"): user['duration_concat'] = str(ue_data['duration']) + " j" elif (str(ue_data['duration_unite']) == "semaine"): user['duration_concat'] = str(ue_data['duration']) + " s" elif (str(ue_data['duration_unite']) == "mois"): user['duration_concat'] = str(ue_data['duration']) + " m" elif (str(ue_data['duration_unite']) == "annee"): user['duration_concat'] = str(ue_data['duration']) + " a" else: user['duration_concat'] = str(ue_data['duration']) + " ?" RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" #### RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les données de unites d'enseignement de la formation " """ Cette fonction permet d'ajouter une ligne de planification d'une UE. En effet une UE a sa propre planification qui comprend : - code - titre - nb_heure C'est ensuite cette planification qui sera utiliser dans la planification de la session avec des dates et heure """ def Add_Update_Unite_Enseignement_Planif(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'code', 'titre', 'description', 'objectif', 'prerequis', 'volume_horaire', 'ue_id', 'comment', 'unite_enseignement_planif_id', 'session_formation_sequence_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'code', 'titre', 'volume_horaire', 'ue_id' ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner make_update = "0" if( "unite_enseignement_planif_id" in diction.keys() and diction['unite_enseignement_planif_id']): # Il s'agit d'une mise à jour. Verifier que unite_enseignement_planif_id est valide is_valide_unite_enseignement_planif_id_count = MYSY_GV.dbname['unite_enseignement_planif'].count_documents({'_id':ObjectId(str(diction['unite_enseignement_planif_id'])), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( is_valide_unite_enseignement_planif_id_count != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de ligne à mettre à jour est invalide ") return False, " L'identifiant de ligne à mettre à jour est invalide " make_update = "1" """ Verifier que l'UE est valide """ is_valide_ue = MYSY_GV.dbname["unite_enseignement"].count_documents({'_id':ObjectId(str(diction['ue_id'])), 'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0'}) if( is_valide_ue != 1 ): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'ue est invalide ") return False, " L'identifiant de l'ue est invalide " mydata = {} local_token = diction['token'] mydata = diction del mydata['token'] # Initialisation des champs non envoyés à vide for val in field_list: if val not in diction.keys(): mydata[str(val)] = "" """ Verifier le volume horaire """ volume_horaire_status, volume_horaire_value = mycommon.IsFloat(str(diction['volume_horaire'])) if (volume_horaire_status is False): mycommon.myprint(str(inspect.stack()[0][3]) + " Le volume horaire est invalide ") return False, "Le volume horaire est invalide " now = str(datetime.now()) mydata['date_update'] = now mydata['valide'] = "1" mydata['locked'] = "0" mydata['partner_owner_recid'] = str(my_partner['recid']) mydata['update_by'] = str(my_partner['_id']) if( make_update == "1"): data_cle = {} data_cle['partner_owner_recid'] = str(my_partner['recid']) data_cle['ue_id'] = str(diction['ue_id']) data_cle['_id'] = ObjectId(str(diction['unite_enseignement_planif_id'])) data_cle['valide'] = "1" data_cle['locked'] = "0" result = MYSY_GV.dbname['unite_enseignement_planif'].find_one_and_update( data_cle, {"$set": mydata}, upsert=True, return_document=ReturnDocument.AFTER ) if (result is None or "_id" not in result.keys()): mycommon.myprint( " impossible de mettre à jour la planification de l'UE (2) ") return False, " impossible mettre à jour la planification de l'UE " else: inserted_id = "" inserted_id = MYSY_GV.dbname['unite_enseignement_planif'].insert_one(mydata).inserted_id if (not inserted_id): mycommon.myprint( " Impossible de créer la planification de l'UE (2) ") return False, " Impossible de créer la planification de l'UE (2) " return True, "La planification de l'UE a été correctement ajoutée / Mise à jour" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de créer la planification de l'UE" """ Supprimer une liste de planification UE. /!\ : Une regles qui est planifiée ne peut etre supprimé """ def Delete_List_Unite_Enseignement_Planif(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'tab_eu_planif_id', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'tab_eu_planif_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner tab_eu_planif_id = "" if ("tab_eu_planif_id" in diction.keys()): if diction['tab_eu_planif_id']: tab_eu_planif_id = diction['tab_eu_planif_id'] tab_eu_planif_id_splited = str(tab_eu_planif_id).split(",") tab_eu_planif_id_splited_Obj = [] for tmp in tab_eu_planif_id_splited: if( tmp): tab_eu_planif_id_splited_Obj.append(ObjectId(str(tmp))) unite_enseignement_planif_data = MYSY_GV.dbname['unite_enseignement_planif'].find_one({'_id':ObjectId(str(tmp)), 'partner_owner_recid':str(my_partner['recid'])}) if( unite_enseignement_planif_data and "session_formation_sequence_id" in unite_enseignement_planif_data.keys() and unite_enseignement_planif_data['session_formation_sequence_id']): mycommon.myprint( str(inspect.stack()[0][3]) + " La ligne de planification "+str(unite_enseignement_planif_data['code'])+" est actuellement utilisée. Suppression annulée") return False, " La ligne de planification "+str(unite_enseignement_planif_data['code'])+" est actuellement utilisée. Suppression annulée " qery_delete = {'_id': {'$in': tab_eu_planif_id_splited_Obj}, 'partner_owner_recid': str(my_partner['recid']), 'locked': '0', } #print(" ### qery_delete = ", qery_delete) ret_del_competence = MYSY_GV.dbname['unite_enseignement_planif'].delete_many(qery_delete) return True, str(len(tab_eu_planif_id_splited_Obj))+ " ligne(s) supprimée(s) " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de supprimer lignes de planification de l'UE " """ Ajouter des lignes de planification en masse avec un fichier csv """ def Add_Update_Unite_Enseignement_Planif_mass(file=None, Folder=None, diction=None): try: diction = mycommon.strip_dictionary(diction) ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint( str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas, Creation session annulée") return False, " Le champ '" + val + "' n'existe pas, Creation session annulée " ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['token', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner status, saved_file = mycommon.Upload_Save_CSV_File(file, Folder) if (status == False): return False, "Impossible d'importer la liste des sessions, le nom du fichier est incorrect " # " Lecture du fichier " # print(" Lecture du fichier : "+saved_file) nb_line = 0 """" update du 31/08/23 : Controle de l'integrité du fichier avant import """ local_controle_status, local_controle_message = Controle_Add_Update_Unite_Enseignement_Planif_mass( saved_file, Folder, diction) if (local_controle_status is False): return local_controle_status, local_controle_message print(" #### local_controle_message = ", local_controle_message) df = pd.read_csv(saved_file, encoding='utf8', on_bad_lines='skip', sep=';', encoding_errors='ignore') df = df.fillna('') df = df.applymap(lambda x: x.strip() if isinstance(x, str) else x) # Dictionnaire des champs utilisables ''' # Verification que les noms des colonne sont bien corrects" ''' field_list = ['code', 'titre', 'description', 'objectif', 'prerequis', 'volume_horaire', 'ue_code'] # Controle du nombre de lignes dans le fichier. total_rows = len(df) if (total_rows > MYSY_GV.MAX_EU_PLANIF_BY_CSV): mycommon.myprint( str(inspect.stack()[0][3]) + " Le fichier comporte plus de " + str( MYSY_GV.MAX_EU_PLANIF_BY_CSV) + " lignes.") return False, " Le fichier comporte plus de " + str(MYSY_GV.MAX_EU_PLANIF_BY_CSV) + " lignes." # print(df.columns) for val in df.columns: if str(val).lower() not in field_list: mycommon.myprint( str(inspect.stack()[0][3]) + " : entete du fichier csv. '" + val + "' n'est pas acceptée") return False, " Entete du fichier csv. La Colonne '" + val + "' n'est pas acceptée" # Verification des champs obligatoires dans le fichier field_list_obligatoire_file = ['code', 'titre', 'volume_horaire', 'ue_code'] for val in field_list_obligatoire_file: if val not in df.columns: mycommon.myprint( str(inspect.stack()[0][ 3]) + " : Le champ '" + val + "' n'est pas présent dans le fichier. Il est obligatoire") return False, " Le champ '" + val + "' n'est pas présent dans le fichier. Il est obligatoire " x = range(0, total_rows) ignored_line = "" nb_inserted_line = 0 for n in x: mydata = {} mydata['token'] = diction['token'] nb_inserted_line = nb_inserted_line + 1 # Si une ligne n'a aucune information obligatoire, alors on ignore la ligne if (str(df['code'].values[n]) == "nan" or str(df['titre'].values[n]) == "nan" or str(df['volume_horaire'].values[n]) == "nan" or str(df['ue_code'].values[n]) == "nan"): mycommon.myprint(str( inspect.stack()[0][ 3]) + " - La ligne " + str(n + 2) + " a été ignorée") ignored_line = str(n + 2) + " , " + str(ignored_line) nb_inserted_line = nb_inserted_line - 1 continue ue_code = "" ue_id = "" if ("ue_code" in df.keys()): if (str(df['ue_code'].values[n])): ue_code = str(df['ue_code'].values[n]).strip() # On verifie la validité de l'unité d'enseignement count_eu = MYSY_GV.dbname['unite_enseignement'].count_documents( {'code': str(ue_code), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) if (count_eu != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " Ligne " + str( n + 2) + " : Le ue_code n'est pas valide.") return False, " Ligne " + str(n + 2) + " : Le ue_code n'est pas valide." eu_data = MYSY_GV.dbname['unite_enseignement'].find_one( {'code': str(ue_code), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) ue_id = str(eu_data['_id']) else: mycommon.myprint( str(inspect.stack()[0][3]) + " Erreur : Ligne " + str( n + 2) + " : Le ue_code n'est pas invalide.") return False, " Erreur : Ligne " + str( n + 2) + " : Le ue_code n'est pas invalide." mydata['ue_id'] = ue_id code = "" if ("code" in df.keys()): code = str(df['code'].values[n]).strip() mydata['code'] = code titre = "" if ("titre" in df.keys()): titre = str(df['titre'].values[n]).strip() mydata['titre'] = titre if (titre == ""): mycommon.myprint( str(inspect.stack()[0][3]) + " Erreur : Ligne " + str( n + 2) + " : Le Titre est vide .") return False, " Erreur : Ligne " + str( n + 2) + " : Le Titre est vide." description = "" if ("description" in df.keys()): description = str(df['description'].values[n]).strip() mydata['description'] = description objectif = "" if ("objectif" in df.keys()): objectif = str(df['objectif'].values[n]).strip() mydata['objectif'] = objectif prerequis = "" if ("prerequis" in df.keys()): prerequis = str(df['prerequis'].values[n]).strip() mydata['prerequis'] = prerequis volume_horaire = "0" if ("volume_horaire" in df.keys()): if (str(df['volume_horaire'].values[n])): volume_horaire = str(df['volume_horaire'].values[n]).strip() local_status, volume_horaire = mycommon.IsFloat(volume_horaire) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][3]) + " Le champ 'volume_horaire' de la ligne " + str( n + 2) + " est incorrecte.") return False, " Le champ 'volume_horaire' de la ligne " + str(n + 2) + " est incorrecte. " mydata['volume_horaire'] = str(volume_horaire) clean_dict = {k: mydata[k] for k in mydata if (str(mydata[k]) != "nan")} #print("#### Add_Update_Unite_Enseignement_Planif_mass : clean_dict ", clean_dict) status, retval = Add_Update_Unite_Enseignement_Planif(clean_dict) if (status is False): return status, retval print(str(total_rows) + " Lignes de planification ont été inserées") message_ignored_line = "" if (ignored_line): message_ignored_line = " ATTENTION - Les lignes [" + str( ignored_line) + "] ont été ignorées. car les toutes informations obligatoires ne sont pas fournies" return True, str(nb_inserted_line) + " Lignes de planification ont été inserées / Mises à jour. " + str(message_ignored_line) except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, "Impossible d'importer les lignes de planification de l'ue de formation en masse " """ Controle fichier avant import """ def Controle_Add_Update_Unite_Enseignement_Planif_mass(saved_file=None, Folder=None, diction=None): try: diction = mycommon.strip_dictionary(diction) ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint( str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas, Creation session annulée") return False, " Le champ '" + val + "' n'existe pas, Creation session annulée " ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['token', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner nb_line = 0 df = pd.read_csv(saved_file, encoding='utf8', on_bad_lines='skip', sep=';', encoding_errors='ignore') df = df.fillna('') df = df.applymap(lambda x: x.strip() if isinstance(x, str) else x) # Dictionnaire des champs utilisables ''' # Verification que les noms des colonne sont bien corrects" ''' field_list = ['code', 'titre', 'description', 'objectif', 'prerequis', 'volume_horaire', 'ue_code'] # Controle du nombre de lignes dans le fichier. total_rows = len(df) if (total_rows > MYSY_GV.MAX_PARTICIPANT_BY_CSV): mycommon.myprint( str(inspect.stack()[0][3]) + " Le fichier comporte plus de " + str( MYSY_GV.MAX_PARTICIPANT_BY_CSV) + " lignes.") return False, " Le fichier comporte plus de " + str(MYSY_GV.MAX_PARTICIPANT_BY_CSV) + " lignes." # print(df.columns) for val in df.columns: if str(val).lower() not in field_list: mycommon.myprint( str(inspect.stack()[0][3]) + " : entete du fichier csv. '" + val + "' n'est pas acceptée") return False, " Entete du fichier csv. La Colonne '" + val + "' n'est pas acceptée" # Verification des champs obligatoires dans le fichier field_list_obligatoire_file = ['code', 'titre', 'description', 'volume_horaire', 'ue_code'] for val in field_list_obligatoire_file: if val not in df.columns: mycommon.myprint( str(inspect.stack()[0][ 3]) + " : Le champ '" + val + "' n'est pas présent dans le fichier. Il est obligatoire") return False, " Le champ '" + val + "' n'est pas présent dans le fichier. Il est obligatoire " x = range(0, total_rows) ignored_line = "" nb_inserted_line = 0 for n in x: mydata = {} nb_inserted_line = nb_inserted_line + 1 # Si une ligne n'a aucune information obligatoire, alors on ignore la ligne if (str(df['code'].values[n]) == "nan" or str(df['titre'].values[n]) == "nan" or str(df['volume_horaire'].values[n]) == "nan" or str(df['ue_code'].values[n]) == "nan"): mycommon.myprint(str( inspect.stack()[0][ 3]) + " - La ligne " + str(n + 2) + " a été ignorée") ignored_line = str(n + 2) + " , " + str(ignored_line) nb_inserted_line = nb_inserted_line - 1 continue ue_code = "" if ("ue_code" in df.keys()): if (str(df['ue_code'].values[n])): ue_code = str(df['ue_code'].values[n]).strip() # On verifie la validité de l'unité d'enseignement count_eu = MYSY_GV.dbname['unite_enseignement'].count_documents( {'code': str(ue_code), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) if (count_eu != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " Ligne " + str( n + 2) + " : Le ue_code n'est pas valide.") return False, " Ligne " + str(n + 2) + " : Le ue_code n'est pas valide." else: mycommon.myprint( str(inspect.stack()[0][3]) + " Erreur : Ligne " + str( n + 2) + " : Le ue_code n'est pas invalide.") return False, " Erreur : Ligne " + str( n + 2) + " : Le ue_code n'est pas invalide." titre = "" if ("titre" in df.keys()): titre = str(df['titre'].values[n]).strip() mydata['titre'] = titre if( titre == ""): mycommon.myprint( str(inspect.stack()[0][3]) + " Erreur : Ligne " + str( n + 2) + " : Le Titre est vide .") return False, " Erreur : Ligne " + str( n + 2) + " : Le Titre est vide." volume_horaire = "0" if ("volume_horaire" in df.keys()): if (str(df['volume_horaire'].values[n])): volume_horaire = str(df['volume_horaire'].values[n]).strip() local_status, volume_horaire = mycommon.IsFloat(volume_horaire) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][3]) + " Le champ 'volume_horaire' de la ligne " + str( n + 2) + " est incorrecte.") return False, " Le champ 'volume_horaire' de la ligne " + str(n + 2) + " est incorrecte. " mydata['volume_horaire'] = str(volume_horaire) return True, str(total_rows)+" sessions dans le fichier" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, "Impossible de controler le fichier de la planification de l'UE en masse " """ Recuperer les lignes de planification d'une UE """ def Get_List_Unite_Enseignement_Planif_lines(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'ue_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas") return False, " Les informations fournies sont incorrectes", """ Verification des champs obligatoires """ field_list_obligatoire = ['token', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes", """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner RetObject = [] val_tmp = 0 find_qry = {'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0', 'ue_id':str(diction['ue_id'])} print("### Get_List_Unite_Enseignement_Planif_lines find_qry = ", find_qry) for retval in MYSY_GV.dbname['unite_enseignement_planif'].find(find_qry).sort([("_id", pymongo.DESCENDING), ]): user = retval user['id'] = str(val_tmp) local_sequence_title = "" if( "session_formation_sequence_id" not in user.keys() ): user['session_formation_sequence_id'] = "" user['in_use'] = "0" if ("session_formation_sequence_id" in user.keys() ): if( user['session_formation_sequence_id'] ): user['in_use'] = "1" # Recuperer le titre de la sequence sequence_data = MYSY_GV.dbname['session_formation_sequence'].find_one( {'_id': ObjectId(str(user['session_formation_sequence_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) if (sequence_data and "sequence_title" in sequence_data.keys() and "session_id" in sequence_data.keys()): session_data = MYSY_GV.dbname['session_formation'].find_one({'_id':ObjectId(str(sequence_data['session_id'])), 'valide':'1', 'partner_owner_recid':str(my_partner['recid'])}) if(session_data and "code_session" in session_data.keys() and "sequence_title" in sequence_data.keys()): local_sequence_title = session_data['code_session'] +" - "+sequence_data['sequence_title'] else: user['in_use'] = "0" user['used_sequence_title'] = local_sequence_title val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la planification de l'UE " """ Recuperer les données d'une lignes de planification UE donnée """ def Get_Given_Unite_Enseignement_Planif_Data(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'tab_eu_planif_id', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'tab_eu_planif_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner tab_eu_planif_id = "" if ("tab_eu_planif_id" in diction.keys()): if diction['tab_eu_planif_id']: tab_eu_planif_id = diction['tab_eu_planif_id'] tab_eu_planif_id_splited = str(tab_eu_planif_id).split(",") tab_eu_planif_id_splited_Obj = [] for tmp in tab_eu_planif_id_splited: if( tmp): tab_eu_planif_id_splited_Obj.append(ObjectId(str(tmp))) find_qry = {'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0', '_id': {'$in': tab_eu_planif_id_splited_Obj},} RetObject = [] val_tmp = 0 for retval in MYSY_GV.dbname['unite_enseignement_planif'].find(find_qry).sort([("_id", pymongo.DESCENDING), ]): user = retval user['id'] = str(val_tmp) local_sequence_title = "" if ("session_formation_sequence_id" not in user.keys()): user['session_formation_sequence_id'] = "" user['in_use'] = "0" if ("session_formation_sequence_id" in user.keys()): if (user['session_formation_sequence_id']): user['in_use'] = "1" # Recuperer le titre de la sequence sequence_data = MYSY_GV.dbname['session_formation_sequence'].find_one({'_id':str(user['session_formation_sequence_id']), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( sequence_data and "sequence_title" in sequence_data.keys() ): local_sequence_title = sequence_data['sequence_data'] else: user['in_use'] = "0" user['used_sequence_title'] = local_sequence_title val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récuperer la ligne de planification de l'UE "