Elyos_FI_Back_Office/unite_enseignement_mgt.py

1891 lines
77 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

"""
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é denseignement (UE) & formation :
Lors du rattachement dune 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 dUE
"""
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 "