Elyos_FI_Back_Office/ent_student_common.py

1300 lines
52 KiB
Python

"""
-- Gestion de l'espace ENT des apprenants
Ce fichier permet de gerer les fonction de ENT ETUDIANT
"""
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
"""
Cette fonctions retourne les _id d'un apprenant return : tab_class_id, tab_inscription_id, tab_session_id
exemple :
"message": [
{
"class_id": "664240065407cfe49176e9a6",
"inscription_id": "66d572731814d63cbc46aa03",
"session_id": "66cdeb5d7e8061b7b6f9b066"
},
{
"class_id": "6655b721fd47c656dcaf8f05",
"inscription_id": "66c0d63c72635968747aa0f4",
"session_id": "66c0d58feca5acd6f5c839c0"
}
],
"""
def Get_Apprenant_Tab_Class_Inscrip_Session_Id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'apprenant_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', 'apprenant_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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
qry = {}
qry['partner_owner_recid'] = apprenant_data['partner_owner_recid']
qry['apprenant_id'] = diction['apprenant_id']
qry['valide'] = "1"
qry['locked'] = "0"
query = [{'$match': {'$and': [qry]}},
{'$sort': {'_id': -1}},
{'$lookup':
{
'from': 'myclass',
'localField': 'class_internal_url',
'foreignField': 'internal_url',
'pipeline': [{'$match': {'$and': [{}, {}]}},
{'$project': {'_id': 1, }}],
'as': 'myclass_collection'
}
},
{'$lookup':
{
'from': 'apprenant',
"let": {'apprenant_id': "$apprenant_id", 'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [{'$match':
{'$expr': {'$and': [
{'$eq': ["$valide", "1"]},
{'$eq': ["$_id", {'$convert': {
'input': "$$apprenant_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
]}}},
], 'as': 'apprenant_collection'}}
]
print("#### Get_Apprenant_Tab_Class_Inscrip_Session_Id laa 01 : query = ", query)
RetObject = []
cpt = 0
for retVal in MYSY_GV.dbname['inscription'].aggregate(query):
val = {}
local_node = {}
local_node['inscription_id'] = (str(retVal['_id']))
local_node['session_id'] = (str(retVal['session_id']))
#print('### retVal = ', retVal)
if ('myclass_collection' in retVal.keys() and len(retVal['myclass_collection']) > 0):
local_node['class_id'] = (str(retVal['myclass_collection'][0]['_id']))
RetObject.append(local_node)
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 les caracteristiques de l'apprenant : tab_class_id, tab_inscription_id, tab_session_id"
"""
Recuperer la liste des formation d'un apprenant pour son espace ENT
"""
def Get_Ent_Student_List_Class(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'apprenant_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', 'apprenant_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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents({'_id':ObjectId(str(diction['apprenant_id'])),
'valide':'1',
'locked':'0'})
if( is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
local_diction = {}
local_diction['token'] = str(diction['token'])
local_diction['apprenant_id'] = str(diction['apprenant_id'])
retval_status, retval_data = Get_Apprenant_Tab_Class_Inscrip_Session_Id(local_diction)
if( retval_status is False ):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " Impossible de récuperer la liste des formations ")
return False, " Impossible de récuperer la liste des formations "
tab_class_id = retval_data
#print(" ### tab_class_id = ", tab_class_id)
RetObject = []
cpt = 0
list_champ_a_retourner = {'_id':1, 'external_code':1, 'title':1, 'objectif':1, 'programme':1, 'prerequis':1, 'description':1,
'internal_url':1, 'partner_owner_recid':1}
for local_data in tab_class_id:
class_data = MYSY_GV.dbname['myclass'].find_one({'_id':ObjectId(str(local_data['class_id'])), 'valide':'1'}, list_champ_a_retourner)
if ("description" in class_data.keys()):
val = class_data['description']
no_html = mycommon.cleanhtml(class_data['description'])
if( len(no_html) > MYSY_GV.MAX_CARACT ):
class_data['description'] = no_html[:MYSY_GV.MAX_CARACT]+" ..."
else:
class_data['description'] = no_html
user = class_data
sess_data = MYSY_GV.dbname['session_formation'].find_one({"_id":ObjectId(str(local_data['session_id'])),
'valide':'1',
'partner_owner_recid':str(class_data['partner_owner_recid'])})
user['session_id'] = str(local_data['session_id'])
user['session_code_session'] = str(sess_data['code_session'])
user['session_date_debut'] = str(sess_data['date_debut'])
user['session_date_fin'] = str(sess_data['date_fin'])
user['inscription_id'] = str(local_data['inscription_id'])
user['id'] = str(cpt)
cpt = cpt +1
RetObject.append(mycommon.JSONEncoder().encode(user))
"""
qry = {}
qry['partner_owner_recid'] = apprenant_data['partner_owner_recid']
qry['apprenant_id'] = diction['apprenant_id']
qry['valide'] = "1"
qry['locked'] = "0"
query = [{'$match': {'$and': [qry ]}},
{'$sort': {'_id': -1}},
{'$lookup':
{
'from': 'myclass',
'localField': 'class_internal_url',
'foreignField': 'internal_url',
'pipeline': [{'$match': {'$and': [{}, {}]}},
{'$project': {'title': 1, 'domaine': 1,
'duration': 1,
'duration_unit': 1,
'_id': 1,
'recyclage_delai': 1,
'recyclage_periodicite': 1,
'recyclage_alert': 1}}],
'as': 'myclass_collection'
}
},
{'$lookup':
{
'from': 'apprenant',
"let": {'apprenant_id': "$apprenant_id", 'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [{'$match':
{'$expr': {'$and': [
{'$eq': ["$valide", "1"]},
{'$eq': ["$_id", {'$convert': {
'input': "$$apprenant_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
]}}},
], 'as': 'apprenant_collection'}}
]
print("#### Get_Ent_Student_Class laa 01 : query = ", query)
RetObject = []
cpt = 0
for retVal in MYSY_GV.dbname['inscription'].aggregate(query):
print("### retValretVal = ", retVal)
val = {}
if ('myclass_collection' in retVal.keys() and len(retVal['myclass_collection']) > 0):
val['id'] = str(cpt)
cpt = cpt + 1
val['_id'] = retVal['_id']
"""
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 formations de l'apprenant "
"""
Cette fonction retourne une formation, prenant en argument le l'_id de la formation
"""
def get_Ent_Student_Given_Class_From_Class_Id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['class_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 = ['class_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",
RetObject = []
val_tmp = 1
qry = {'valide': '1', 'locked': '0', '_id':ObjectId(str(diction['class_id']))}
for retval in MYSY_GV.dbname['myclass'].find(qry):
user = retval
user['id'] = str(val_tmp)
if ("recyclage_delai" not in user.keys()):
user['recyclage_delai'] = ""
if ("recyclage_alert" not in user.keys()):
user['recyclage_alert'] = ""
if ("recyclage_periodicite" not in user.keys()):
user['recyclage_periodicite'] = ""
if ("note_finale_calculation_rule_id" not in user.keys()):
user['note_finale_calculation_rule_id'] = ""
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 les données de la formation "
"""
Recuperer la liste des sequence d'une formation
"""
def Get_Ent_Student_Session_Sequence_List(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'session_id', 'apprenant_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', 'session_id', 'apprenant_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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
RetObject = []
val_tmp = 0
nb_sequence = 0
for New_retVal in MYSY_GV.dbname['session_formation_sequence'].find({'session_id':str(diction['session_id']),
'partner_owner_recid':str(apprenant_data['partner_owner_recid']), 'valide':'1', 'locked':'0'}):
user = New_retVal
user['id'] = str(val_tmp)
val_tmp = val_tmp + 1
nb_sequence = nb_sequence + 1
if ("type" not in user.keys()):
user['type'] = ""
if( "grp_apprenant_id" not in user.keys() ):
user['grp_apprenant_id'] = ""
grp_apprenant_color = ""
grp_apprenant_code = ""
"""
Recuperer la couleur du groupe
"""
if( "grp_apprenant_id" in user.keys() and user['grp_apprenant_id']):
groupe_data = MYSY_GV.dbname['groupe_inscription'].find_one({'_id':ObjectId(str(user['grp_apprenant_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(apprenant_data['partner_owner_recid'])},
{'code':1, 'color':1})
if( groupe_data and "color" in groupe_data.keys() and groupe_data['color']):
grp_apprenant_color = groupe_data['color']
if (groupe_data and "code" in groupe_data.keys() and groupe_data['code']):
grp_apprenant_code = groupe_data['code']
user['grp_apprenant_color'] = grp_apprenant_color
user['grp_apprenant_code'] = grp_apprenant_code
if ("ue_id" not in user.keys()):
user['ue_id'] = ""
if ("unite_enseignement_planif_id" not in user.keys()):
user['unite_enseignement_planif_id'] = ""
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 liste des séquences de la session "
"""
Cette fonction retourne la liste des UE aux quelles un apprenant est inscrit
"""
def Get_Ent_Student_Inscrit_List_EU(diction):
try:
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = ['token', 'inscription_id', 'apprenant_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 ne sont pas correctes"
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
"""
Verifier la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
"""
Verifier que l'inscription est valide
"""
is_valide_inscription_count = MYSY_GV.dbname['inscription'].count_documents(
{'_id': ObjectId(diction['inscription_id']),
'partner_owner_recid': str(apprenant_data['partner_owner_recid']),
'valide': '1'})
if (is_valide_inscription_count != 1):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
RetObject = []
val_tmp = 0
qery_match = {'inscription_id': str(diction['inscription_id']), 'partner_owner_recid': str(apprenant_data['partner_owner_recid']), 'valide': '1',
'locked': '0'}
pipe_qry = ([
{'$match': qery_match},
{'$lookup': {
'from': 'unite_enseignement',
"let": {'class_eu_id': "$class_eu_id",
'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [
{'$match':
{'$expr':
{'$and':
[
{'$eq': ["$_id", {'$convert': {
'input': "$$class_eu_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
{'$eq': ["$valide", "1"]},
{'$eq': ["$partner_owner_recid", '$$partner_owner_recid']}
]
}
}
},
],
'as': 'collection_unite_enseignement'
}
},
{
'$unwind': '$collection_unite_enseignement'
}
])
for val in MYSY_GV.dbname['inscription_liste_ue'].aggregate(pipe_qry):
user = {}
main_field = ['_id', 'class_eu_id', 'class_id', 'inscription_id', ]
second_field = ['_id', 'code', 'titre']
for tmp in main_field :
if( str(tmp) in val.keys() ):
user[str(tmp)] = val[str(tmp)]
for tmp in second_field:
if (str(tmp) in val['collection_unite_enseignement'].keys()):
user["ue_"+str(tmp)] = val['collection_unite_enseignement'][str(tmp)]
RetObject.append(mycommon.JSONEncoder().encode(user))
return True, RetObject
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, " Impossible de recuperer les UE associées à l'inscription"
"""
Cette fonction permet de recuperer la liste des types d'evaluation pour un inscrit
exemple : Pour l'UE xxxx, il est inscrit pour les 'PROJETS' seulement
"""
def Get_Ent_Student_Inscrit_List_EU_Type_Evaluation(diction):
try:
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = ['token', 'inscription_id', 'apprenant_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 ne sont pas correctes"
mytoken = ""
if ("token" in diction.keys()):
if diction['token']:
mytoken = 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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
"""
Verifier que l'inscription est valide
"""
is_valide_inscription_count = MYSY_GV.dbname['inscription'].count_documents(
{'_id': ObjectId(diction['inscription_id']),
'partner_owner_recid': str(apprenant_data['partner_owner_recid']),
'valide': '1'})
if (is_valide_inscription_count != 1):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
RetObject = []
val_tmp = 0
qery_match = {'inscription_id': str(diction['inscription_id']), 'partner_owner_recid': str(apprenant_data['partner_owner_recid']), 'valide': '1',
'locked': '0'}
pipe_qry = ([
{'$match': qery_match},
{'$lookup': {
'from': 'type_evaluation',
"let": {'type_evaluation_id': '$type_evaluation_id',
'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [
{'$match':
{'$expr':
{'$and':
[
{'$eq': ["$_id", {'$convert': {
'input': "$$type_evaluation_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
{'$eq': ["$valide", "1"]},
{'$eq': ["$partner_owner_recid", '$$partner_owner_recid']}
]
}
}
},
],
'as': 'collection_type_evaluation'
}
},
{
'$unwind': '$collection_type_evaluation'
}
])
#print(" ### Get_Inscrit_List_EU_Type_Evaluation pipe_qry = ", pipe_qry)
for val in MYSY_GV.dbname['inscription_liste_ue_type_evalution'].aggregate(pipe_qry):
user = {}
main_field = ['_id', 'class_eu_id', 'class_id', 'inscription_id', 'type_evaluation_id']
second_field = ['_id', 'code', 'nom']
for tmp in main_field :
if( str(tmp) in val.keys() ):
user[str(tmp)] = val[str(tmp)]
class_eu_code = ""
if( "class_eu_id" in val.keys() ):
class_eu_id_data = MYSY_GV.dbname['unite_enseignement'].find_one({'_id':ObjectId(str(val['class_eu_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(apprenant_data['partner_owner_recid'])})
if( class_eu_id_data and 'code' in class_eu_id_data.keys() ):
class_eu_code = class_eu_id_data['code']
user['class_eu_code'] = class_eu_code
for tmp in second_field:
if (str(tmp) in val['collection_type_evaluation'].keys()):
user["type_eval_"+str(tmp)] = val['collection_type_evaluation'][str(tmp)]
RetObject.append(mycommon.JSONEncoder().encode(user))
#print(" ### Get_Inscrit_List_EU_Type_Evaluation RetObject = ", RetObject)
return True, RetObject
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, " Impossible de recuperer la liste des évaluation par UE "
"""
Recuperation des notes d'un inscrit a une liste de session_id
"""
def Get_Ent_Student_Participant_Notes(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'tab_session_id', 'tab_inscription_id', 'apprenant_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_session_id', 'tab_inscription_id', 'apprenant_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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
tab_session_id = ""
if ("tab_session_id" in diction.keys()):
if diction['tab_session_id']:
tab_session_id = diction['tab_session_id']
tab_inscription_id = ""
if ("tab_inscription_id" in diction.keys()):
if diction['tab_inscription_id']:
tab_inscription_id = diction['tab_inscription_id']
tab_my_session_ids = str(tab_session_id).split(",")
tab_my_inscription_ids = str(tab_inscription_id).split(",")
"""
Verification de la validité de la liste des inscription
"""
#print("tab_my_session_ids = ", tab_my_session_ids)
for my_inscription_id in tab_my_inscription_ids:
# Verifier qui la formation n'a pas deja été evaluée
tmp_count = MYSY_GV.dbname['inscription'].count_documents({'session_id': {'$in': tab_my_session_ids, },
'_id': ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(apprenant_data['partner_owner_recid']),
} )
if (tmp_count != 1):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'inscription "+str(my_inscription_id)+" est invalide ")
return False, " L'inscription "+str(my_inscription_id)+" est invalide "
"""
Verification de la validité de la liste des sessions
"""
for my_session_ids in tab_my_session_ids:
# Verifier qui la formation n'a pas deja été evaluée
tmp_count = MYSY_GV.dbname['session_formation'].count_documents({'_id': ObjectId(str(my_session_ids)),
'partner_owner_recid': str(apprenant_data['partner_owner_recid']),
'valide':'1'
} )
if (tmp_count != 1):
mycommon.myprint(str(inspect.stack()[0][3]) + " La session "+str(my_session_ids)+" est invalide ")
return False, " L'inscription "+str(my_session_ids)+" est invalide "
RetObject = []
nb_val = 0
query = [{'$match': {'inscription_id': {'$in':tab_my_inscription_ids}, 'partner_owner_recid': str(apprenant_data['partner_owner_recid']),
}},
{'$sort': {'_id': -1}},
{'$lookup':
{
'from': 'note_evaluation',
"let": {'evaluation_id': "$evaluation_id", 'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [{'$match':
{'$expr': {'$and': [
{'$eq': ["$valide", "1"]},
{'$eq': ["$_id", {'$convert': {
'input': "$$evaluation_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
]}}},
], 'as': 'note_evaluation_collection'}
}
]
#print("#### Get_List_Participant_Notes laa 01 : query = ", query)
for retval in MYSY_GV.dbname['note_evaluation_participant'].aggregate(query):
if( "note_evaluation_collection" in retval.keys() ):
user = {}
nom_apprenant = ""
prenom_apprenant = ""
email_apprenant = ""
groupe = ""
if( "inscription_id" in retval and retval['inscription_id']):
# Recuprer les données de l'inscrit
inscription_data = MYSY_GV.dbname['inscription'].find_one({'_id':ObjectId(str(retval['inscription_id']))},
{'_id':1, 'apprenant_id':1})
apprenant_data = MYSY_GV.dbname['apprenant'].find_one({'_id':ObjectId(str(inscription_data['apprenant_id']))},
{'_id':1, 'nom':1, 'prenom':1, 'email':1, 'partner_owner_recid':1})
nom_apprenant = apprenant_data['nom']
prenom_apprenant = apprenant_data['prenom']
email_apprenant = apprenant_data['email']
user['_id'] = str(retval['_id'])
user['evaluation_id'] = str(retval['evaluation_id'])
user['inscription_id'] = str(retval['inscription_id'])
user['note'] = str(retval['note'])
user['note_evaluation_id'] = str(retval['note_evaluation_collection'][0]['_id'])
user['note_evaluation_code'] = str(retval['note_evaluation_collection'][0]['code'])
user['note_evaluation_titre'] = str(retval['note_evaluation_collection'][0]['titre'])
user['session_id'] = str(retval['note_evaluation_collection'][0]['session_id'])
user['class_eu_id'] = str(retval['note_evaluation_collection'][0]['class_eu_id'])
class_ue_code = ""
class_ue_titre = ""
ue_data = MYSY_GV.dbname['unite_enseignement'].find_one({'_id':ObjectId(str(user['class_eu_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(apprenant_data['partner_owner_recid'])})
if( ue_data and "code" in ue_data.keys() ):
class_ue_code = str(ue_data['code'])
if ( ue_data and "titre" in ue_data.keys()):
class_ue_titre = str(ue_data['titre'])
user['class_ue_code'] = class_ue_code
user['class_ue_titre'] = class_ue_titre
user['type_eval_id'] = str(retval['note_evaluation_collection'][0]['type_eval_id'])
type_eval_data = MYSY_GV.dbname['type_eval_id'].find_one({'_id':ObjectId(str(user['type_eval_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(apprenant_data['partner_owner_recid'])})
if( type_eval_data and 'code' in type_eval_data.keys() ):
user['type_eval_code'] = type_eval_data['code']
else:
user['type_eval_code'] = ""
user['eval_date_heure_debut'] = str(retval['note_evaluation_collection'][0]['eval_date_heure_debut'])
user['eval_date_heure_fin'] = str(retval['note_evaluation_collection'][0]['eval_date_heure_fin'])
user['nom'] = nom_apprenant
user['prenom'] = prenom_apprenant
user['email'] = email_apprenant
user['id'] = str(nb_val)
if( "note" not in retval.keys() ):
user['note'] = "-1"
nb_val = nb_val + 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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de récupérer les notes de l'apprenant"
"""
Recuperer la liste des UE d'une formation donnée
"""
def Get_Ent_Student_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', 'apprenant_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','apprenant_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 la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
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(apprenant_data['partner_owner_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(apprenant_data['partner_owner_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(apprenant_data['partner_owner_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 "
"""
Recuperation de la liste des evaluations d'une formation
"""
def Get_Ent_Student_List_Class_Evaluation(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'class_id', 'apprenant_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', 'class_id', 'apprenant_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']
"""
Verifier la validité de l'apprenant
"""
is_apprenant_valide_count = MYSY_GV.dbname['apprenant'].count_documents(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
if (is_apprenant_valide_count != 1):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide ",
apprenant_data = MYSY_GV.dbname['apprenant'].find_one(
{'_id': ObjectId(str(diction['apprenant_id'])),
'valide': '1',
'locked': '0'})
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
"""
Clés de mise à jour
"""
data_cle = {}
data_cle['class_id'] = str(diction['class_id'])
data_cle['partner_owner_recid'] = str(apprenant_data['partner_owner_recid'])
data_cle['valide'] = "1"
data_cle['locked'] = "0"
RetObject = []
val_tmp = 0
for retval in MYSY_GV.dbname['class_unite_enseignement_type_evaluation'].find(data_cle).sort([("_id", pymongo.DESCENDING), ]):
user = retval
user['id'] = str(val_tmp)
type_evaluation_code = ""
if( "type_evaluation_id" in retval.keys() and retval['type_evaluation_id']):
type_evaluation_data = MYSY_GV.dbname['type_evaluation'].find_one({'_id':ObjectId(str(retval['type_evaluation_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(apprenant_data['partner_owner_recid'])})
if( type_evaluation_data and 'code' in type_evaluation_data.keys() ):
type_evaluation_code= type_evaluation_data['code']
user['type_evaluation_code'] = type_evaluation_code
class_ue_code = ""
if ("class_ue_id" in retval.keys() and retval['class_ue_id']):
class_ue_data = MYSY_GV.dbname['unite_enseignement'].find_one(
{'_id': ObjectId(str(retval['class_ue_id'])),
'valide': '1',
'locked': '0',
'partner_owner_recid': str(my_partner['recid'])})
if (class_ue_data and 'code' in class_ue_data.keys()):
class_ue_code = class_ue_data['code']
user['class_ue_code'] = class_ue_code
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 liste des évaluations de la formation "