""" -- Gestion de l'espace ENT des apprenants Ce fichier permet de gerer les fonction de ENT ETUDIANT """ import bson import pymongo from dns.e164 import query 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 from datetime import datetime from zoneinfo import ZoneInfo """ Important : Seules session avec 'session_status'= '1' (actif sont pris en compte) 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'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'}) 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 = {} if( "session_id" in retVal.keys() ): is_active_session_count = MYSY_GV.dbname['session_formation'].count_documents({'_id':ObjectId(str(retVal['session_id'])), 'valide':'1', 'partner_owner_recid':str(retVal['partner_owner_recid']), 'session_status':'1'}) if( is_active_session_count == 1): 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'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'}) 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', 'session_status':'1', 'partner_owner_recid':str(class_data['partner_owner_recid'])}) if( sess_data ): 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 " """ Recuperation des session / promotion d'un apprenant pour son espace ENT """ def Get_Ent_Student_List_Session_Promotion(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'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'}) 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, 'title':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 = {} if( "session_id" in retVal.keys() ): is_active_session_count = MYSY_GV.dbname['session_formation'].count_documents({'_id':ObjectId(str(retVal['session_id'])), 'valide':'1', 'partner_owner_recid':str(retVal['partner_owner_recid']), 'session_status':'1'}) if( is_active_session_count == 1): session_data = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(str(retVal['session_id'])), 'valide': '1', 'partner_owner_recid': str(retVal['partner_owner_recid']), 'session_status': '1'}, ) local_node = {} local_node['id'] = str(cpt) local_node['inscription_id'] = (str(retVal['_id'])) local_node['session_id'] = (str(retVal['session_id'])) local_node['session_code_session'] = (str(session_data['code_session'])) local_node['session_titre'] = (str(session_data['titre'])) local_node['session_date_du'] = (str(session_data['date_debut'])) local_node['session_date_au'] = (str(session_data['date_fin'])) #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'])) local_node['class_title'] = (str(retVal['myclass_collection'][0]['title'])) RetObject.append(mycommon.JSONEncoder().encode(local_node)) cpt = cpt + 1 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 liste des promotion et sessions 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 A cela on va associer le planing personnel de l'etudiant pour recuperer par exemple : - ses evaluation - ses convocations Ceci si et seulement si l'inscription_id est fournie """ 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', 'inscription_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 filt_session = {} if( 'session_id' in diction.keys() and diction['session_id']): filt_session = {'session_id':str(diction['session_id'])} """ 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'}) if ("inscription_id" in diction.keys() and diction['inscription_id'] and "session_id" in diction.keys() and diction['session_id']): is_valide_inscription_count = MYSY_GV.dbname['inscription'].count_documents({'session_id':str(diction['session_id']), '_id':ObjectId(str(diction['inscription_id'])), 'valide':'1'}) if( is_valide_inscription_count != 1 ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de l'inscription est invalide ") return False, " L'identifiant de l'inscription est invalide ", RetObject = [] val_tmp = 0 nb_sequence = 0 for New_retVal in MYSY_GV.dbname['session_formation_sequence'].find({'$and':[ {'partner_owner_recid':str(apprenant_data['partner_owner_recid']), 'valide':'1', 'locked':'0', 'confirmed':'1'}, filt_session]}): 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 ("obligatoire" not in user.keys()): user['obligatoire'] = "0" if ("volume_theorique" not in user.keys()): user['volume_theorique'] = "" if( "grp_apprenant_id" not in user.keys() ): user['grp_apprenant_id'] = "" grp_apprenant_color = "" grp_apprenant_code = "" """ Recuperer la couleur du groupe """ grp_apprenant_id = "" 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 user['grp_apprenant_id'] = grp_apprenant_id liste_groupe_code = "" """ Recuperer la couleur du groupe 21/05/2026 : il faut aller chercher la config général qui dit si le code couleur se fait par groupe d'apprenant ou par couleur du BC / UE """ sequence_code_couleur_setup = "ue" base_partner_setup_code_couleur = MYSY_GV.dbname['base_partner_setup'].find_one( {'config_name': 'sequence_code_couleur', 'partner_owner_recid': str(my_partner['partner_owner_recid'])}) if (base_partner_setup_code_couleur and "config_value" in base_partner_setup_code_couleur.keys() and base_partner_setup_code_couleur['config_value'] in ['ue', 'grp_apprenant', '']): sequence_code_couleur_setup = base_partner_setup_code_couleur['config_value'] domaine_competence_color = "#3573CB" if (sequence_code_couleur_setup == "ue"): if ("domaine_competence_id" in New_retVal.keys() and New_retVal['domaine_competence_id']): domaine_competence_id_data = MYSY_GV.dbname['domaine_competence_pedagogique'].find_one( {'_id': ObjectId(str(New_retVal['domaine_competence_id'])), 'partner_owner_recid': str(my_partner['partner_owner_recid'])}, {'couleur': 1}) if (domaine_competence_id_data and 'couleur' in domaine_competence_id_data.keys()): domaine_competence_color = domaine_competence_id_data['couleur'] user['domaine_competence_color'] = domaine_competence_color user['sequence_code_couleur_setup'] = sequence_code_couleur_setup tab_grp_apprenant_ids = [] if ("tab_grp_apprenant_ids" in user.keys() and user['tab_grp_apprenant_ids']): for local_grp_apprenant in user['tab_grp_apprenant_ids']: groupe_data = MYSY_GV.dbname['groupe_inscription'].find_one( {'_id': ObjectId(str(local_grp_apprenant['_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['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']): liste_groupe_code = str(groupe_data['code']) + ", " + str(liste_groupe_code) grp_apprenant_code = groupe_data['code'] if (groupe_data and '_id' in groupe_data.keys()): tab_grp_apprenant_ids.append(str(groupe_data['_id'])) user['grp_apprenant_color'] = grp_apprenant_color user['grp_apprenant_code'] = grp_apprenant_code user['liste_groupe_code'] = liste_groupe_code if ("mode_animation" not in user.keys()): user['mode_animation'] = "0" ue_titre = "" ue_code = "" eu_id = "" if ("ue_id" not in user.keys()): user['ue_id'] = "" elif ("ue_id" in user.keys() and user['ue_id'] and "session_id" in diction.keys()): """ 16/05/2026 - Verifier que cette UE est toujours rattachée à la session concernée Important : ca permet de gerer les eventuelle suppression d'UE dans d'une session """ is_valide_UE_in_Session_count = MYSY_GV.dbname['session_formation'].count_documents( {'list_unite_enseignement._id': str(user['ue_id']), '_id': ObjectId(str(diction['session_id']))}) if (is_valide_UE_in_Session_count > 0): eu_id_data = MYSY_GV.dbname['unite_enseignement'].find_one({'_id': ObjectId(str(user['ue_id'])), 'partner_owner_recid': str( my_partner['partner_owner_recid'])}) if (eu_id_data and "titre" in eu_id_data.keys()): ue_titre = eu_id_data['titre'] if (eu_id_data and "code" in eu_id_data.keys()): ue_code = eu_id_data['code'] if (eu_id_data and "_id" in eu_id_data.keys()): eu_id = str(eu_id_data['_id']) else: ue_titre = "" ue_code = "" eu_id = "" user['ue_code'] = ue_code user['ue_titre'] = ue_titre user['ue_id'] = eu_id if ("unite_enseignement_planif_id" not in user.keys()): user['unite_enseignement_planif_id'] = "" """ Recuperer les ressource associées (materiel et humaine) """ tab_ressource_humaine = [] tab_ressource_materiel = [] for local_affectation in MYSY_GV.dbname['session_formation_sequence_affectation'].find( {'sequence_session_id': str(New_retVal['_id']), 'partner_owner_recid':str(apprenant_data['partner_owner_recid'])}): related_target_collection_id_nom = "" related_target_collection_object = "" # Si l'affectation a un 'related_target_collection_id', alors cela veut dire qu'il faut aller chercheer # la cible de cette affection. if ("related_target_collection_id" in local_affectation.keys() and "related_target_collection" in local_affectation.keys()): if (local_affectation["related_target_collection_id"] and local_affectation[ "related_target_collection"]): # Si l'affectation concerne une ressource_humaine if (local_affectation["related_target_collection"] == "ressource_humaine"): affectation_target_data = MYSY_GV.dbname[ local_affectation["related_target_collection"]].find_one( {"_id": ObjectId(str(local_affectation["related_target_collection_id"])), 'partner_recid': str(apprenant_data['partner_owner_recid']), 'valide': '1', 'locked': "0"}) if (affectation_target_data is not None): related_target_collection_id_nom = affectation_target_data["nom"] + " " + \ affectation_target_data["prenom"] related_target_collection_object = "Ressource Humaine" tab_ressource_humaine.append(related_target_collection_id_nom) elif (local_affectation["related_target_collection"] == "ressource_materielle"): # Si l'affectation concerne une ressource_materielle affectation_target_data = MYSY_GV.dbname[ local_affectation["related_target_collection"]].find_one( {"_id": ObjectId(str(local_affectation["related_target_collection_id"])), 'partner_recid': str(apprenant_data['partner_owner_recid']), 'valide': '1'}) if (affectation_target_data is not None): related_target_collection_id_nom = affectation_target_data["nom"] related_target_collection_object = "Ressource Materielle" tab_ressource_materiel.append(related_target_collection_id_nom) user['tab_ressource_humaine'] = tab_ressource_humaine user['tab_ressource_materiel'] = tab_ressource_materiel RetObject.append(mycommon.JSONEncoder().encode(user)) """ On recuperer les évaluation de l'etudiant """ if ("inscription_id" in diction.keys() and diction['inscription_id'] and "session_id" in diction.keys() and diction['session_id']): inscription_data = MYSY_GV.dbname['inscription'].find_one({'session_id':str(diction['session_id']), '_id':ObjectId(str(diction['inscription_id'])), 'valide':'1'}) for New_retVal in MYSY_GV.dbname['agenda'].find({'partner_owner_recid': str(apprenant_data['partner_owner_recid']), 'valide': '1', 'locked': '0', 'related_collection':'inscription', 'related_collection_recid':diction['inscription_id']}): user = {} user['id'] = str(val_tmp) val_tmp = val_tmp + 1 nb_sequence = nb_sequence + 1 if ("event_start" not in New_retVal.keys()): user['sequence_start'] = "" else: user['sequence_start'] = New_retVal['event_start'] if ("event_end" not in New_retVal.keys()): user['sequence_end'] = "" else: user['sequence_end'] = New_retVal['event_end'] if ("event_title" not in New_retVal.keys()): user['sequence_title'] = "" else: user['sequence_title'] = New_retVal['event_title'] if ("event_type" not in New_retVal.keys()): user['type'] = "" else: user['type'] = New_retVal['event_type'] if ("color" not in New_retVal.keys()): user['color'] = "" else: user['color'] = New_retVal['color'] if ("color" not in New_retVal.keys()): user['grp_apprenant_color'] = "" else: user['grp_apprenant_color'] = New_retVal['color'] if ("mode_animation" not in New_retVal.keys()): user['mode_animation'] = "0" if ("ue_id" not in New_retVal.keys()): user['ue_id'] = "" if ("unite_enseignement_planif_id" not in New_retVal.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() mycommon.myprint(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() mycommon.myprint(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']), 'publish':'1' }}, {'$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) instant_t = datetime.now() pipeline = [ { "$match": { "inscription_id": { "$in": tab_my_inscription_ids }, "partner_owner_recid": str(apprenant_data['partner_owner_recid']), "publish": "1", "is_validated": "1" } }, { "$addFields": { "mysy_publish_date": { "$dateFromString": { "dateString": "$publish_date", "format": "%d/%m/%Y %H:%M", "onError": None, "onNull": None } } } }, { "$match": { "mysy_publish_date": { "$lte": instant_t } } }, { "$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": None, "onNull": None } } ] } ] } } } ], "as": "note_evaluation_collection" } } ] print("#### Get_List_Participant_Notes pipeline laa 01 : query = ", pipeline) for retval in MYSY_GV.dbname['note_evaluation_participant'].aggregate(pipeline): if( "note_evaluation_collection" in retval.keys() ): print(" ## ii retval['note_evaluation_collection'] = ", retval['note_evaluation_collection']) 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'est pas autorisé ") 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', '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 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', ) ne sont plus obligatories, mais au moins une doit etre fourni """ if (str(object_owner_collection).strip() == "" ): mycommon.myprint( str(inspect.stack()[0][3]) + " Il manque des informations obligatoires ") return False, " Les informations fournies sont incorrectes", """ 01/03/2026 Si aucun detail détail n'est donnée sur l'inscription_id, on va aller chercher tous les documents de l'apprenant_id """ tab_inscription_id = [] if( str(object_owner_id).strip() == "" ): for tmp in MYSY_GV.dbname['inscription'].find({'apprenant_id':str(diction['apprenant_id']), 'valide':'1', 'partner_owner_recid':my_partner['partner_owner_recid']}): tab_inscription_id.append(str(tmp['_id'])) else: tab_inscription_id.append(str(object_owner_id)) my_query = {} if (object_owner_collection): my_query['object_owner_collection'] = object_owner_collection if (object_owner_id): my_query['object_owner_id'] = {'$in':tab_inscription_id} my_query['valide'] = "1" my_query['partner_owner_recid'] = str(apprenant_data['partner_owner_recid']) query = my_query print(" ### query = ", query) RetObject = [] nb_val = 0 for retval in MYSY_GV.dbname['download_files'].find(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" if ("liste_ue_code" in retval.keys()): ret_file['liste_ue_code'] = retval['liste_ue_code'] else: ret_file['liste_ue_code'] = "" """ 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 liste_ue_code = "" """ Recuperation des EU eventuellement ajouté à 'Get_Ent_Student_List_object_owner_collection_Stored_Files' """ if( "tab_related_collection" in retval.keys() ): for single_related_col in retval['tab_related_collection']: if( "related_collection" in single_related_col.keys() and single_related_col['related_collection'] == "unite_enseignement"): ue_data = MYSY_GV.dbname['unite_enseignement'].find_one({'_id':ObjectId(single_related_col['related_collection_id']), 'valide':'1', 'partner_owner_recid':str(apprenant_data['partner_owner_recid'])}) if( ue_data and "code" in ue_data.keys()): liste_ue_code = liste_ue_code + str(ue_data['code'])+", " ret_file['liste_ue_code'] = liste_ue_code """ Recuperation des session eventuellement ajouté à 'Get_Ent_Student_List_object_owner_collection_Stored_Files' """ liste_session_code = "" if ("tab_related_collection" in retval.keys()): for single_related_col in retval['tab_related_collection']: if ("related_collection" in single_related_col.keys() and single_related_col[ 'related_collection'] == "session_formation"): session_data = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(single_related_col['related_collection_id']), 'valide': '1', 'partner_owner_recid': str(apprenant_data['partner_owner_recid'])}, {'code_session':1}) if (session_data and "code_session" in session_data.keys()): liste_session_code = liste_session_code + str(session_data['code_session']) + ", " ret_file['liste_session_code'] = liste_session_code """ Recuperation des session eventuellement ajouté à 'Get_Ent_Student_List_object_owner_collection_Stored_Files' """ class_title = "" class_id = "" if ("tab_related_collection" in retval.keys()): for single_related_col in retval['tab_related_collection']: if ("related_collection" in single_related_col.keys() and single_related_col[ 'related_collection'] == "myclass"): class_data = MYSY_GV.dbname['myclass'].find_one( {'_id': ObjectId(single_related_col['related_collection_id']), 'valide': '1', 'partner_owner_recid': str(apprenant_data['partner_owner_recid'])}, {'title': 1, '_id':1}) if (class_data and "title" in class_data.keys()): class_title = str(class_data['title']) class_id = str(class_data['_id']) ret_file['class_title'] = class_title ret_file['class_id'] = class_id 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', 'type_connexion', 'rh_id', 'date_naissance', 'num_cvec', 'civilite', 'nom_fille', 'date_naissance'] 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', 'type_connexion'] 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)) global_apprenant_id = "" global_apprenant_partner_owner_recid = "" global_apprenant_email = "" if( diction['type_connexion'] == "apprenant"): """ 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 ", global_apprenant_id = str(diction['apprenant_id']) apprenant_data = MYSY_GV.dbname['apprenant'].find_one( {'_id': ObjectId(str(diction['apprenant_id'])), 'valide': '1', 'locked': '0'}) global_apprenant_partner_owner_recid = str(apprenant_data['partner_owner_recid']) global_apprenant_email = str(apprenant_data['email']) elif( diction['type_connexion'] == "tuteur"): """ Verifier la validité de rh_id """ is_ressource_humaine_valide_count = MYSY_GV.dbname['ressource_humaine'].count_documents( {'_id': ObjectId(str(diction['rh_id'])), 'valide': '1', 'locked': '0'}) if (is_ressource_humaine_valide_count != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant du tuteur est invalide ") return False, " L'identifiant du tuteur est invalide ", ressource_humaine_data = MYSY_GV.dbname['ressource_humaine'].find_one( {'_id': ObjectId(str(diction['rh_id'])), 'valide': '1', 'locked': '0'}) """ 22/05/2026 - On va aussi mettre à jour le dossier apprenant """ dictionnary_for_apprenant = {} coll_name = MYSY_GV.dbname['user_account'] now = datetime.now() dictionnary = {} dictionnary['active'] = "1" dictionnary['update_date'] = str(now) mytoken ="" myemail = "" if ("adr_street" in diction.keys()): dictionnary['adr_street'] = str(diction['adr_street']) dictionnary_for_apprenant['adresse'] = str(diction['adr_street']) if ("adr_city" in diction.keys()): dictionnary['adr_city'] = str(diction['adr_city']) dictionnary_for_apprenant['ville'] = str(diction['adr_city']) if( "adr_zip" in diction.keys() ): dictionnary['adr_zip'] = str(diction['adr_zip']) dictionnary_for_apprenant['code_postal'] = str(diction['adr_zip']) if ("adr_country" in diction.keys()): dictionnary['adr_country'] = str(diction['adr_country']) dictionnary_for_apprenant['pays'] = str(diction['adr_country']) if ("mob_phone" in diction.keys()): dictionnary['mob_phone'] = str(diction['mob_phone']) dictionnary_for_apprenant['telephone'] = str(diction['mob_phone']) if ("link_facebook" in diction.keys()): dictionnary['link_facebook'] = str(diction['link_facebook']) dictionnary_for_apprenant['link_facebook'] = str(diction['link_facebook']) if ("link_twitter" in diction.keys()): dictionnary['link_twitter'] = str(diction['link_twitter']) dictionnary_for_apprenant['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()): dictionnary['prenom'] = str(diction['prenom']) dictionnary_for_apprenant['prenom'] = str(diction['prenom']) if ("nom" in diction.keys()): dictionnary['nom'] = str(diction['nom']) dictionnary_for_apprenant['nom'] = str(diction['nom']) if ("date_naissance" in diction.keys()): dictionnary['birthday'] = str(diction['date_naissance']) dictionnary_for_apprenant['date_naissance'] = str(diction['date_naissance']) if ("link_linkedin" in diction.keys()): dictionnary['link_linkedin'] = str(diction['link_linkedin']) dictionnary_for_apprenant['link_linkedin'] = str(diction['link_linkedin']) if ("num_cvec" in diction.keys()): dictionnary_for_apprenant['num_cvec'] = str(diction['num_cvec']) if ("civilite" in diction.keys()): dictionnary['civilite'] = str(diction['civilite']) dictionnary_for_apprenant['civilite'] = str(diction['civilite']) if ("num_cvec" in diction.keys()): dictionnary_for_apprenant['num_cvec'] = str(diction['num_cvec']) if ("nom_fille" in diction.keys()): dictionnary['nom_fille'] = str(diction['nom_fille']) dictionnary_for_apprenant['nom_fille'] = str(diction['nom_fille']) 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' """ is_token_valide_for_one_user = MYSY_GV.dbname['user_account'].count_documents({'token':str(diction['token']), 'active':'1', 'locked':'0', }) 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 " if( diction['type_connexion'] == "tuteur"): ret_val = coll_name.find_one_and_update({'token':str(diction['token']), 'active':'1', 'locked':'0', 'rh_id': diction['rh_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" elif (diction['type_connexion'] == "apprenant"): 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'] and len(str(global_apprenant_id)) > 5 ): """ 22/05/2026 - Mise à jour de l'apprenant, si c'est un apprenant qui est connecté IMPORTANT : On va mettre à jour toutes les inscriptions dont date_du <= TODAY_DAYE <= date_au. L'idée des recuperer toutes les inscrition dont les sessiosn sont valides """ pipeline_valide_inscription_by_date_session = [ { "$match": { "partner_owner_recid": str(global_apprenant_partner_owner_recid), "apprenant_id": str(global_apprenant_id), "email": str(global_apprenant_email), "valide": "1", "locked": "0" } }, { "$lookup": { "from": "session_formation", "let": { "local_session_id": "$session_id", "local_partner_owner_recid": "$partner_owner_recid" }, "pipeline": [ { "$match": { "$expr": { "$and": [ { "$eq": [ "$_id", { "$convert": { "input": "$$local_session_id", "to": "objectId", "onError": None, "onNull": None } } ] }, { "$eq": [ "$partner_owner_recid", "$$local_partner_owner_recid" ] }, { "$eq": ["$valide", "1"] }, { "$eq": ["$archive", "0"] } ] } } }, { "$addFields": { "mysy_date_debut": { "$dateFromString": { "dateString": "$date_debut", "format": "%d/%m/%Y", "onError": None, "onNull": None } }, "mysy_date_fin": { "$dateFromString": { "dateString": "$date_fin", "format": "%d/%m/%Y", "onError": None, "onNull": None } }, "mysy_today": { "$dateTrunc": { "date": "$$NOW", "unit": "day" } } } }, { "$match": { "$expr": { "$and": [ { "$lte": [ "$mysy_date_debut", "$mysy_today" ] }, { "$gte": [ "$mysy_date_fin", "$mysy_today" ] } ] } } }, { "$project": { "_id": 1, "code_session": 1, "titre": 1, "date_debut": 1, "date_fin": 1, "session_status": 1, "session_etape": 1 } } ], "as": "session_formation_collection" } }, { "$match": { "session_formation_collection.0": { "$exists": True } } }, { "$addFields": { "session_formation": { "$arrayElemAt": [ "$session_formation_collection", 0 ] } } }, { "$project": { "_id": 1 } }, { "$sort": { "_id": -1 } } ] tab_valide_inscription_ObjectId = [] for local_inscr in MYSY_GV.dbname["inscription"].aggregate(pipeline_valide_inscription_by_date_session): tab_valide_inscription_ObjectId.append(ObjectId(str(local_inscr['_id']))) dictionnary_for_apprenant['update_date'] = str(datetime.now()) dictionnary_for_apprenant['update_by'] = str(my_partner['_id']) print(" ### dictionnary_for_apprenant = ", dictionnary_for_apprenant) result_apprenant = MYSY_GV.dbname['apprenant'].find_one_and_update( {'_id': ObjectId(str(global_apprenant_id)), 'partner_owner_recid': global_apprenant_partner_owner_recid}, {"$set": dictionnary_for_apprenant}, upsert=False, return_document=ReturnDocument.AFTER ) result_inscrit = MYSY_GV.dbname['inscription'].update_many( {'_id':{'$in':tab_valide_inscription_ObjectId}, 'partner_owner_recid': global_apprenant_partner_owner_recid}, {"$set": dictionnary_for_apprenant}, ) print(" result_inscrit update = ", result_inscrit.modified_count) 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['update_date'] = 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" """ 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 """ Cette fonction retourne les données du Footer d'un apprenant_id """ def Get_ENT_Data_From_Subdomain_Light_Data_For_Footer(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'apprenant_id'] """ 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 RetObject = [] val_tmp = 0 for retval in MYSY_GV.dbname['partnair_account'].find({'recid': my_partner['partner_owner_recid'], 'active': '1', 'locked': '0', 'firstconnexion': '0','is_partner_admin_account':'1' }, {'_id':1, 'recid':1, 'nom':1}): user = retval user['id'] = str(val_tmp) val_tmp = val_tmp + 1 """ Recuperer les elements graphique : logo, bannière, etc """ config_partner_catalog_data = MYSY_GV.dbname['config_partner_catalog'].find_one({'valide':'1', 'locked':'0', 'partner_owner_recid':str(retval['recid'])}) user['catalog_pub_cgv_pdf'] = "" if (config_partner_catalog_data and "cgv_pdf" in config_partner_catalog_data.keys()): user['catalog_pub_cgv_pdf'] = config_partner_catalog_data['cgv_pdf'] user['catalog_pub_cgu_pdf'] = "" if (config_partner_catalog_data and "cgu_pdf" in config_partner_catalog_data.keys()): user['catalog_pub_cgu_pdf'] = config_partner_catalog_data['cgu_pdf'] user['catalog_pub_mention_legale_pdf'] = "" if (config_partner_catalog_data and "mention_legale_pdf" in config_partner_catalog_data.keys()): user['catalog_pub_mention_legale_pdf'] = config_partner_catalog_data['mention_legale_pdf'] user['catalog_pub_logo'] = "" #if( config_partner_catalog_data and "logo" in config_partner_catalog_data.keys() ): # user['catalog_pub_logo'] = config_partner_catalog_data['logo'] RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" ### Get_Partner_Data_From_Subdomain 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écuperer les données liées au sous domaine " """ Cette fonction recuprer les absence d'un apprenant avec un filtre sur les inscription """ def Get_Ent_Student_Absence(diction): try: full_file_path = "" diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'apprenant_id', 'session_id', 'inscription_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 = 1 total_heure_absence = 0 print(" ENT ETUDIANT ABSENCE QRY = ", {'related_collection': "apprenant", 'inscription_id': str(diction['inscription_id']), 'event_type':'absence', 'partner_owner_recid': str(my_partner['partner_owner_recid']), 'valide': '1', 'locked': '0'}) for New_retVal in MYSY_GV.dbname['agenda'].find({'related_collection': "apprenant", 'inscription_id': str(diction['inscription_id']), 'event_type':'absence', 'partner_owner_recid': str(my_partner['partner_owner_recid']), 'valide': '1', 'locked': '0'}): user = New_retVal a = datetime.fromisoformat(New_retVal['event_start']) b = datetime.fromisoformat(New_retVal['event_end']) total = (b - a) print(" total = ", total.total_seconds() / 3600) total_heure_absence = total_heure_absence + total.total_seconds() / 3600 user['heure_absence'] = str(total.total_seconds() / 3600) code_session = "" if ("session_id" in New_retVal.keys() and New_retVal['session_id']): local_session_data = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(str(New_retVal['session_id'])), 'partner_owner_recid': str(my_partner['partner_owner_recid'])}, {'_id': 1, 'code_session': 1}) if (local_session_data and "code_session" in local_session_data.keys()): code_session = local_session_data['code_session'] user['code_session'] = code_session user['id'] = str(val_tmp) val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) else: user['session_id'] = "" user['code_session'] = "" user['id'] = str(val_tmp) 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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les absences " """ Cette fonction recuprer le total des heures d'absence d'un apprenant """ def Get_Ent_Student_total_Heure_Absence(diction): try: full_file_path = "" diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'apprenant_id', 'session_id', 'inscription_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 = 1 total_heure_absence = 0 total_heure_absence_justifie = 0 total_heure_absence_non_justifie = 0 print(" ENT ETUDIANT NB HREU ABSENCE QRY = ", {'related_collection': "apprenant", 'inscription_id': str(diction['inscription_id']), 'event_type': 'absence', 'partner_owner_recid': str(my_partner['partner_owner_recid']), 'valide': '1', 'locked': '0'}) for New_retVal in MYSY_GV.dbname['agenda'].find({'related_collection': "apprenant", 'inscription_id': str(diction['inscription_id']), 'event_type': 'absence', 'partner_owner_recid': str(my_partner['partner_owner_recid']), 'valide': '1', 'locked': '0'}): user = New_retVal a = datetime.fromisoformat(New_retVal['event_start']) b = datetime.fromisoformat(New_retVal['event_end']) total = (b - a) total_heure_absence = total_heure_absence + total.total_seconds() / 3600 if( "justified" in New_retVal.keys() and New_retVal['justified'] == "true"): total_heure_absence_justifie = total_heure_absence_justifie + total.total_seconds() / 3600 else: total_heure_absence_non_justifie = total_heure_absence_non_justifie + total.total_seconds() / 3600 total_heure_absence = round(total_heure_absence, 2 ) total_heure_absence_justifie = round(total_heure_absence_justifie, 2) total_heure_absence_non_justifie = round(total_heure_absence_non_justifie, 2) node_total_absence = {'total_heure_absence':str(total_heure_absence), 'total_heure_absence_justifie':str(total_heure_absence_justifie), 'total_heure_absence_non_justifie':str(total_heure_absence_non_justifie)} RetObject.append(mycommon.JSONEncoder().encode(node_total_absence)) 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 le total des heures d'absences " """ Cette fonction recuperer l'image de profil d'un inscrit en partant de l'ENT """ def Get_Ent_Student_Apprenant_IMG_Profil(diction): try: """ Verification de la liste 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 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 ", qery_images = {'locked': '0', 'valide': '1', 'related_collection': 'apprenant', 'related_collection_recid': str(diction['apprenant_id'])} # print(" ### qery_images = ", qery_images) RetObject = [] partner_images = {} # Recuperation des image 'logo' et 'cachet' si le partenaire en a for retVal in MYSY_GV.dbname['mysy_images'].find(qery_images): if ('type_img' in retVal.keys()): if (retVal['type_img'] == "user"): partner_images['logo_img'] = retVal['img'].decode() partner_images['logo_img_recid'] = retVal['recid'] RetObject.append(mycommon.JSONEncoder().encode(partner_images)) 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 recuperer l'image de profil de l'apprenant" """ Cette fonction permet a un étudiant de declarer une absence depuis son ENT Important : Pour une séquence, un inscrit ne peut demander qu'une seule demande d'absence """ def Create_Ent_Student_Absence(tab_files, Folder, diction): try: full_file_path = "" diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'apprenant_id', 'session_id', 'sequence_id', 'inscription_id', 'raison', 'comment'] 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', 'session_id', 'sequence_id', 'inscription_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', 'partner_owner_recid':str(my_partner['partner_owner_recid'])}) 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 ", """ Verifier la validité de l'inscritpion (en liaison avec l'apprenant) """ is_valide_inscription_count = MYSY_GV.dbname['inscription'].count_documents({'_id':ObjectId(str(diction['inscription_id'])), 'apprenant_id':str(diction['apprenant_id']), 'session_id':str(diction['session_id']), 'partner_owner_recid':str(my_partner['partner_owner_recid'])}) 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 ", inscription_data = MYSY_GV.dbname['inscription'].find_one( {'_id': ObjectId(str(diction['inscription_id'])), 'apprenant_id': str(diction['apprenant_id']), 'session_id': str(diction['session_id']), 'partner_owner_recid': str(my_partner['partner_owner_recid'])}, {'email':1, '_id':1}) """ Verifier la validité de la séquence, en liaison avec la sequence """ is_valide_sequence_count = MYSY_GV.dbname['session_formation_sequence'].count_documents( {'_id': ObjectId(str(diction['sequence_id'])), 'session_id': str(diction['session_id']), 'partner_owner_recid':str(my_partner['partner_owner_recid'])}) if (is_valide_sequence_count != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de la séquence est invalide ") return False, " L'identifiant de la séquence est invalide ", sequence_data = MYSY_GV.dbname['session_formation_sequence'].find_one( {'_id': ObjectId(str(diction['sequence_id'])), 'session_id': str(diction['session_id']), 'partner_owner_recid':str(my_partner['partner_owner_recid'])}) """ Verifier qu'il n'y pas déja une demande d'absence pour cette sequence pour cet inscrit """ exist_demande_absene_inscrit_sequence = MYSY_GV.dbname['agenda'].count_documents({ 'partner_owner_recid':str(my_partner['partner_owner_recid']), 'related_collection':'apprenant', 'inscription_id':diction['inscription_id'], 'sequence_session_id':str(diction['sequence_id'])}) if( exist_demande_absene_inscrit_sequence > 0 ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " Vous avez déjà une demande d’absence pour cette séquence de formation. ") return False, "Vous avez déjà une demande d’absence pour cette séquence de formation. ", """ Verifier la validité de la session de formation """ is_valide_session_count = MYSY_GV.dbname['session_formation'].count_documents( {'_id': ObjectId(str(diction['session_id'])), 'partner_owner_recid': str(my_partner['partner_owner_recid'])}) if (is_valide_session_count != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de la promotion est invalide ") return False, " L'identifiant de la promotion est invalide ", session_date = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(str(diction['session_id'])), 'partner_owner_recid': str(my_partner['partner_owner_recid'])}, {'formateur_id':1, '_id':1}) tab_destinataire_email = [] if(session_date and "formateur_id" in diction.keys() and diction['formateur_id'] ): formateur_data = MYSY_GV.dbname['ressource_humaine'].find_one({'_id':ObjectId(diction['formateur_id'] ), 'partner_recid': str(my_partner['partner_owner_recid'])}, {'email':1, 'nom':1, 'prenom':1}) if( formateur_data and 'email' in formateur_data.keys() and formateur_data['email']): tab_destinataire_email.append(formateur_data['email']) # Aller chercher la boite mail qui admin qui traite ces demandes base_partner_setup_data = MYSY_GV.dbname['base_partner_setup'].find_one({'config_name':'absence_recever_mail', 'partner_owner_recid': str( my_partner['partner_owner_recid']) }) if(base_partner_setup_data and "config_value" in base_partner_setup_data.keys() and base_partner_setup_data['config_value']): list_tmp_adr = str(base_partner_setup_data['config_value']).strip().replace(";", ",").split(",") for tmp in list_tmp_adr: if( tmp ): tab_destinataire_email.append(tmp) """ Sauvegarde des pièce justificative """ # Sauvegarde des fichiers joints depuis le front tab_saved_file_full_path = [] for file in tab_files: status, saved_file_full_path = mycommon.Upload_Save_PDF_IMG_File(file, Folder) if (status is False): mycommon.myprint("Impossible de récupérer correctement le fichier à importer") return False, "Impossible de récupérer correctement le fichier à importer" tab_saved_file_full_path.append(saved_file_full_path) #print(" #### tab_saved_file_full_path = ", tab_saved_file_full_path) # Traitement de l'eventuel fichier joint tab_files_to_attache_to_mail = [] tab_list_file_for_queu_mail_and_internal_mail = [] for file_name_full_path in tab_saved_file_full_path: """status, saved_file = mycommon.Upload_Save_PDF_IMG_File(file, Folder) if (status is False): mycommon.myprint("Impossible de récupérer correctement le fichier à importer") return False, "Impossible de récupérer correctement le fichier à importer" """ file_to_attache_to_mail = MIMEBase('application', "octet-stream") file_to_attache_to_mail.set_payload(open(file_name_full_path, "rb").read()) encoders.encode_base64(file_to_attache_to_mail) file_to_attache_to_mail.add_header('Content-Disposition', 'attachment; filename="{0}"'.format(os.path.basename(file_name_full_path))) new_node = {"attached_file": file_to_attache_to_mail} tab_files_to_attache_to_mail.append(new_node) tab_list_file_for_queu_mail_and_internal_mail.append(file_name_full_path) """ Recuperer le modèle de document """ courrier_data_retval = MYSY_GV.dbname['courrier_template'].find_one( {'ref_interne': "ENT_ABSENCE_CREATION", 'valide': '1', 'locked': '0', 'type_doc':'email', 'partner_owner_recid': str(my_partner['partner_owner_recid'])} ) if( not courrier_data_retval or "contenu_doc" not in courrier_data_retval.keys() or str(courrier_data_retval['contenu_doc']) == ""): mycommon.myprint( str(inspect.stack()[0][ 3]) + " Le modèle de courrier 'ENT_ABSENCE_CREATION' n'est pas correctement configuré ") return False, " Le modèle de courrier 'ENT_ABSENCE_CREATION' n'est pas correctement configuré " """ Recuperation du dictionnaire des info """ new_diction = {} new_diction['token'] = diction['token'] new_diction['list_stagiaire_id'] = [ObjectId(str(diction['inscription_id']))] new_diction['list_session_id'] = [] new_diction['list_class_id'] = [] new_diction['list_client_id'] = [] new_diction['list_apprenant_id'] = [] local_status, local_retval = mycommon.From_ENT_Get_Dictionnary_data_For_Template(new_diction, my_partner) if (local_status is False): return local_status, local_retval absence_dictionnary_data = local_retval """ Aller créer l'absence a proprement dit """ absence_data = {} absence_data['related_collection'] = "apprenant" absence_data['related_collection_recid'] = str(diction['apprenant_id']) absence_data['related_collection_email'] = str(inscription_data['email']) absence_data['inscription_id'] = str(inscription_data['_id']) absence_data['session_id'] = str(diction['session_id']) absence_data['sequence_session_id'] = str(diction['sequence_id']) absence_data['event_type'] = "absence" absence_data['justified'] = "false" absence_data['partner_owner_recid'] = str(my_partner['partner_owner_recid']) absence_data['valide'] = "1" absence_data['locked'] = "0" absence_data['event_title'] = str(sequence_data['sequence_title']) abs_from = str(sequence_data['sequence_start']) abs_from_fromated = datetime.strptime(abs_from, "%d/%m/%Y %H:%M").replace(tzinfo=ZoneInfo("Europe/Paris")).isoformat() abs_to = str(sequence_data['sequence_end']) abs_to_fromated = datetime.strptime(abs_to, "%d/%m/%Y %H:%M").replace(tzinfo=ZoneInfo("Europe/Paris")).isoformat() absence_data['event_start'] = str(abs_from_fromated) absence_data['event_end'] = str(abs_to_fromated) if( "comment" in diction.keys() ): absence_data['comment'] = str(diction['comment'])[0:320] else: absence_data['comment'] = "" if ("raison" in diction.keys()): absence_data['raison'] = str(diction['raison']) else: absence_data['raison'] = "" if( "ue_id" in sequence_data.keys() ): absence_data['ue_id'] = sequence_data['ue_id'] else: absence_data['ue_id'] = "" if ("obligatoire" in sequence_data.keys()): absence_data['obligatoire'] = sequence_data['obligatoire'] else: absence_data['obligatoire'] = "" if ("mode_animation" in sequence_data.keys()): absence_data['obligatoire'] = sequence_data['mode_animation'] else: absence_data['mode_animation'] = "" insert = MYSY_GV.dbname['agenda'].insert_one(absence_data) absence_data['event_start_converted'] = datetime.fromisoformat(absence_data['event_start']).strftime("%d/%m/%Y %H:%M") absence_data['event_end_converted'] = datetime.fromisoformat(absence_data['event_end']).strftime("%d/%m/%Y %H:%M") if ("raison" in absence_data.keys() and absence_data['raison']): absence_data['raison'] = str(diction['raison']).capitalize() absence_data['accept_button'] = MYSY_GV.CLIENT_URL_BASE + "absence/accept/" + str(my_partner['partner_owner_recid']) + "/" + str(insert.inserted_id) + "/" absence_data['refuse_button'] = MYSY_GV.CLIENT_URL_BASE + "absence/refuse/" + str(my_partner['partner_owner_recid']) + "/" + str(insert.inserted_id) + "/" absence_dictionnary_data['absence_data'] = absence_data company_data = { "params": absence_dictionnary_data, } #print(" ### company_data = ", company_data) """ 1 - Creation du PDF """ contenu_doc_Template = jinja2.Template(str(courrier_data_retval['contenu_doc'])) sourceHtml = contenu_doc_Template.render(params=company_data["params"]) html_mime = MIMEText(sourceHtml, 'html') # Creation de l'email à enoyer msg = MIMEMultipart("alternative") # Creation de l'email à enoyer msg = MIMEMultipart("alternative") # Attacher l'eventuelle pièces jointes for myfile in tab_files_to_attache_to_mail: msg.attach(myfile['attached_file']) # Traitement du sujet du mail sujet_mail_Template = jinja2.Template(str(courrier_data_retval['sujet'])) sujetHtml = sujet_mail_Template.render(params=company_data["params"]) # Recuperation des donnes smpt local_stpm_status, partner_SMTP_COUNT_smtpsrv, partner_own_smtp_value, partner_SMTP_COUNT_password, partner_SMTP_COUNT_user, partner_SMTP_COUNT_From_User, partner_SMTP_COUNT_port = mycommon.Get_Partner_SMTP_Param( my_partner['partner_owner_recid']) if (local_stpm_status is False): return local_stpm_status, partner_SMTP_COUNT_smtpsrv msg.attach(html_mime) if (str(partner_own_smtp_value) == "1"): msg['From'] = partner_SMTP_COUNT_From_User else: msg['From'] = str(partner_SMTP_COUNT_From_User) msg['Bcc'] = 'contact@mysy-training.com' msg['Subject'] = sujetHtml toaddrs = ",".join(tab_destinataire_email) msg['to'] = toaddrs #msg['cc'] = str(responsable_compte_email) smtpserver = smtplib.SMTP(partner_SMTP_COUNT_smtpsrv, partner_SMTP_COUNT_port) smtpserver.ehlo() smtpserver.starttls() smtpserver.login(partner_SMTP_COUNT_user, partner_SMTP_COUNT_password) val = smtpserver.send_message(msg) smtpserver.close() print(" Email envoyé icii" + str(val)) return True, "L'absence a été enregistré" 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'enregistrer l'absence" """ Cette fonction sur le clic d'un bouton d'accepter le motif d'une absence important : il n'est pas possible de faire cette action plusieurs fois sur un document. on va donc introduire une notion 'decided':'1' """ def Absence_Accept_Motif_From_Mail_No_Token(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['agenda_id', 'partner_owner_recid' ] 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 = ['agenda_id', 'partner_owner_recid' ] 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" """ Verifier la validité de l'absence """ is_valide_absence_count = MYSY_GV.dbname['agenda'].count_documents({'_id':ObjectId(diction['agenda_id']), 'partner_owner_recid':str(diction['partner_owner_recid']), 'decision_done':{'$ne':'1'}}) if( is_valide_absence_count != 1 ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de l'absence est invalide ") return False, " Document invalide " update_data = {} update_data['update_date'] = str(datetime.now()) update_data['update_by'] = "web_button" update_data['justified'] = "true" update_data['decision_done'] = "1" ret_val2 = MYSY_GV.dbname['agenda'].find_one_and_update( {'_id': ObjectId(diction['agenda_id']), 'partner_owner_recid': str(diction['partner_owner_recid']), 'decision_done': {'$ne': '1'}}, {"$set": update_data}, return_document=ReturnDocument.AFTER, upsert=False, ) return True, "L'absence a été correctement mises à 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'absence " def Absence_Refuse_Motif_From_Mail_No_Token(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['agenda_id', 'partner_owner_recid'] 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 = ['agenda_id', 'partner_owner_recid'] 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" """ Verifier la validité de l'absence """ is_valide_absence_count = MYSY_GV.dbname['agenda'].count_documents({'_id': ObjectId(diction['agenda_id']), 'partner_owner_recid': str( diction['partner_owner_recid']), 'decision_done': {'$ne': '1'}}) if (is_valide_absence_count != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de l'absence est invalide ") return False, " Document invalide " update_data = {} update_data['update_date'] = str(datetime.now()) update_data['update_by'] = "web_button" update_data['justified'] = "false" update_data['decision_done'] = "1" ret_val2 = MYSY_GV.dbname['agenda'].find_one_and_update( {'_id': ObjectId(diction['agenda_id']), 'partner_owner_recid': str(diction['partner_owner_recid']), 'decision_done': {'$ne': '1'}}, {"$set": update_data}, return_document=ReturnDocument.AFTER, upsert=False, ) return True, "L'absence a été correctement mises à 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'absence " """ Recuperation des emails d'un étudiant sur l'espace ENT """ def Get_ENT_Given_Internal_Mail(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'apprenant_id', 'internal_mail_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', 'internal_mail_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 liste_user_id_read = [] RetObject = [] val_tmp = 0 print(" ### data_cle = ", {'_id':ObjectId(str(diction['internal_mail_id'])), 'valide':'1', 'locked':'0'}) for retval in MYSY_GV.dbname['internal_mail'].find({'_id':ObjectId(str(diction['internal_mail_id'])), 'valide':'1', 'locked':'0'}): user = retval user['id'] = str(val_tmp) val_tmp = val_tmp + 1 if( "liste_user_id_read" in retval.keys() ): for user_id in retval['liste_user_id_read']: liste_user_id_read.append(str(user_id)) """ Recuperer les eventuels _id des fichiers joints stocké dans la collection : 'download_files' """ tab_saved_file_id = [] for download_files_data in MYSY_GV.dbname['download_files'].find({'type_document':'email', 'object_owner_collection':'internal_mail', 'object_owner_id':str(retval['_id'])}, {'_id':1, 'file_business_object':1, 'file_name':1},): local_node = {} local_node['download_files_id'] = str(download_files_data['_id']) local_node['download_files_file_business_object'] = str(download_files_data['file_business_object']) local_node['download_files_file_name'] = str(download_files_data['file_name']) tab_saved_file_id.append(local_node) user['tab_saved_file_id'] = tab_saved_file_id RetObject.append(mycommon.JSONEncoder().encode(user)) """ Mettre à jour pour dire que le mail est lu par ce user connecté """ current_user_id = str(my_partner['_id']) if( current_user_id not in liste_user_id_read ): liste_user_id_read.append(str(current_user_id)) updata_data = {} updata_data['liste_user_id_read'] = liste_user_id_read MYSY_GV.dbname['internal_mail'].find_one_and_update( {'_id':ObjectId(str(diction['internal_mail_id'])), 'valide':'1', 'locked':'0'}, {'$set': updata_data}) 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 les données du mail "