""" -- 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 ("mode_animation" not in user.keys()): user['mode_animation'] = "0" if ("ue_id" not in user.keys()): user['ue_id'] = "" if ("unite_enseignement_planif_id" not in user.keys()): user['unite_enseignement_planif_id'] = "" print(" ### user = ", user) 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 " """ Recuperation de la liste des fichiers stocké par : - object_owner_collection - object_owner_id """ def Get_Ent_Student_List_object_owner_collection_Stored_Files(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'object_owner_collection', 'object_owner_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 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'}) object_owner_collection = "" if ("object_owner_collection" in diction.keys()): if diction['object_owner_collection']: object_owner_collection = diction['object_owner_collection'] object_owner_id = "" if ("object_owner_id" in diction.keys()): if diction['object_owner_id']: object_owner_id = diction['object_owner_id'] """ Verification des champs obligatoires 24/12/2024 : Les 2 valeurs ('object_owner_collection', 'object_owner_id') ne sont plus obligatories, mais au moins une doit etre fourni """ if (str(object_owner_collection).strip() == "" and str(object_owner_id).strip() == ""): mycommon.myprint( str(inspect.stack()[0][3]) + " Il manque des informations obligatoires ") return False, " Les informations fournies sont incorrectes", my_query = {} if (object_owner_collection): my_query['object_owner_collection'] = object_owner_collection if (object_owner_id): my_query['object_owner_id'] = object_owner_id my_query['valide'] = "1" my_query['partner_owner_recid'] = str(apprenant_data['partner_owner_recid']) RetObject = [] nb_val = 0 for retval in MYSY_GV.dbname['download_files'].find(my_query).sort( [("_id", pymongo.DESCENDING), ("document_display_order", pymongo.DESCENDING)]): ret_file = {} ret_file['id'] = str(nb_val) ret_file['_id'] = str(retval['_id']) ret_file['file_business_object'] = retval['file_business_object'] if ("creation_date" in retval.keys()): ret_file['creation_date'] = retval['creation_date'] else: ret_file['creation_date'] = "9999-99-99" if ("file_business_object" in retval.keys()): ret_file['file_business_object'] = retval['file_business_object'] if ("object_owner_collection" in retval.keys()): ret_file['object_owner_collection'] = retval['object_owner_collection'] if ("object_owner_id" in retval.keys()): ret_file['object_owner_id'] = retval['object_owner_id'] if ("file_name" in retval.keys()): ret_file['file_name'] = retval['file_name'] if ("status" in retval.keys()): ret_file['status'] = retval['status'] if ("document_display_order" in retval.keys()): ret_file['document_display_order'] = retval['document_display_order'] if ("full_path" in retval.keys()): ret_file['full_path'] = retval['full_path'] if ("file_cononical_name" in retval.keys()): ret_file['file_cononical_name'] = retval['file_cononical_name'] else: ret_file['file_cononical_name'] = "Anonyme" if ("type_document" in retval.keys()): ret_file['type_document'] = retval['type_document'] else: ret_file['type_document'] = "Autre" if ("file_extention" in retval.keys()): ret_file['file_extention'] = retval['file_extention'] else: ret_file['file_extention'] = "file_extention" """ Recuperation du nom du client """ client_raison_sociale = "" client_nom = "" if ("object_owner_id" in retval.keys() and retval["object_owner_id"]): partner_client_data = MYSY_GV.dbname['partner_client'].find_one( {'_id': ObjectId(str(retval["object_owner_id"]))}, {'raison_sociale': 1, 'nom': 1}) if (partner_client_data and "raison_sociale" in partner_client_data.keys()): client_raison_sociale = partner_client_data['raison_sociale'] if (partner_client_data and "nom" in partner_client_data.keys()): client_nom = partner_client_data['nom'] ret_file['client_raison_sociale'] = client_raison_sociale ret_file['client_nom'] = client_nom RetObject.append(mycommon.JSONEncoder().encode(ret_file)) nb_val = nb_val + 1 # print(" ### Get_List_object_owner_collection_Stored_Files = ", 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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False """ Fonction de recuperation des pièces jointes d'objet en utilisant le 'file_name' """ def Get_Ent_Student_Stored_Downloaded_File(diction): try: full_file_path = "" diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'file_name', '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', 'file_name', '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'}) file_name = "" if ("file_name" in diction.keys()): if diction['file_name']: file_name = diction['file_name'] if(len(str(file_name)) > MYSY_GV.FILE_NAME_MAX_SIZE ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " - Fichier invalide, trop long = "+ str(file_name)+ ". le nom doit faire moins de 100 caractère") return False, " - Nom de fichier trop long. Il doit faire moins de 100 caractères" if( not file_name ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " - Fichier invalide = ", str(file_name)) return False, " - nom du fichier invalide " full_path = MYSY_GV.upload_folder+str(file_name) """ Recuperation de la list des fichier dans la collection """ tmp_count = MYSY_GV.dbname['download_files'].count_documents({'full_path': str(full_path), 'valide': '1', 'partner_owner_recid':apprenant_data['partner_owner_recid']}) if (tmp_count <= 0): mycommon.myprint( str(inspect.stack()[0][ 3]) + " - Fichier invalide = ", str(full_path)) return False, " - Fichier invalide " if os.path.exists(full_path): print(" ### full_path = ", full_path) return True, send_file(full_path, as_attachment=True) else: mycommon.myprint( str(inspect.stack()[0][ 3]) + " - Fichier invalide (2)= ", str(full_path)) return False, " - Fichier invalide (2) " 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, False """ Mise à jour du compte de l'apprenant dans lespace ENT """ def Get_Ent_Student_update_user_account(diction): try: # Dictionnaire des champs utilisables ''' # 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 = ['adr_street', 'adr_city', 'adr_zip', 'adr_country', 'mob_phone','link_facebook', 'link_twitter', 'token', 'prenom', 'nom', 'birthday','link_linkedin', '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, "Impossible de mettre à jour le compte utilisateur" ''' 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', 'apprenant_id'] for val in field_list_obligatoire : if val not in diction: mycommon.myprint(" La valeur '"+val+"' n'est pas presente dans la liste des arguments ") return False, "Impossible de mettre à jour le compte utilisateur" mycommon.myprint(" MAJ de diction "+str(diction)) """ 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'}) coll_name = MYSY_GV.dbname['user_account'] now = datetime.now() dictionnary = {} dictionnary['active'] = "1" dictionnary['date_update'] = str(now) mytoken ="" myemail = "" if ("adr_street" in diction.keys()): if( str(diction['adr_street']) ): dictionnary['adr_street'] = str(diction['adr_street']) if ("adr_city" in diction.keys()): if (str(diction['adr_city'])): dictionnary['adr_city'] = str(diction['adr_city']) if( "adr_zip" in diction.keys() ): if (str(diction['adr_zip'])): dictionnary['adr_zip'] = str(diction['adr_zip']) if ("adr_country" in diction.keys()): if (str(diction['adr_country'])): dictionnary['adr_country'] = str(diction['adr_country']) if ("mob_phone" in diction.keys()): if (str(diction['mob_phone'])): dictionnary['mob_phone'] = str(diction['mob_phone']) if ("link_facebook" in diction.keys()): if (str(diction['link_facebook'])): dictionnary['link_facebook'] = str(diction['link_facebook']) if ("link_twitter" in diction.keys()): if (str(diction['link_twitter'])): dictionnary['link_twitter'] = str(diction['link_twitter']) if ("token" in diction.keys()): if (str(diction['token'])): mytoken = str(diction['token']) if ("email" in diction.keys()): if (str(diction['email'])): myemail = str(diction['email']) if ("prenom" in diction.keys()): if (str(diction['prenom'])): dictionnary['prenom'] = str(diction['prenom']) if ("nom" in diction.keys()): if (str(diction['nom'])): dictionnary['nom'] = str(diction['nom']) if ("birthday" in diction.keys()): if (str(diction['birthday'])): dictionnary['birthday'] = str(diction['birthday']) if ("link_linkedin" in diction.keys()): if (str(diction['link_linkedin'])): dictionnary['link_linkedin'] = str(diction['link_linkedin']) local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner mycommon.myprint(" dictionnary = "+str(dictionnary)) """ 23/12/2024 : Après avoir verifier la validité du token, on va aller chercher l'_id dans la collection 'user_account' """ print(" ### QRY011 = ", {'token':str(diction['token']), 'active':'1', 'locked':'0', 'apprenant_id':diction['apprenant_id']}) is_token_valide_for_one_user = MYSY_GV.dbname['user_account'].count_documents({'token':str(diction['token']), 'active':'1', 'locked':'0', 'apprenant_id':diction['apprenant_id']}) if( is_token_valide_for_one_user != 1 ): mycommon.myprint(str(inspect.stack()[0][3]) + " Le token correspond a plusieurs au aucun utilisateur de la collection 'user_account' ") return False, "La session est invalide " print(" ## qry == ", {'token':str(diction['token']), 'active':'1', 'locked':'0', 'apprenant_id':diction['apprenant_id']}) ret_val = coll_name.find_one_and_update({'token':str(diction['token']), 'active':'1', 'locked':'0', 'apprenant_id': diction['apprenant_id'], }, {"$set": dictionnary }, return_document=ReturnDocument.AFTER ) if (ret_val and ret_val['_id']): mycommon.myprint(str(inspect.stack()[0][3]) + " -La modif a bien ete faite =" + str(ret_val['_id'])) return True, " Le compte utilisateur a ete mis à jour" else: mycommon.myprint(str(inspect.stack()[0][3]) + " - Aucune modif n'a été faite") return False, "Impossible de mettre à jour le compte utilisateur" 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 mettre à jour le compte utilisateur" ''' changement du mot de passe d'un user ''' def Get_Ent_Student_change_user_pwd(diction): try: # Dictionnaire des champs utilisables field_list = ['token', 'pwd', 'new_pwd', 'conf_new_pwd', '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(" Le champ '" + val + "' n'est pas autorisé, requete annulée") return False, " Impossible de récupérer le user account" ''' 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', 'pwd', 'new_pwd', 'conf_new_pwd', 'apprenant_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint(" La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Impossible de mettre à jour le mot de passe" # recuperation des paramettre email_value = "" token_value = "" pwd = "" new_pwd = "" conf_new_pwd = "" mydata ={} if ("pwd" in diction.keys()): if diction['pwd']: pwd = diction['pwd'] if ("conf_new_pwd" in diction.keys()): if diction['conf_new_pwd']: conf_new_pwd = diction['conf_new_pwd'] if ("token" in diction.keys()): if diction['token']: token_value = diction['token'] if ("new_pwd" in diction.keys()): if diction['new_pwd']: new_pwd = diction['new_pwd'] mydata['pwd'] = diction['new_pwd'] # Verification de la validité du token/mail dans le cas des user en mode connecté if ( len(str(token_value)) <= 0): mycommon.myprint(str(inspect.stack()[0][3]) + " Le token est vide") return False, "Le token est vide" if ( len(str(pwd)) <= 0): mycommon.myprint(str(inspect.stack()[0][3]) + " Le mot de passe est vide") return False, "Le mot de passe est vide" if ( len(str(new_pwd)) <= 0): mycommon.myprint(str(inspect.stack()[0][3]) + " Le nouveau mot de passe est vide") return False, "Le nouveau mot de passe est vide" if (str(new_pwd) != str(conf_new_pwd)): mycommon.myprint(str(inspect.stack()[0][3]) + " - Les mots de passe ne sont pas identiques. nouveau pass :'"+str(new_pwd) +"', la oonfirmation envoyées est "''+str(conf_new_pwd)+"' ") return False, " Les mots de passe ne sont pas identiques " """ 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'}) mydata['date_update'] = str(datetime.now()) coll_name = MYSY_GV.dbname['user_account'] #print(" le recid = "+user_recid+" mydata = "+str(mydata)) ret_val = coll_name.find_one_and_update({'apprenant_id': str(diction['apprenant_id']), 'locked': '0', 'active':'1', 'pwd':pwd}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) if (ret_val and ret_val['_id']): mycommon.myprint(str(inspect.stack()[0][3]) + " Le mot de passe de l'utilisateur (apprenant_id) =" + str(diction['apprenant_id'])+" a été modifié") return True, "Le mot de passe a bien été mise à jour " else: mycommon.myprint(str(inspect.stack()[0][3]) + " Impossible de mettre à jour le password du user : apprenant_id = "+str(diction['apprenant_id'])) return False, "Impossible de mettre à jour le password" 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 mettre à jour le mot de passe"