""" Ce fichier contient les fonctions utilisées dans les Crons Le principe : Les crons automatisent des fonctions qui existent dans le système avec certaines particularités. Pour eviter de surcharger ces fonctions "classique", elles seront reécrites dans ce fichier """ import ast from calendar import monthrange from zipfile import ZipFile import bson import pymongo import xlsxwriter from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime, date, timedelta import Session_Formation import module_editique import partner_base_setup import partner_client import partner_order 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 import survey_mgt as survey_mgt import attached_file_mgt import Inscription_mgt as Inscription_mgt """ Cette fonction prepare et envoi les conVocation a chaque participant à la session de formation. /!\ : 27/02/2025 : Dans le cadre du traitement automatique, il ne faut pas envoyé de convocation au personnes de la meme session qui ont déjà recues leur convocation. Comment faire : Aller verifier dans la collection : 'courrier_template_tracking_history' si on a un rengistrement avec les infos suivantes : - target_collection_recid = '_id' de l'inscrit - courrier_template_tracking_id = '_id' du type de document (qui se trouve ici : courrier_template_tracking). (dans la collection 'courrier_template_tracking', on enregistre les type de modèle de documents comme : CONVOCATION_STAGIAIRE, EMARGEMENT_FORMATION, CONVENTION_STAGIAIRE_ENTREPRISE, etc) """ def Prepare_and_Send_Convocation_From_Session_By_Email(tab_files, Folder, diction): try: field_list_obligatoire = ['token', 'session_id', 'courrier_template_id', 'email_test', 'email_production'] 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, " La valeur '" + val + "' n'est pas presente dans liste" 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 # Verifier que la session est valide is_session_valide = MYSY_GV.dbname['session_formation'].count_documents( {'_id': ObjectId(str(diction['session_id'])), 'valide': '1', 'partner_owner_recid': str(my_partner['recid'])}) if (is_session_valide != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de la session est invalide ") return False, " L'identifiant de la session est invalide " # Verifier qu'il ya bien des inscriptions valides dans la session pour le client liste_participants = MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']), 'status': '1', 'partner_owner_recid': str(my_partner['recid']), } ) #print(" ### la liste des liste_participants = ", liste_participants) # 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) # Recupération des données du modèle de document is_convention_by_client = "0" courrier_template_data = MYSY_GV.dbname['courrier_template'].find_one( {'_id': ObjectId(str(diction['courrier_template_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])} ) if( courrier_template_data and "edit_by_client" in courrier_template_data.keys() and courrier_template_data['edit_by_client'] == "1"): is_convention_by_client = "1" """ 27/02/2025 : Recuperer les données du type de modèle de courrier : 'courrier_template_type_document' """ courrier_template_tracking_data = MYSY_GV.dbname['courrier_template_tracking'].find_one( {'courrier_template_type_document_ref_interne': str( courrier_template_data['courrier_template_type_document_ref_interne']), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}) print(" EMAIL courrier_template_tracking_data = ", courrier_template_tracking_data) if (not courrier_template_tracking_data or courrier_template_tracking_data is None): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant du courrier_template_type_document est invalide ") return False, " L'identifiant du courrier_template_type_document est invalide " if (str(is_convention_by_client) == "0"): liste_inscription = MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']), 'status': '1', 'partner_owner_recid': str(my_partner['recid']), } ) #print(" ### LIST liste_inscription: ", liste_inscription) for single_inscrit in liste_inscription : #print(" ### convocation traitement de l'inscrit : ",single_inscrit ) #field_list_obligatoire = [ 'token', 'inscription_id', 'courrier_template_id', 'email_test', 'email_production' ] new_diction_no_client = {} new_diction_no_client['token'] = str(diction['token']) new_diction_no_client['inscription_id'] = str(single_inscrit['_id']) new_diction_no_client['courrier_template_id'] = diction['courrier_template_id'] new_diction_no_client['session_id'] = diction['session_id'] new_diction_no_client['email_test'] = diction['email_test'] new_diction_no_client['email_production'] = diction['email_production'] """ 27/02/2025 : Dans le cadre du traitement automatique, il ne faut pas envoyé de convocation au personnes de la meme session qui ont déjà recues leur convocation. Comment faire : Aller verifier dans la collection : 'courrier_template_tracking_history' si on a un rengistrement avec les infos suivantes : - target_collection_recid = '_id' de l'inscrit - courrier_template_tracking_id = '_id' du type de document (qui se trouve ici : courrier_template_tracking). (dans la collection 'courrier_template_tracking', on enregistre les type de modèle de documents comme : CONVOCATION_STAGIAIRE, EMARGEMENT_FORMATION, CONVENTION_STAGIAIRE_ENTREPRISE, etc) """ qry_courrier_template_tracking_history = {'partner_owner_recid': str(my_partner['recid']), 'courrier_template_tracking_id': str(courrier_template_tracking_data['_id']), 'target_collection_recid': str(single_inscrit['_id']), 'courrier_template_id':diction['courrier_template_id']} print(" ### email : qry_courrier_template_tracking_history = ", qry_courrier_template_tracking_history) is_convocation_already_sent_count = MYSY_GV.dbname[ 'courrier_template_tracking_history'].count_documents( {'partner_owner_recid': str(my_partner['recid']), 'courrier_template_tracking_id': str(courrier_template_tracking_data['_id']), 'target_collection_recid': str(single_inscrit['_id']), 'courrier_template_id':diction['courrier_template_id']} ) if (is_convocation_already_sent_count == 0): #print(" ##### new_diction_no_client = ", new_diction_no_client) local_status, local_retval = Sent_Convocation_Stagiaire_By_Email(tab_saved_file_full_path, Folder, new_diction_no_client) if (local_status is False): mycommon.myprint(" WARNING impossible d'envoyer la convocation a l'apprenant : " + str(single_inscrit['_id']) ) # Traitement de l'eventuel fichier joint tab_files_to_attache_to_mail = [] """ 25/01/2024 : pour loger une action dans la collection ==> courrier_template_tracking_history """ """local_status, local_retval = module_editique.Editic_Log_History_Action(my_partner, courrier_template_data, str(diction['session_id']), "inscription", str(single_inscrit['_id']))""" local_status, local_retval = module_editique.Editic_Log_History_Action_From_courrier_template_type_document_ref_interne( my_partner, "CONVOCATION_STAGIAIRE", str(diction['session_id']), 'inscription', str(single_inscrit['_id']), str(str(courrier_template_data['_id']))) else: print(" EMAIL DE CONVOCATION DEJA ENVOYEEEE ") return True, " Les convocations ont été correctement envoyées par emails" 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 d'envoyer les conventions par email " """ Cette fonction permet de telecharger PDF toutes les conVOCAtion d'une session. Depuis la session, l'utilisateur choisi le modele PDF et telecharge les conventions d'entreprise et les conventions individuelles. On créé un fichier zip qui sera téléchargé avec toutes les pièces jointes. /!\ : 27/02/2025 : Dans le cadre du traitement automatique, il ne faut pas envoyé de convocation au personnes de la meme session qui ont déjà recues leur convocation. Comment faire : Aller verifier dans la collection : 'courrier_template_tracking_history' si on a un rengistrement avec les infos suivantes : - target_collection_recid = '_id' de l'inscrit - courrier_template_tracking_id = '_id' du type de document (qui se trouve ici : courrier_template_tracking). (dans la collection 'courrier_template_tracking', on enregistre les type de modèle de documents comme : CONVOCATION_STAGIAIRE, EMARGEMENT_FORMATION, CONVENTION_STAGIAIRE_ENTREPRISE, etc) """ def Prepare_and_Send_Convocation_From_Session_By_PDF(diction): try: field_list_obligatoire = ['token', 'session_id', 'courrier_template_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, " La valeur '" + val + "' n'est pas presente dans liste" 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 # Verifier que la session est valide is_session_valide = MYSY_GV.dbname['session_formation'].count_documents( {'_id': ObjectId(str(diction['session_id'])), 'valide': '1', 'partner_owner_recid': str(my_partner['recid'])}) if (is_session_valide != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de la session est invalide ") return False, " L'identifiant de la session est invalide " # Stokage des nom de fichier à zipper list_file_name_to_zip = [] liste_inscription = MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']), 'status': '1', 'partner_owner_recid': str(my_partner['recid']), } ) """ for val in liste_inscription_no_client: print(" ### la liste des liste_inscription_no_client = ", str(val)) """ # Recupération des données du modèle de document is_convention_by_client = "0" courrier_template_data = MYSY_GV.dbname['courrier_template'].find_one( {'_id': ObjectId(str(diction['courrier_template_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])} ) if (courrier_template_data and "edit_by_client" in courrier_template_data.keys() and courrier_template_data[ 'edit_by_client'] == "1"): is_convention_by_client = "1" """ 27/02/2025 : Recuperer les données du type de modèle de courrier : 'courrier_template_type_document' """ courrier_template_tracking_data = MYSY_GV.dbname['courrier_template_tracking'].find_one( {'courrier_template_type_document_ref_interne':str(courrier_template_data['courrier_template_type_document_ref_interne']), 'valide':'1', 'locked':'0', 'partner_owner_recid':str(my_partner['recid'])}) if( not courrier_template_tracking_data or courrier_template_tracking_data is None): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant du courrier_template_type_document est invalide ") return False, " L'identifiant du courrier_template_type_document est invalide " if (str(is_convention_by_client) == "0"): liste_participants = MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']), 'status': '1', 'partner_owner_recid': str(my_partner['recid']), } ) for val in liste_participants: local_diction = {} # field_list_obligatoire = ['token', 'session_id', 'courrier_template_id', 'client_id'] local_diction['token'] = diction['token'] local_diction['session_id'] = diction['session_id'] local_diction['courrier_template_id'] = diction['courrier_template_id'] local_diction['inscription_id'] = str(val['_id']) """ 27/02/2025 : Dans le cadre du traitement automatique, il ne faut pas envoyé de convocation au personnes de la meme session qui ont déjà recues leur convocation. Comment faire : Aller verifier dans la collection : 'courrier_template_tracking_history' si on a un rengistrement avec les infos suivantes : - target_collection_recid = '_id' de l'inscrit - courrier_template_tracking_id = '_id' du type de document (qui se trouve ici : courrier_template_tracking). (dans la collection 'courrier_template_tracking', on enregistre les type de modèle de documents comme : CONVOCATION_STAGIAIRE, EMARGEMENT_FORMATION, CONVENTION_STAGIAIRE_ENTREPRISE, etc) """ qry_courrier_template_tracking_history = {'partner_owner_recid':str(my_partner['recid']), 'courrier_template_tracking_id':str(courrier_template_tracking_data['_id']), 'target_collection_recid':str(val['_id']), 'courrier_template_id':diction['courrier_template_id']} print(" ### qry_courrier_template_tracking_history = ", qry_courrier_template_tracking_history) is_convocation_already_sent_count = MYSY_GV.dbname['courrier_template_tracking_history'].count_documents( {'partner_owner_recid':str(my_partner['recid']), 'courrier_template_tracking_id':str(courrier_template_tracking_data['_id']), 'target_collection_recid':str(val['_id']), 'courrier_template_id':diction['courrier_template_id']} ) if( is_convocation_already_sent_count == 0 ): #print(" #### Traitemnt de local_diction bb = ", local_diction) local_status, local_full_file_name = Create_Convocation_By_Stagiaire_PDF(local_diction) if( local_status is False): return local_status, local_full_file_name else: list_file_name_to_zip.append(str(local_full_file_name)) else: print(" PDF DE CONVOCATION DEJA ENVOYEEEE ") # Create a ZipFile Object todays_date = str(date.today().strftime("%d/%m/%Y")) ts = datetime.now().timestamp() ts = str(ts).replace(".", "").replace(",", "")[-3:] zip_file_name = str(MYSY_GV.TEMPORARY_DIRECTORY_V2)+"List_Convocation_session_"+str(diction['session_id'])+"_"+str(ts)+".zip" with ZipFile(zip_file_name, 'w') as zip_object: for pdf_files in list_file_name_to_zip : #print(" ### fichier a zipper = ", pdf_files) zip_object.write(str(pdf_files)) if os.path.exists(zip_file_name): # print(" ### ok os.path.exists(outputFilename) "+str(outputFilename)) return True, send_file(zip_file_name, as_attachment=True) return False, " Impossible de générer les conventions par pdf PDF (1) " 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 générer les conventions par pdf PDF " """ Envoi d'une convocation pour un participant donné par email Si le participants est rattaché à un client , alors on va mettre en copie de l'email les contacts de communication du client de rattachement """ def Sent_Convocation_Stagiaire_By_Email(tab_files, Folder, diction): try: field_list_obligatoire = ['token', 'inscription_id', 'courrier_template_id', 'email_test', 'email_production'] 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, " La valeur '" + val + "' n'est pas presente dans liste" 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 # Verifier que le stagiaire est bien inscrit. Le statut de l'inscription doit etre "1" is_inscription_valide = MYSY_GV.dbname['inscription'].count_documents( {'_id': ObjectId(str(diction['inscription_id'])), 'status': '1', 'partner_owner_recid': str(my_partner['recid'])}) if (is_inscription_valide != 1): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de l'inscription est invalide ") return False, " L'identifiant de l'inscription est invalide " # Traitement de l'eventuel fichier joint tab_files_to_attache_to_mail = [] for saved_file in tab_files: """ 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(saved_file, "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(saved_file))) new_node = {"attached_file": file_to_attache_to_mail} tab_files_to_attache_to_mail.append(new_node) # Verification de la validité des adresses email_recu """ /!\ : Si l'email de test est repli, alors on considere que c'est un test, on ne prend pas en compte l'email de email_production. Ceci pour forcer les utilisateur à ne remplir que l'email de prod s'il veulent l'envoyer en prod. Si l'adresse email_prodution = "defaul", cela veut dire qu'on envoie la convention à : - l'adresse email du stagiaire et ses tuteurs (si les tuteurs on cochés la case 'inclu com' """ send_in_production = 0 tab_emails_destinataire = [] if ("email_test" in diction.keys() and diction['email_test']): send_in_production = 0 tab_email_test = str(diction['email_test']).replace(";", ",").split(",") for email in tab_email_test: email = email.strip() if (mycommon.isEmailValide(email) is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'adresse email " + str(email) + " est invalide ") return False, " L'adresse email " + str(email) + " est invalide " tab_emails_destinataire = tab_email_test elif ("email_production" in diction.keys() and diction['email_production']): send_in_production = 1 if (str(diction['email_production']) != "default"): tab_email_prod = str(diction['email_production']).replace(";", ",").split(",") for email in tab_email_prod: email = email.strip() if (mycommon.isEmailValide(str(email)) is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'adresse email '" + str(email) + "' est invalide ") return False, " L'adresse email " + str(email) + " est invalide " tab_emails_destinataire = tab_email_prod else: send_in_production = 1 # On va chercher les adresse email de communication du local_dict = {'token': str(diction['token']), '_id': str(diction['inscription_id'])} local_status, tab_apprenant_contact = Inscription_mgt.Get_Statgiaire_Communication_Contact(local_dict) if (local_status is False): return local_status, tab_apprenant_contact tmp_tab = [] # print(" ### tab_apprenant_contact = ", tab_apprenant_contact) for tmp in tab_apprenant_contact: if ("email" in tmp.keys()): tab_emails_destinataire.append((tmp['email'])) if (len(tab_emails_destinataire) <= 0): mycommon.myprint( str(inspect.stack()[0][ 3]) + " Aucune adresse email n'a été fourni. ") return False, " Aucune adresse email n'a été fourni. " # print(" ### tab_emails_destinataire = ", tab_emails_destinataire) # Verifier que le 'courrier_template_id' est valide # Ici le template doit etre un email is_courrier_template_id_valide = MYSY_GV.dbname['courrier_template'].count_documents( {'_id': ObjectId(str(diction['courrier_template_id'])), 'valide': '1', 'type_doc': 'email', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])} ) if (is_courrier_template_id_valide != 1): mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant du modèle de courrier est invalide ") return False, " L'identifiant du modèle de courrier est invalide " # Recupération des données du modèle de document courrier_template_data = MYSY_GV.dbname['courrier_template'].find_one( {'_id': ObjectId(str(diction['courrier_template_id'])), 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])} ) # Recuperation des eventuelles pièces jointes du modèle que courrier local_dic = {} local_dic['token'] = str(diction['token']) local_dic['object_owner_collection'] = "courrier_template" local_dic['object_owner_id'] = str(courrier_template_data['_id']) local_status, local_retval = attached_file_mgt.Get_List_object_owner_collection_Stored_Files(local_dic) if (local_status is False): return local_status, local_retval # print(" ### file stocké = ", local_retval) # Recuperation des fichiers attachés au modele de courrier, s'il y en a for file in local_retval: local_JSON = ast.literal_eval(file) saved_file = local_JSON['full_path'] file_to_attache_to_mail = MIMEBase('application', "octet-stream") file_to_attache_to_mail.set_payload(open(saved_file, "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(saved_file))) new_node = {"attached_file": file_to_attache_to_mail} tab_files_to_attache_to_mail.append(new_node) # Recuperation des données du stagaire inscription_data = MYSY_GV.dbname['inscription'].find_one({'_id': ObjectId(str(diction['inscription_id'])), 'status': '1', 'partner_owner_recid': str(my_partner['recid'])}) tab_apprenant_client_rattachement_contact = "" # Verifier si le modele de courrier n'est par 'edit_by_client', au quel cas on verifie que l'appressant est bien lié à un client if ("edit_by_client" in courrier_template_data.keys() and str(courrier_template_data['edit_by_client']) == "0"): stagiaire_client_id = "" if ("client_rattachement_id" in inscription_data.keys() and inscription_data['client_rattachement_id']): stagiaire_client_id = str(inscription_data['client_rattachement_id']) local_diction = {"token":str(diction['token']), "_id":stagiaire_client_id } #print(" ##### local_diction pr Get_Partner_Client_Communication_Contact= ", local_diction) local_status, partner_client_contact_communication = partner_client.Get_Partner_Client_Communication_Contact(local_diction) if (local_status is True): #print(" ### partner_client_contact_communication = ", partner_client_contact_communication) tab_local_email_production = [] for tmp in partner_client_contact_communication: tmp_JSON = ast.literal_eval(tmp) if ("email" in tmp_JSON.keys()): tab_local_email_production.append(str(tmp_JSON["email"])) tab_apprenant_client_rattachement_contact = ",".join(tab_local_email_production) tab_participant = [] tab_participant.append(inscription_data['_id']) """ update du 30/01/2024 si le participant à un apprenant_id, alors on recupere la valeur de l'apprenant id. On doit travailler sur cette dernier. En general, avant d'envoyer une convocation ou convention, l'inscription est validée et donc le dossier apprenant existe. Du coup si tout se passe bien, dans cette fonction, on travaillera tjrs avec l'apprenant_id """ tab_apprenant = [] if( "apprenant_id" in inscription_data.keys() and inscription_data['apprenant_id']) : tab_apprenant.append(ObjectId(str(inscription_data['apprenant_id']))) # Recuperations des info de la session de formation session_data = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(str(inscription_data['session_id'])), 'valide': '1', 'partner_owner_recid': str(my_partner['recid'])}) tab_session = [] tab_session.append(session_data['_id']) # Recuperation du titre de la formation class_data = MYSY_GV.dbname['myclass'].find_one( {'internal_url': str(session_data['class_internal_url']), 'valide': '1', 'partner_owner_recid': str(my_partner['recid']), 'locked': '0'}) tab_class = [] tab_class.append(class_data['_id']) # Creation du dictionnaire d'information à utiliser pour la creation du doc convention_dictionnary_data = {} new_diction = {} new_diction['token'] = diction['token'] new_diction['list_stagiaire_id'] = tab_participant new_diction['list_session_id'] = tab_session new_diction['list_class_id'] = tab_class new_diction['list_client_id'] = [] new_diction['list_apprenant_id'] = tab_apprenant local_status, local_retval = mycommon.Get_Dictionnary_data_For_Template(new_diction) if (local_status is False): return local_status, local_retval convention_dictionnary_data = local_retval body = { "params": convention_dictionnary_data, } orig_file_name = None outputFilename = None # Verifier s'il s'agit d'un document à envoyer avec une version de pièce jointe. if ("joint_pdf" in courrier_template_data.keys() and str(courrier_template_data['joint_pdf']) == "1"): # Il s'agit bien d'un envoie avec 'contenu_doc' en pièce jointe PDF """ 1 - Creation du PDF """ contenu_doc_Template = jinja2.Template(str(courrier_template_data['contenu_doc'])) sourceHtml = contenu_doc_Template.render(params=body["params"]) todays_date = str(date.today().strftime("%d/%m/%Y")) ts = datetime.now().timestamp() ts = str(ts).replace(".", "").replace(",", "")[-5:] orig_file_name = "Convocation_" + str(my_partner['recid'])[0:5] + "_" + str(ts) + ".pdf" outputFilename = str(MYSY_GV.TEMPORARY_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() # Attachement du fichier joint file_to_attache_to_mail = MIMEBase('application', "octet-stream") file_to_attache_to_mail.set_payload(open(outputFilename, "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(outputFilename))) new_node = {"attached_file": file_to_attache_to_mail} tab_files_to_attache_to_mail.append(new_node) ## Creation du mail au format email corps_mail_Template = jinja2.Template(str(courrier_template_data['corps_mail'])) sourceHtml = corps_mail_Template.render(params=body["params"]) html_mime = MIMEText(sourceHtml, 'html') # Creation de l'email à enoyer msg = MIMEMultipart("alternative") else: # Il s'agit d'une simple email ## Creation du mail au format email contenu_doc_Template = jinja2.Template(str(courrier_template_data['contenu_doc'])) sourceHtml = contenu_doc_Template.render(params=body["params"]) html_mime = MIMEText(sourceHtml, 'html') # Creation de l'email à enoyer msg = MIMEMultipart("alternative") """ Recuperation des parametre SMTP du partner si le client a decidé d'utiliser son propre smpt """ partner_own_smtp_value = "0" partner_own_smtp = MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'partner_smtp', 'valide': '1', 'locked': '0'}) if (partner_own_smtp and "config_value" in partner_own_smtp.keys()): partner_own_smtp_value = partner_own_smtp['config_value'] if (str(partner_own_smtp_value) == "1"): partner_SMTP_COUNT_password = str(MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'smtp_user_pwd', 'valide': '1', 'locked': '0'}, {'config_value': 1})['config_value']) partner_SMTP_COUNT_smtpsrv = str(MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'smtp_server', 'valide': '1', 'locked': '0'}, {'config_value': 1})['config_value']) partner_SMTP_COUNT_user = str(MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'smtp_user', 'valide': '1', 'locked': '0'}, {'config_value': 1})['config_value']) partner_SMTP_COUNT_From_User = str(MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'smtp_count_from_name', 'valide': '1', 'locked': '0'}, {'config_value': 1})['config_value']) partner_SMTP_COUNT_port = str(MYSY_GV.dbname['base_partner_setup'].find_one( {'partner_owner_recid': str(my_partner['recid']), 'config_name': 'smtp_count_port', 'valide': '1', 'locked': '0'}, {'config_value': 1})['config_value']) if (str(partner_own_smtp_value) == "1"): smtpserver = smtplib.SMTP(partner_SMTP_COUNT_smtpsrv, partner_SMTP_COUNT_port) else: smtpserver = smtplib.SMTP(MYSY_GV.O365_SMTP_COUNT_smtpsrv, MYSY_GV.O365_SMTP_COUNT_port) if (str(partner_own_smtp_value) == "1"): msg.attach(html_mime) msg['From'] = partner_SMTP_COUNT_From_User msg['Cc'] = tab_apprenant_client_rattachement_contact msg['Bcc'] = 'contact@mysy-training.com' msg['Subject'] = courrier_template_data['sujet'] # msg['to'] = "billardman01@hotmail.com" toaddrs = ",".join(tab_emails_destinataire) msg['to'] = str(toaddrs) # Attacher l'eventuelle pièces jointes for myfile in tab_files_to_attache_to_mail: msg.attach(myfile['attached_file']) smtpserver.ehlo() smtpserver.starttls() smtpserver.login(partner_SMTP_COUNT_user, partner_SMTP_COUNT_password) else: msg.attach(html_mime) msg['From'] = MYSY_GV.O365_SMTP_COUNT_From_User msg['Cc'] = tab_apprenant_client_rattachement_contact msg['Bcc'] = 'contact@mysy-training.com' msg['Subject'] = courrier_template_data['sujet'] # msg['to'] = "billardman01@hotmail.com" toaddrs = ",".join(tab_emails_destinataire) msg['to'] = str(toaddrs) for myfile in tab_files_to_attache_to_mail: msg.attach(myfile['attached_file']) smtpserver.ehlo() smtpserver.starttls() smtpserver.login(MYSY_GV.O365_SMTP_COUNT_user, MYSY_GV.O365_SMTP_COUNT_password) val = smtpserver.send_message(msg) smtpserver.close() print(" Email envoyé 66" + str(val)) """ # Ajout de l'evenement dans l'historique """ # L'action n'est loggué pour les envois reels (en prod) if (send_in_production == 1): now = str(datetime.now().strftime("%d/%m/%Y %H:%M:%S")) history_event_dict = {} history_event_dict['token'] = diction['token'] history_event_dict['related_collection'] = "inscription" history_event_dict['related_collection_recid'] = str(diction['inscription_id']) history_event_dict['action_date'] = str(now) history_event_dict['action_description'] = "Convocation envoyée par email à la liste : " + str( tab_emails_destinataire) local_status, local_retval = mycommon.Collection_Historique.Add_Historique_Event(history_event_dict) if (local_status is False): mycommon.myprint( " WARNING : Impossible de logguer l'historique pour l'évènement : " + str(history_event_dict)) """ Après avoir loggé l'envoie reelle en production, on va aller stocker le document dans la collection : download_files si et seulement si on bien un fichier PDF (voir variables : orig_file_name et outputFilename) """ if( orig_file_name and outputFilename ): new_file = {} new_file['token'] = diction['token'] type_document = "Convocation" new_file['file_business_object'] = str(orig_file_name) new_file['file_name'] = str(orig_file_name) new_file['status'] = "1" new_file['type_document'] = str(type_document) new_file['object_owner_collection'] = "partner_client" new_file['object_owner_id'] = str(inscription_data['client_rattachement_id']) new_file['file_name_to_store'] = outputFilename # print(" ### new_file new_file = ", new_file) local_status, local_retval = attached_file_mgt.Internal_Usage_Store_User_Downloaded_File( MYSY_GV.upload_folder, new_file) if (local_status is False): print(" ## WARINNGGG Impossible de stocker le fichier de Convocation ") return True, "L'email a été correctement envoyé " 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 d'envoyer la convention par email " """ Cette fonction créer une convocation PDF par participant à une session de formation Peu importe le rattachement client ou pas. """ def Create_Convocation_By_Stagiaire_PDF(diction): try: field_list_obligatoire = ['token', 'session_id', 'courrier_template_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 liste ") return False, " La valeur '" + val + "' n'est pas presente dans liste" 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 qry = {'_id':ObjectId(str(diction['courrier_template_id'])), 'valide':'1', 'locked':'0', 'ref_interne': 'CONVOCATION_STAGIAIRE', 'partner_owner_recid':str(my_partner['recid'])} #print(" ##### qry = ", qry) # 1 - Verifier que le modele de courrier est bien editable par individu template_courrier_data = MYSY_GV.dbname['courrier_template'].find_one({'_id':ObjectId(str(diction['courrier_template_id'])), 'valide':'1', 'locked':'0', 'ref_interne': 'CONVOCATION_STAGIAIRE', 'partner_owner_recid':str(my_partner['recid'])}) if( template_courrier_data is None ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant du modèle de courrier est invalide ") return False, " L'identifiant du modèle de courrier est invalide " if ("edit_by_client" in template_courrier_data.keys() and str(template_courrier_data['edit_by_client']) == "1"): mycommon.myprint( str(inspect.stack()[0][ 3]) + " Le modèle de courrier n'est pas éditable par stagiaire. ") return False, " Le modèle de courrier n'est pas éditable par stagiaire " # Verifier que le statgiaire est valide local_qry = {'_id':ObjectId(str(diction['inscription_id'])), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])} statgiaire_data = MYSY_GV.dbname['inscription'].find_one({'_id':ObjectId(str(diction['inscription_id'])), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])}) if (statgiaire_data is None ): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant du stagiaire est invalide ") return False, " L'identifiant du stagiaire est invalide " tab_stagiaire = [] tab_stagiaire.append(statgiaire_data['_id']) """ update du 30/01/2024 si le participant à un apprenant_id, alors on recupere la valeur de l'apprenant id. On doit travailler sur cette dernier. En general, avant d'envoyer une convocation ou convention, l'inscription est validée et donc le dossier apprenant existe. Du coup si tout se passe bien, dans cette fonction, on travaillera tjrs avec l'apprenant_id """ tab_apprenant = [] if ("apprenant_id" in statgiaire_data.keys() and statgiaire_data['apprenant_id']): tab_apprenant.append(ObjectId(str(statgiaire_data['apprenant_id']))) # Verifier que la session est valide session_data = MYSY_GV.dbname['session_formation'].find_one( {'_id': ObjectId(str(diction['session_id'])), 'valide': '1', 'partner_owner_recid': str(my_partner['recid'])}) if (session_data is None): mycommon.myprint( str(inspect.stack()[0][ 3]) + " L'identifiant de la session est invalide ") return False, " L'identifiant de la session est invalide " tab_session = [] tab_session.append(session_data['_id']) # Recuperation du titre de la formation class_data = MYSY_GV.dbname['myclass'].find_one( {'internal_url': str(session_data['class_internal_url']), 'valide': '1', 'partner_owner_recid': str(my_partner['recid']), 'locked': '0'}) tab_class = [] tab_class.append(class_data['_id']) # Recuperation des sequences d'une session de formation tab_sequence_session = [] for local_retval in MYSY_GV.dbname['session_formation_sequence'].find({'partner_owner_recid':str(my_partner['recid']), 'session_id':str(session_data['_id']), 'valide':'1', 'locked':'0'}): tab_sequence_session.append(local_retval['_id']) # Creation du dictionnaire d'information à utiliser pour la creation du doc convention_dictionnary_data = {} new_diction = {} new_diction['token'] = diction['token'] new_diction['list_stagiaire_id'] = tab_stagiaire new_diction['list_session_id'] = tab_session new_diction['list_class_id'] = tab_class new_diction['list_client_id'] = [] new_diction['list_apprenant_id'] = tab_apprenant new_diction['list_sequence_session_id'] = tab_sequence_session local_status, local_retval = mycommon.Get_Dictionnary_data_For_Template(new_diction) if (local_status is False): return local_status, local_retval convention_dictionnary_data = local_retval body = { "params": convention_dictionnary_data, } """ Creation du ficier PDF """ contenu_doc_Template = jinja2.Template(str(template_courrier_data['contenu_doc'])) sourceHtml = contenu_doc_Template.render(params=body["params"]) todays_date = str(date.today().strftime("%d/%m/%Y")) ts = datetime.now().timestamp() ts = str(ts).replace(".", "").replace(",", "")[-5:] orig_file_name = "Convocation_" + str(my_partner['recid'])[0:5] + "_" + str(ts) + ".pdf" outputFilename = str(MYSY_GV.TEMPORARY_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() """ 25/01/2024 : pour loger une action dans la collection ==> courrier_template_tracking_history """ #print(" str(diction['inscription_id']) == ", str(diction['inscription_id'])) """ local_status, local_retval = module_editique.Editic_Log_History_Action(my_partner, template_courrier_data, str(diction['session_id']), 'inscription', str(diction['inscription_id'])) """ local_status, local_retval = module_editique.Editic_Log_History_Action_From_courrier_template_type_document_ref_interne( my_partner, "CONVOCATION_STAGIAIRE", str(diction['session_id']), 'inscription', str(diction['inscription_id']), str(str(template_courrier_data['_id']))) return True, outputFilename 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 créer le fichier pdf de convocation par stagiaire "