""" Ce ficher gerer l'emargement à une session de formation """ import ast import pymongo from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import partners 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 xhtml2pdf import pisa import jinja2 import ftplib import pysftp from flask import send_file """ Creation du tableau d'emargement Au debut de la session, le formateur clique sur un bouton qui va aller créer le tableau d'emargement. la finction prend en argement : - url_formation - session_id """ def CreateTableauEmargement(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 = ['class_internal_url', 'session_id', 'token'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['class_internal_url', 'session_id', 'token'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies" mydata = {} my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] # Verifier la validité du token retval = mycommon.check_partner_token_validity("", my_token) if retval is False: return "Err_Connexion", " La session de connexion n'est pas valide" class_internal_url = "" if ("class_internal_url" in diction.keys()): if 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 " session_id = "" if ("session_id" in diction.keys()): if diction['session_id']: mydata['session_id'] = diction['session_id'] session_id = diction['session_id'] """ Recuperation des info de la session de formation """ date_du = "" date_au = "" ville = "" code_postal = "" adresse = "" qry = {"_id":ObjectId(str(session_id)), "valide": "1"} session_formation_count = MYSY_GV.dbname['session_formation'].count_documents({"_id":ObjectId(str(session_id)), "valide": "1"}) if( session_formation_count <= 0 ): mycommon.myprint( str(inspect.stack()[0][3]) + " - Aucune session de formation valide ") return False, "Aucune session de formation valide " for tmp_val in MYSY_GV.dbname['session_formation'].find({"_id":ObjectId(str(session_id)), "valide": "1"}): date_du = str(tmp_val['date_debut'])[0:10] date_au = str(tmp_val['date_fin'])[0:10] ville = "" if ("ville" in tmp_val.keys() and tmp_val['ville']): ville = tmp_val['ville'] code_postal = "" if ("code_postal" in tmp_val.keys() and tmp_val['code_postal']): code_postal = tmp_val['code_postal'] adresse = "" if ("adresse" in tmp_val.keys() and tmp_val['adresse']): adresse = tmp_val['adresse'] is_date_du = mycommon.CheckisDate(date_du) is_date_au = mycommon.CheckisDate(date_au) if( is_date_du is False or is_date_au is False): mycommon.myprint( str(inspect.stack()[0][3]) + " - Impossible de créer la liste d'emargement, Les dates de la session sont incorrectes ") return False, " Impossible de créer la liste d'emargement, Les dates de la session sont incorrectes " date_emargement = datetime.strptime(str(date_du), '%d/%m/%Y') i = 0 # compteur de securité # On vide la collection emargement pour la session = str(session_id) local_myquery = {"session_id": str(session_id)} x = MYSY_GV.dbname['emargement'].delete_many(local_myquery) mycommon.myprint( str(inspect.stack()[0][ 3]) + " - INFO : "+str(x.deleted_count)+" Documents supprimés de la collection : emargement : session = "+str(session_id) ) liste_date = [] #print(" Date du = "+str(date_du)+" CMP Date au ="+str(date_au)) while( date_emargement < datetime.strptime(str(date_au), '%d/%m/%Y') and i < 360): date_10 = str(date_emargement) liste_date.append(date_10[0:10]) i = i + 1 date_emargement = date_emargement + timedelta(days=1) #print(" ### date_emargement = "+str(date_emargement) + " < "+str(datetime.strptime(str(date_au), '%d/%m/%Y') )) #print("#### Liste Date = "+str(liste_date)) coll_emargement = MYSY_GV.dbname['emargement'] #Recuperation de la liste des participants dont l'inscription est validée for tmp_val in MYSY_GV.dbname['inscription'].find({'session_id':str(session_id), 'class_internal_url':str(class_internal_url), 'status':'1'} ): my_local_data = {} my_local_data['email'] = tmp_val['email'] my_local_data['nom'] = tmp_val['nom'] my_local_data['prenom'] = tmp_val['prenom'] my_local_data['session_id'] = tmp_val['session_id'] my_local_data['class_internal_url'] = tmp_val['class_internal_url'] my_local_data['date_update'] = str(datetime.now()) for local_date in liste_date : query_key = {} query_key['email'] = tmp_val['email'] query_key['session_id'] = tmp_val['session_id'] query_key['class_internal_url'] = tmp_val['class_internal_url'] query_key['date'] = str(local_date) new_my_local_data = {} new_my_local_data['email'] = tmp_val['email'] new_my_local_data['nom'] = tmp_val['nom'] new_my_local_data['prenom'] = tmp_val['prenom'] new_my_local_data['session_id'] = tmp_val['session_id'] new_my_local_data['class_internal_url'] = tmp_val['class_internal_url'] new_my_local_data['date_update'] = str(datetime.now()) new_my_local_data['date'] = str(local_date) new_my_local_data['matin'] = False new_my_local_data['apresmidi'] = False #print("### emargement insertion de : "+str(new_my_local_data)) local_ret_val = coll_emargement.find_one_and_update( query_key, {"$set": new_my_local_data}, upsert=True, return_document=ReturnDocument.AFTER ) return True, "ok" 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 le tableau d'emargement" """ 23/11/2023 : Evolution en prenant les sequences. Evolution de la fonction de création du tableau d'emargement. Dans cette evolution, on plutot recuperer les sequence déjà créesq 31/01/2024 : le statut de l'émargement : 0 ==> Initial 1 ==> Envoyé 2 ==> Validé """ def CreateTableauEmargement_From_Sequence(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 = ['class_internal_url', 'session_id', 'token'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée") return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['class_internal_url', 'session_id', 'token'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies" mydata = {} my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner partner_recid = my_partner['recid'] class_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: class_internal_url = diction['class_internal_url'] session_id = "" if ("session_id" in diction.keys()): if diction['session_id']: mydata['session_id'] = diction['session_id'] session_id = diction['session_id'] """ Recuperation des info de la session de formation """ date_du = "" date_au = "" ville = "" code_postal = "" adresse = "" qry = {"_id":ObjectId(str(session_id)), "valide": "1", 'partner_owner_recid':str(partner_recid)} # Verifier que la session existe session_formation_count = MYSY_GV.dbname['session_formation'].count_documents(qry) if( session_formation_count <= 0 ): mycommon.myprint( str(inspect.stack()[0][3]) + " - Aucune session de formation valide ") return False, "Aucune session de formation valide " session_data = MYSY_GV.dbname['session_formation'].find_one({"_id":ObjectId(str(session_id)), "valide": "1"}) # Verifier qu'on a bien des sequences valides pour cette session de formation is_existe_sequence_session_valide_count = MYSY_GV.dbname['session_formation_sequence'].count_documents({'session_id':str(diction['session_id']), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(partner_recid)}) if( is_existe_sequence_session_valide_count <= 0): mycommon.myprint( str(inspect.stack()[0][3]) + " - Aucune sequence valide pour cette session de formation ") return False, "Aucune sequence valide pour cette session de formation " # Recuperation de la liste des sequences associées à la session i = 0 # compteur de securité # On vide la collection emargement pour la session = str(session_id) local_myquery = {"session_id": str(session_id), 'partner_owner_recid':str(partner_recid)} x = MYSY_GV.dbname['emargement'].delete_many(local_myquery) mycommon.myprint( str(inspect.stack()[0][ 3]) + " - INFO : "+str(x.deleted_count)+" Documents supprimés de la collection : emargement : session = "+str(session_id) ) # Verification qu'il y a bien des séquences de formation pour cette session session_sequence_count = MYSY_GV.dbname['session_formation_sequence'].count_documents({'session_id': str(session_id), 'locked': '0', 'valide': '1', 'partner_owner_recid': str( partner_recid)}) if( session_sequence_count <= 0 ): mycommon.myprint( str(inspect.stack()[0][3]) + " - Cette session de formation n'a aucune séquence valide ") return False, " Cette session de formation n'a aucune séquence valide " #print("#### Liste Date = "+str(liste_date)) coll_emargement = MYSY_GV.dbname['emargement'] #Recuperation de la liste des participants dont l'inscription est validée for tmp_val in MYSY_GV.dbname['inscription'].find({'session_id':str(session_id), 'class_internal_url':str(class_internal_url), 'status':'1', 'partner_owner_recid':str(partner_recid)} ): my_local_data = {} my_local_data['email'] = tmp_val['email'] my_local_data['nom'] = tmp_val['nom'] my_local_data['prenom'] = tmp_val['prenom'] my_local_data['session_id'] = tmp_val['session_id'] my_local_data['class_internal_url'] = tmp_val['class_internal_url'] my_local_data['date_update'] = str(datetime.now()) # Recuperation de liste des sequences for session_sequence in MYSY_GV.dbname['session_formation_sequence'].find({'session_id':str(session_id), 'locked':'0', 'valide':'1', 'partner_owner_recid':str(partner_recid)} ): query_key = {} query_key['email'] = tmp_val['email'] query_key['session_id'] = tmp_val['session_id'] query_key['class_internal_url'] = tmp_val['class_internal_url'] query_key['date'] = str(session_sequence['sequence_start'])[0:10] query_key['sequence_start'] = str(session_sequence['sequence_start']) query_key['sequence_end'] = str(session_sequence['sequence_end']) new_my_local_data = {} new_my_local_data['inscription_id'] = str(tmp_val['_id']) new_my_local_data['email'] = tmp_val['email'] new_my_local_data['nom'] = tmp_val['nom'] new_my_local_data['prenom'] = tmp_val['prenom'] new_my_local_data['session_id'] = tmp_val['session_id'] new_my_local_data['class_internal_url'] = tmp_val['class_internal_url'] new_my_local_data['date_update'] = str(datetime.now()) query_key['date'] = str(session_sequence['sequence_start'])[0:10] new_my_local_data['is_present'] = False new_my_local_data['matin'] = False new_my_local_data['apresmidi'] = False new_my_local_data['statut'] = "0" new_my_local_data['date_envoi'] = "" new_my_local_data['date_update'] = str(datetime.now()) new_my_local_data['update_by'] = str(my_partner['recid']) new_my_local_data['valide'] = "1" new_my_local_data['locked'] = "0" new_my_local_data['partner_owner_recid'] = str(str(partner_recid)) #print("### emargement insertion de : "+str(new_my_local_data)) local_ret_val = coll_emargement.find_one_and_update( query_key, {"$set": new_my_local_data}, upsert=True, return_document=ReturnDocument.AFTER ) return True, " Le tableau d'émargement a été correctement initialisé " 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 initialiser de créer le tableau d'émargement " """ Cette fonction récupérer la liste d'emargement """ def GetTableauEmargement(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 = ['class_internal_url', 'session_id', 'token'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str(inspect.stack()[0][ 3]) + " - Le champ '" + val + "' n'est pas autorisé") return False, "Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['class_internal_url', 'session_id', 'token'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, "Toutes les informations necessaires n'ont pas été fournies" mydata = {} my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] # Verifier la validité du token retval = mycommon.check_partner_token_validity("", my_token) if retval is False: return "Err_Connexion", " La session de connexion n'est pas valide" class_internal_url = "" if ("class_internal_url" in diction.keys()): if 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 " session_id = "" if ("session_id" in diction.keys()): if diction['session_id']: mydata['session_id'] = diction['session_id'] session_id = diction['session_id'] val_tmp = 1 RetObject = [] for retval in MYSY_GV.dbname['emargement'].find({'session_id':str(session_id), 'class_internal_url':str(class_internal_url)}).sort([("date",pymongo.ASCENDING), ("sequence_start",pymongo.ASCENDING), ("sequence_end",pymongo.ASCENDING),]): user = retval if( "statut" not in retval.keys() ): user['statut'] = "" if ("date_envoi" not in retval.keys()): user['date_envoi'] = "" if ("date_emargement" not in retval.keys()): user['date_emargement'] = "" if ("mysy_manual_signature_img" not in retval.keys()): user['mysy_manual_signature_img'] = "" user['id'] = str(val_tmp) RetObject.append(mycommon.JSONEncoder().encode(user)) val_tmp = val_tmp + 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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, "Impossible de récupérer le tableau d'emargement" """ Fonction de mise à jour de l'emargement d'une personne sur une date """ def UpdateUserEmargementDate(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 = [ 'session_id', 'token', 'email', 'date', 'matin', 'apresmidi', 'class_internal_url', 'is_present', '_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é, Creation partenaire annulée") return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = [ 'session_id', 'token', 'email', 'date', '_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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner partner_recid = str(my_partner['recid']) # Verifier la validité du token retval = mycommon.check_partner_token_validity("", my_token) if retval is False: return "Err_Connexion", " La session de connexion n'est pas valide" """ Verifier que l'emargement est valide """ is_valide_emargement = MYSY_GV.dbname['emargement'].count_documents({'_id':ObjectId(str(diction['_id'])), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(partner_recid)}) if( is_valide_emargement != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'emargement est invalide ") return False, " L'identifiant de l'emargement est invalide " is_valide_emargement_data = MYSY_GV.dbname['emargement'].find_one( {'_id': ObjectId(str(diction['_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(partner_recid)}) is_present = "" if ("is_present" in diction.keys()): if diction['is_present']: is_present = diction['is_present'] mydata = {} now = str(datetime.now().strftime("%d/%m/%Y %H:%M:%S")) mydata['date_update'] = now if (is_present == "1"): mydata['is_present'] = True elif (is_present == "0"): mydata['is_present'] = False mydata['statut'] = "2" mydata['date_emargement'] = now mydata['update_by'] = str(my_partner['_id']) query_key = {} query_key['_id'] = ObjectId(str(diction['_id'])) query_key['partner_owner_recid'] = str(partner_recid) #print(" #### query_key = ", query_key) #print(" #### mydata = ", mydata) coll_emargement = MYSY_GV.dbname['emargement'] local_ret_val = coll_emargement.find_one_and_update( query_key, {"$set": mydata}, upsert=False, return_document=ReturnDocument.AFTER ) return True, "La mise à jour a été correctement faite" 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 valider l'émargement" """ 04/07/2024 - Mise à jour d'une liste d'emargement """ def Update_Given_Emargement_Data(diction): try: diction = mycommon.strip_dictionary(diction) ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = [ 'session_id', 'token', 'email', 'date', 'matin', 'apresmidi', 'class_internal_url', 'is_present', 'tab_ids', '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é, Creation partenaire annulée") return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = [ 'token', 'tab_ids'] 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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner partner_recid = str(my_partner['recid']) # Verifier la validité du token retval = mycommon.check_partner_token_validity("", my_token) if retval is False: return "Err_Connexion", " La session de connexion n'est pas valide" tab_ids = "" if ("tab_ids" in diction.keys()): if diction['tab_ids']: tab_ids = diction['tab_ids'] tab_ids_splited = str(tab_ids).split(",") tab_ids_splited_Objectid = [] for my_id in tab_ids_splited: # Verifier que les survey sont valides qry = {'_id': ObjectId(str(my_id)), 'partner_owner_recid': str(my_partner['recid']), 'valide': '1', 'locked': '0'} print(" qry2 === ", qry) tmp_count = MYSY_GV.dbname['emargement'].count_documents(qry) if (tmp_count <= 0): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'emargement " + str(my_id) + " est invalide ") return False, " L'identifiant de l'emargement " + str(my_id) + " est invalide " tab_ids_splited_Objectid.append(ObjectId(str(my_id))) is_present = "" if ("is_present" in diction.keys()): if diction['is_present']: is_present = diction['is_present'] comment = "" if ("comment" in diction.keys()): if diction['comment']: comment = str(diction['comment']).strip() mydata = {} mydata['comment'] = str(comment) mydata['date_update'] = str(datetime.now()) mydata['update_by'] = str(my_partner['_id']) query_key = {} query_key['_id'] ={'$in':tab_ids_splited_Objectid} query_key['partner_owner_recid'] = str(partner_recid) query_key['valide'] = "1" query_key['locked'] = "0" print(" #### query_key = ", query_key) print(" #### mydata = ", mydata) coll_emargement = MYSY_GV.dbname['emargement'] local_ret_val = coll_emargement.find_one_and_update( query_key, {"$set": mydata}, upsert=False, return_document=ReturnDocument.AFTER ) return True, "La mise à jour a été correctement faite" 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 l'émargement" """ Recuperer les données d'un emargemend donnée """ def Get_Given_Emargement_Data(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', '_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é, Creation partenaire annulée") return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = [ 'token', '_id' ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner partner_recid = str(my_partner['recid']) # Verifier la validité du token retval = mycommon.check_partner_token_validity("", my_token) if retval is False: return "Err_Connexion", " La session de connexion n'est pas valide" RetObject = [] val_tmp = 0 for New_retVal in MYSY_GV.dbname['survey'].find_one({'_id':ObjectId(str(diction['_id'])), 'valide':'1', 'locked':'0', 'partner_owner_recid':my_partner['recid']}): user = New_retVal 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écuperer les données de l'émargement" """ Cette fonction genere un fichier pdf pour les liste d'emargement """ def GerneratePDFEmargementList(diction): try: field_list = ['session_id', 'email', 'date', 'token', 'class_internal_url', 'courrier_template_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]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Impossible d'imprimer la liste d'emargement" """ Verification de la liste des champs obligatoires """ field_list_obligatoire = ['session_id', 'token', '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 la liste des arguments ") return False, " Impossible d'imprimer la liste d'emargement" query_get_data = {} local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner 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 d'imprimer la liste d'emargement ") return False, " Impossible d'imprimer la liste d'emargement" """ 28/03/2024 : Verifier que le modele de courrier est valide """ courrier_template_id = "" if( "courrier_template_id" in diction.keys() ): courrier_template_id = diction['courrier_template_id'] if( courrier_template_id == ""): mycommon.myprint(str(inspect.stack()[0][3]) + " Le modèle de courrier est invalide ") return False, " Le modèle de courrier est invalide " courrier_template_id_count = MYSY_GV.dbname['courrier_template'].count_documents({"_id":ObjectId(courrier_template_id), 'partner_owner_recid':my_partner['recid'], 'valide':'1', 'locked':'0'}) if( courrier_template_id_count != 1): mycommon.myprint(str(inspect.stack()[0][3]) + " Le modèle de courrier est invalide (1) ") return False, " Le modèle de courrier est invalide (1)" courrier_template_id_data = MYSY_GV.dbname['courrier_template'].find_one( {"_id": ObjectId(courrier_template_id), 'partner_owner_recid': my_partner['recid'], 'valide': '1', 'locked': '0'}) myclass_internal_url = "" if ("class_internal_url" in diction.keys()): if diction['class_internal_url']: myclass_internal_url = diction['class_internal_url'] query_get_data['class_internal_url'] = diction['class_internal_url'] mysession_id = "" if ("session_id" in diction.keys()): if diction['session_id']: mysession_id = diction['session_id'] query_get_data['session_id'] = diction['session_id'] myemail = "" if ("email" in diction.keys()): if diction['email']: myemail = diction['email'] query_get_data['email'] = diction['email'] mydate = "" if ("date" in diction.keys()): if diction['date']: mydate = diction['date'] query_get_data['date'] = diction['date'] ## Recuperation des données coll_emargement = MYSY_GV.dbname['emargement'] print("### query_get_data = " + str(query_get_data)) tab_users = [] for val_tmp in coll_emargement.find(query_get_data).sort([("date", pymongo.ASCENDING), ("sequence_start", pymongo.ASCENDING), ("sequence_end", pymongo.ASCENDING)]): local_tmp = {} local_tmp['email'] = val_tmp['email'] local_tmp['nom'] = val_tmp['nom'] local_tmp['prenom'] = val_tmp['prenom'] local_tmp['date'] = val_tmp['date'] local_tmp['sequence_start'] = val_tmp['sequence_start'] local_tmp['sequence_end'] = val_tmp['sequence_end'] local_tmp['is_present'] = val_tmp['is_present'] mysy_manual_signature_img = "" if( "mysy_manual_signature_img" in val_tmp.keys() ): mysy_manual_signature_img = val_tmp['mysy_manual_signature_img'] local_tmp['mysy_manual_signature_img'] = mysy_manual_signature_img tab_users.append(local_tmp) #print(' ### tab_users = ' + str(tab_users)) date_for_tmplate = {} # Recuperation des données du partenaire date_for_tmplate['partner_name'] = my_partner['nom'] date_for_tmplate['partner_adresse'] = "" if( "partner_adresse" in my_partner.keys () ): date_for_tmplate['partner_adresse'] = my_partner['adr_street'] date_for_tmplate['partner_ville'] = "" if ("adr_city" in my_partner.keys()): date_for_tmplate['partner_ville'] = my_partner['adr_city'] date_for_tmplate['partner_code_postal'] = "" if ("invoice_adr_zip" in my_partner.keys()): date_for_tmplate['partner_code_postal'] = my_partner['invoice_adr_zip'] date_for_tmplate['partner_pays'] = "" if ("adr_country" in my_partner.keys()): date_for_tmplate['partner_pays'] = my_partner['adr_country'] date_for_tmplate['partner_phone'] = "" if ("telephone" in my_partner.keys()): date_for_tmplate['partner_phone'] = my_partner['telephone'] date_for_tmplate['partner_mail'] = "" if ("email" in my_partner.keys()): date_for_tmplate['partner_mail'] = my_partner['email'] if( "website" in my_partner.keys() ): date_for_tmplate['partner_website'] = my_partner['website'] else: date_for_tmplate['partner_website'] = "" mytoday = str(datetime.today().strftime("%d/%m/%Y")) date_for_tmplate['today_date'] = str(mytoday) ## Recuperation des info de la formation class_title = "" for val_tmp in MYSY_GV.dbname['myclass'].find({'internal_url': str(myclass_internal_url)}): class_title = "" if("title" in val_tmp): class_title = val_tmp['title'] date_for_tmplate['title'] = class_title ## Recuperation des info de la session code_session = "" for val_tmp_2 in MYSY_GV.dbname['session_formation'].find({'_id': ObjectId(str(mysession_id))}): code_session = "" if ("code_session" in val_tmp_2): code_session = val_tmp_2['code_session'] date_for_tmplate['code_session'] = code_session session_date_debut = "" if ("date_debut" in val_tmp_2): session_date_debut = val_tmp_2['date_debut'] date_for_tmplate['session_date_debut'] = session_date_debut session_date_fin = "" if ("date_fin" in val_tmp_2): session_date_fin = val_tmp_2['date_fin'] date_for_tmplate['session_date_fin'] = session_date_fin session_titre = "" if ("titre" in val_tmp_2): session_titre = val_tmp_2['titre'] date_for_tmplate['session_titre'] = session_titre session_distantiel = "" if ("distantiel" in val_tmp_2): session_distantiel = val_tmp_2['distantiel'] date_for_tmplate['session_distantiel'] = session_distantiel session_presentiel = "" if ("presentiel" in val_tmp_2): session_presentiel = val_tmp_2['presentiel'] date_for_tmplate['session_presentiel'] = session_presentiel #print(json.dumps(date_for_tmplate, indent=1)) # Recuperation des données du partenaire associé à l'utilisateur connecté local_status, local_retval = mycommon.Get_Connected_User_Partner_Data_From_RecID(my_partner['recid']) if (local_status is False): return local_status, local_retval local_company_data = local_retval # Recuperation des données de la société company_data = {} company_liste_champ = ['nom', 'email', 'telephone', 'num_nda', 'website', 'adr_street', 'adr_zip', 'adr_city', 'adr_country'] for champ in company_liste_champ: if (champ in local_company_data): new_champ_name = "societe_" + str(champ) new_field = {new_champ_name: str(local_company_data[champ])} company_data.update(new_field) else: new_champ_name = "societe_" + str(champ) new_field = {new_champ_name: ""} company_data.update(new_field) # Recuperation du logo et du cachet de la société si il y en a local_part_status, local_part_imgs = partners.getRecodedParnterImage_from_front( {'token': str(diction['token'])}) if (local_part_status is False): mycommon.myprint( str(inspect.stack()[0][3]) + " WARNING : Impossible de récuperer le logo et le cachet du partenaire ") new_field_logo = {'societe_logo': ''} new_field_cachet = {'societe_cachet': ''} company_data.update(new_field_logo) company_data.update(new_field_cachet) else: local_JSON = ast.literal_eval(local_part_imgs[0]) new_field_logo = {'societe_logo': "data:image/png;base64," + local_JSON['logo_img']} new_field_cachet = {'societe_cachet': "data:image/png;base64," + local_JSON['cachet_img']} company_data.update(new_field_logo) company_data.update(new_field_cachet) #print(" -------- ") #print(json.dumps(tab_users, indent=1)) # This data can come from database query body = { "data": date_for_tmplate, "user": tab_users, "company_data" :company_data } ### Recuperation du modèle de courrier pdf pour l'emargement """ data_doc = {} data_doc['document_ref_intere'] = "EMARGEMENT" data_doc['document_type'] = "pdf" data_doc['partner_owner_recid'] = str(partner_recid) local_retval_status, local_retval_message = mycommon.Get_Personnalized_Document_From_courrier_template(data_doc) if( local_retval_status is False): return local_retval_status, local_retval_message """ partner_document_data = courrier_template_id_data contenu_doc_Template = jinja2.Template(str(partner_document_data['contenu_doc'])) sourceHtml = contenu_doc_Template.render(json_data=body["data"], users=body["user"], company_data=body["company_data"]) orig_file_name = "Emargement.pdf" outputFilename = str(MYSY_GV.EMARGEMENT_DIRECTORY) + str(orig_file_name) # open output file for writing (truncated binary) resultFile = open(outputFilename, "w+b") # convert HTML to PDF pisaStatus = pisa.CreatePDF( src=sourceHtml, # the HTML to convert dest=resultFile) # file handle to receive result # close output file resultFile.close() #print(" ### outputFilename = "+str(outputFilename)) if os.path.exists(outputFilename): #print(" ### ok os.path.exists(outputFilename) "+str(outputFilename)) return True, send_file(outputFilename, as_attachment=True) # myprint(str(inspect.stack()[0][3]) +" debut envoie de la factureeee "+diction['invoice_id']) # email.SendInvoiceEmail(str(diction['invoice_email']), diction) # On deplace la facture vers le serveur ftp '''myprint( str(inspect.stack()[0][3]) + " deplacement de la facture vers " + str( outputFilename)) ''' """ cnopts = pysftp.CnOpts() cnopts.hostkeys = None with pysftp.Connection(host=MYSY_GV.MYSY_FTP_HOST, username=MYSY_GV.MYSY_FTP_LOGIN, password=MYSY_GV.MYSY_FTP_PWD, cnopts=cnopts) as session: print("Connection successfully established ... ") localFilePath = outputFilename remoteFilePath = str(MYSY_GV.INVOICE_FTP_LOCAL_STORAGE_DIRECTORY) + str(orig_file_name) # print(" DEPLACEMENT DE " + str(localFilePath) + " VERS " + str(remoteFilePath) + " AVANTTT TRAITEMENT") # Use put method to upload a file session.put(localFilePath, remoteFilePath) # Switch to a remote directory myprint( str(inspect.stack()[0][3]) + " DEPLACEMENT DE " + str(localFilePath) + " VERS " + str( remoteFilePath) + " EST OKKKK") """ # return True on success and False on errors print(pisaStatus.err, type(pisaStatus.err)) return True, " le fichier generé " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, False