""" Ce fichier gere les sessions de formation """ import pymongo from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import prj_common as mycommon import secrets import inspect import sys, os import csv import pandas as pd from pymongo import ReturnDocument import GlobalVariable as MYSY_GV from math import isnan import GlobalVariable as MYSY_GV from datetime import timedelta from datetime import timedelta """ Fonction de creation et mise à jour d'une session de formation. la clée est : - formation_session_id, - class_internal_url """ def Add_Update_SessionFormation(diction): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token', 'date_debut', 'date_fin', 'nb_participant', 'formation_session_id', 'adresse', 'code_postal', 'ville', 'class_internal_url', 'session_status', 'date_debut_inscription', 'date_fin_inscription'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Creation partner account : Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, "Impossible de créer le stagiaire. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['token', 'formation_session_id', 'class_internal_url',] 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 liste ") return False, "Impossible de créer le stagiaire, Toutes les informations necessaires n'ont pas été fournies" mydata = {} query_key = {} my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] class_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: mydata['class_internal_url'] = diction['class_internal_url'] query_key['class_internal_url'] = diction['class_internal_url'] class_internal_url = diction['class_internal_url'] partner_recid = mycommon.get_parnter_recid_from_token(my_token) if (partner_recid is False): mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de créer la liste d'emargement ") return False, "Impossible de créer la liste d'emargement " formation_session_id = "" if ("formation_session_id" in diction.keys()): if diction['formation_session_id']: mydata['formation_session_id'] = diction['formation_session_id'] query_key['formation_session_id'] = diction['formation_session_id'] session_id = diction['formation_session_id'] if ("date_debut" in diction.keys()): if diction['date_debut']: mydata['date_debut'] = diction['date_debut'] if ("date_fin" in diction.keys()): if diction['date_fin']: mydata['date_fin'] = diction['date_fin'] if ("nb_participant" in diction.keys()): if diction['nb_participant']: mydata['nb_participant'] = diction['nb_participant'] if ("adresse" in diction.keys()): if diction['adresse']: mydata['adresse'] = diction['adresse'] if ("code_postal" in diction.keys()): if diction['code_postal']: ## Verifier si le CP n'est composé que de "0", au quel cas il s'agit d'une formation en ligne. local_cp = str(diction['code_postal']).replace("0","").strip() print(" #### local_cp = "+str(local_cp)) if( local_cp == ""): mydata['code_postal'] = "0" else: mydata['code_postal'] = diction['code_postal'] if ("ville" in diction.keys()): if diction['ville']: mydata['ville'] = diction['ville'] if ("session_status" in diction.keys()): if diction['session_status']: mydata['session_status'] = diction['session_status'] if ("date_debut_inscription" in diction.keys()): if diction['date_debut_inscription']: mydata['date_debut_inscription'] = diction['date_debut_inscription'] if ("date_fin_inscription" in diction.keys()): if diction['date_fin_inscription']: mydata['date_fin_inscription'] = diction['date_fin_inscription'] mydata['date_update'] = str(datetime.now()) mydata['valide'] = "1" coll_name = MYSY_GV.dbname['session_formation'] ret_val = coll_name.find_one_and_update( query_key, {"$set": mydata}, upsert=True, return_document=ReturnDocument.AFTER ) if (ret_val['_id'] is False): mycommon.myprint( str(inspect.stack()[0][3]) + " : Impossible d'ajouter le mode de payement") return False, "Impossible d'ajouter le mode de payement" return True, " La session de formation à bien été crééé" 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 créer la session de formation" """ Fonction de recuperation d'une session de formation """ def GetSessionFormation(diction): try: field_list = ['token', 'class_internal_url', 'formation_session_id', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, "de recuperer la liste des stagiaires . Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['token', 'class_internal_url', 'formation_session_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans liste ") return False, "Impossible de recuperer la liste des stagiaires, Toutes les informations necessaires n'ont pas été fournies" # Recuperation du recid du partner mydata = {} mytoken = "" if ("token" in diction.keys()): if diction['token']: mytoken = diction['token'] partner_recid = mycommon.get_parnter_recid_from_token(mytoken) if (partner_recid is False): mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de recuperer la liste des stagiaires, ") return False, "Impossible de recuperer la liste des stagiaires, Les informations d'identification sont incorrectes " class_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: class_internal_url = diction['class_internal_url'] formation_session_id = "" if ("formation_session_id" in diction.keys()): if diction['formation_session_id']: formation_session_id = diction['formation_session_id'] coll_session = MYSY_GV.dbname['session_formation'] myquery = {} myquery['formation_session_id'] = formation_session_id myquery['class_internal_url'] = class_internal_url #print(" ##### myquery = "+str(myquery)) RetObject = [] for retval in coll_session.find(myquery): #print(" ##### retval = " + str(retval)) RetObject.append(mycommon.JSONEncoder().encode(retval)) 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 session de formation" """ Cette fonction recupere la liste de toutes les sessions de formation actives et valide d'une formation données. """ def GetActiveSessionFormation_List(diction): try: field_list = ['class_internal_url'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, " Impossible de recuperer la liste des session de formation" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = [ 'class_internal_url', ] 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 liste ") return False, "Impossible de recuperer la liste des session de formation" # Recuperation du recid du partner mydata = {} class_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: class_internal_url = diction['class_internal_url'] coll_session = MYSY_GV.dbname['session_formation'] myquery = {} myquery['class_internal_url'] = class_internal_url myquery['valide'] = "1" myquery['session_status'] = "true" RetObject = [] for retval in coll_session.find(myquery): local_tmp = retval ## Verification des conditions supplementaires : # en realité c'est parce que c'est chaud de le faire avec pymongo tmp_debut_inscritp = str(local_tmp['date_debut_inscription']).strip().split(" ") tmp_fin_inscritp = str(local_tmp['date_fin_inscription']).strip().split(" ") debut_inscr = datetime.strptime(str(tmp_debut_inscritp[0]).strip(), '%d/%m/%Y') fin_inscr = datetime.strptime(str(tmp_fin_inscritp[0]).strip(), '%d/%m/%Y') #print(" #### date_debut_inscription = ", debut_inscr, " date_fin_inscription = ", fin_inscr, " NOW = ",datetime.now()) if(debut_inscr <= datetime.now() and fin_inscr >= datetime.now()): RetObject.append(mycommon.JSONEncoder().encode(retval)) #print("####RetObject = "+str(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 sessions de formation valides et actives." """ Cette fonction recuperer la liste des toutes les sessions de formation valides pour une formation données. Qu'elles soient cloturées ou pas.""" def GetAllValideSessionFormation_List(diction): try: field_list = ['class_internal_url'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, " Impossible de recuperer la liste des session de formation" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['class_internal_url', ] 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 liste ") return False, "Impossible de recuperer la liste des session de formation" # Recuperation du recid du partner mydata = {} class_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: class_internal_url = diction['class_internal_url'] coll_session = MYSY_GV.dbname['session_formation'] myquery = {} myquery['class_internal_url'] = class_internal_url myquery['valide'] = "1" #print(" ##### myquery = "+str(myquery)) RetObject = [] for retval in coll_session.find(myquery): #print(" ##### retval = " + str(retval)) RetObject.append(mycommon.JSONEncoder().encode(retval)) 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 sessions de formation valides et actives."