Elyos_FI_Back_Office/Job_Cron_Common.py

3005 lines
140 KiB
Python

"""
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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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']}
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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 "
"""
Fonction Cron qui permet de gerer automatiquement les questionnaires de positionnement
Alog :
1 - Initialisation, si pas initialisé
2 - Envoyer si psi pas envoyé
"""
def Cron_Gestion_Questionnaire_Positionnement(diction):
try:
return True, ""
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 traiter autoamtiquement le questionnaire de positionnement (Cron_Gestion_Questionnaire_Positionnement) "
"""
Cette fonction initialise (suppression et creation ) d'une enquete pour tous
les inscrits d'un session.
algorithm :
On supprime tous enregistrements avec le clé :
- session_id,
28/06/2024: Pour permettre d'adresser une enquete à plusieurs objects,
pas que les inscrits, (le cas de enseignants), on va rajouter
- related_collection = collection concernée
- related_collection_id = l'_id concernée dans la collection
/!\ 27/02/2025 : On ne fait l'initialisation que ceux pour qui il n'y a aucune initialisation
"""
def Cron_Init_Survey_Tab_For_Inscrit_Not_Init(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['session_id', 'token', 'formulaire_id', 'survey_type', 'sending_date', 'related_collection',
'related_collection_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"
field_list_obligatoire = ['session_id', 'token', 'formulaire_id', 'survey_type', 'sending_date']
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"
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_valide_session = MYSY_GV.dbname['session_formation'].count_documents(
{'_id': ObjectId(str(diction['session_id'])),
'valide': '1',
'partner_owner_recid': str(my_partner['recid'])})
if (is_valide_session <= 0):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de la session est invalide ")
return False, " L'identifiant de la session est invalide "
tab_my_inscription_ids = []
tab_my_inscription_ids_all = []
for my_inscription_id in MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']),
'partner_owner_recid': str(
my_partner['recid']),
'valide': '1', 'status': '1'}):
tab_my_inscription_ids_all.append(str(my_inscription_id['_id']))
"""
27/02/2025 :
Recuperer la liste des inscrits pour lesquelles il n'y pas encore d'initialisation dans le survey
"""
for local_inscrit in tab_my_inscription_ids_all:
is_inscrit_exist_in_survey_count = MYSY_GV.dbname['survey'].count_documents(
{'partner_owner_recid': str(my_partner['recid']),
'session_id': str(diction['session_id']),
'survey_type': str(diction['survey_type']),
'inscription_id':str(local_inscrit)} )
if( is_inscrit_exist_in_survey_count == 0 ):
tab_my_inscription_ids.append(str(local_inscrit))
"""
Verifier la validité du formulaire
"""
is_valide_form = MYSY_GV.dbname['formulaire'].count_documents(
{'_id': ObjectId(str(diction['formulaire_id'])),
'valide': '1',
'locked': '0',
'default': '1',
'partner_owner_recid': str(my_partner['recid'])})
if (is_valide_form <= 0):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant du formulaire est invalide ")
return False, " L'identifiant du formulaire est invalide "
"""
Les inscriptions etant toutes valides, on va proceder à
1 - la creation de l'enquete
"""
for my_inscription_id in tab_my_inscription_ids:
# Recuperation des données de l'inscript
local_Insc_retval = MYSY_GV.dbname['inscription'].find_one({'session_id': str(diction['session_id']),
'_id': ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(
my_partner['recid']),
'valide': '1', 'status': '1'
})
if (local_Insc_retval is None):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
server_diction = {}
server_diction['token'] = diction['token']
server_diction['session_id'] = diction['session_id']
server_diction['inscrit_id'] = str(local_Insc_retval['_id'])
server_diction['formulaire_id'] = diction['formulaire_id']
server_diction['survey_type'] = diction['survey_type']
server_diction['sending_date'] = diction['sending_date']
if ("related_collection" in diction.keys() and diction['related_collection']
and "related_collection_id" in diction.keys() and diction['related_collection_id']):
server_diction['related_collection'] = diction['related_collection']
server_diction['related_collection_id'] = diction['related_collection_id']
else:
server_diction['related_collection'] = "inscription"
server_diction['related_collection_id'] = str(local_Insc_retval['_id'])
local_status, local_retval = Cron_Create_One_Survey_To_Inscrit(server_diction)
if (local_status is False):
return local_status, local_retval
return True, " La demande d'enquete a été créée "
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 formulaire "
"""
Cron Cette fonction envoi un formulaire pour inscrit
"""
def Cron_Create_One_Survey_To_Inscrit(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['session_id', 'token', 'inscrit_id', 'formulaire_id', 'survey_type', 'sending_date', 'related_collection',
'related_collection_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"
field_list_obligatoire = ['session_id', 'token', 'inscrit_id', 'formulaire_id', 'survey_type', 'sending_date']
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"
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
"""
Recuperer les données du formulaire
"""
my_formulaire_data = MYSY_GV.dbname['formulaire'].find_one({'_id':ObjectId(str(diction['formulaire_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(my_partner['recid'])})
if( my_formulaire_data is None):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant du formulaire est invalide ")
return False, " L'identifiant du formulaire est invalide "
"""
Recuperation des données de la session
"""
my_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 (my_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 "
"""
Recuperation des données de la formation concernée
"""
class_internal_url = ""
if( "class_internal_url" in my_session_data.keys() ):
class_internal_url = my_session_data['class_internal_url']
my_class_data = MYSY_GV.dbname['myclass'].find_one({'internal_url': str(class_internal_url),
'valide': '1',
'locked':'0',
'partner_owner_recid': str( my_partner['recid'])})
if( my_class_data is None ):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de la formation est invalide ")
return False, " L'identifiant de la formation est invalide "
"""
Recuperer les données de l'inscrit
"""
print(" qry insc = ", {'_id': ObjectId(str(diction['inscrit_id'])),
'valide': '1',
'partner_owner_recid': str(
my_partner['recid'])})
my_inscrit_data = MYSY_GV.dbname['inscription'].find_one({'_id': ObjectId(str(diction['inscrit_id'])),
'valide': '1',
'partner_owner_recid': str(
my_partner['recid'])})
if (my_inscrit_data is None):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
related_collection = ""
related_collection_id = ""
if( "related_collection" in diction.keys() and "related_collection_id" in diction.keys()):
related_collection = diction['related_collection']
related_collection_id = diction['related_collection_id']
new_survey_data = {}
new_survey_data['email'] = my_inscrit_data['email']
new_survey_data['nom'] = my_inscrit_data['nom']
new_survey_data['prenom'] = my_inscrit_data['prenom']
new_survey_data['inscription_id'] = str(my_inscrit_data['_id'])
new_survey_data['formulaire_id'] = str(diction['formulaire_id'])
new_survey_data['partner_owner_recid'] = str(my_partner['recid'])
new_survey_data['session_id'] = str(diction['session_id'])
new_survey_data['class_id'] = str(my_class_data['_id'])
new_survey_data['survey_type'] = str(diction['survey_type'])
new_survey_data['related_collection'] = str(related_collection)
new_survey_data['related_collection_id'] = str(related_collection_id)
new_survey_data['date_envoi'] = ""
new_survey_data['automatique_sending_request_date'] = str(diction['sending_date'])
new_survey_data['automatique_traitement'] = "1"
new_survey_data['automatique_traitement_done'] = "0"
new_survey_data['locked'] = "0"
new_survey_data['valide'] = "1"
new_survey_data['statut'] = "0"
new_survey_data['type'] = "1"
new_survey_data['date_update'] = str(datetime.now())
new_survey_data['update_by'] = str(my_partner['_id'])
new_survey_data['created_by'] = str(my_partner['_id'])
print(" ~~~~ A inserer = new_survey_data = ", new_survey_data)
inserted_id = MYSY_GV.dbname['survey'].insert_one(new_survey_data).inserted_id
if (not inserted_id):
mycommon.myprint(" Impossible de créer l'enquete (2) ")
return False, " Impossible de créer l'enquete (2) "
return True, " L'enquete a été créée"
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 l'enquete unitaire "
"""
Cron d'envoie du questionnaire.
Pour memo, dans le mode automatique (CRON) ne doit etre envoyé que les questionnaire non envoyés
L'utilisateur a toujours la possibilité de RENvoyer manuellement ceux non envoyés
"""
def Cron_Send_Survey_TabIds(diction):
try:
field_list_obligatoire = [ 'token', 'session_id', 'survey_type']
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"
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
"""
Recuperation des données de la session
"""
my_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 (my_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 "
my_survey_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
my_survey_ids = diction['tab_ids']
tab_my_survey_ids = []
print(" ### tab_my_survey_ids QRY = ", {'session_id':str(diction['session_id']),
'partner_owner_recid':str(my_partner['recid']),
'valide':'1',
'locked':'0',
'survey_type':str(diction['survey_type']),
'related_collection':str(diction['related_collection']),
'statut':'0'})
for survey_data in MYSY_GV.dbname['survey'].find({'session_id':str(diction['session_id']),
'partner_owner_recid':str(my_partner['recid']),
'valide':'1',
'locked':'0',
'survey_type':str(diction['survey_type']),
'related_collection':str(diction['related_collection']),
'statut':'0'}):
tab_my_survey_ids.append(str(survey_data['_id']))
"""
Les enquete etant toutes valides, on va preceder à l'envoie des email de demande
"""
"""
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'])
for my_survey_id in tab_my_survey_ids:
# Recuperation des données de l'inscript
qry = { '_id':ObjectId(str(my_survey_id)), 'partner_owner_recid': str(my_partner['recid']),'valide':'1', 'locked':'0'}
print(" ### my_survey_id Qry = ", qry)
local_survey_retval = MYSY_GV.dbname['survey'].find_one(qry)
if(local_survey_retval is None ):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
"""
13/067/2024 :
S'il s'agit de l'evaluation d'une ressource humaine, ( related_collection = 'ressource_humaine' et related_collection_id : est valide)
on va aller chercher la personne concernée dans la collection 'ressource_humaine'.
par exemple pour evaluer un enseignant, on va aller recuperer les données de l'enseignant
"""
related_rh_data = None
if ("related_collection" in local_survey_retval.keys() and local_survey_retval['related_collection'] and local_survey_retval['related_collection'] == "ressource_humaine" and
'related_collection_id' in local_survey_retval.keys() and local_survey_retval[
'related_collection_id'] ):
# Verifier la validité de la ressource humaine concernée
is_valide_related_collection_id = MYSY_GV.dbname['ressource_humaine'].count_documents(
{'_id': ObjectId(str(local_survey_retval['related_collection_id'])),
'valide': '1', 'locked': '0', 'partner_recid': str(my_partner['recid'])})
if ( is_valide_related_collection_id != 1):
str(inspect.stack()[0][3]) + " L'identifiant de la personne à évaluer est invalide "
return False, " L'identifiant de la personne à évaluer est invalide "
related_rh_data = MYSY_GV.dbname['ressource_humaine'].find_one(
{'_id': ObjectId(str(local_survey_retval['related_collection_id'])),
'valide': '1', 'locked': '0', 'partner_recid': str(my_partner['recid'])}, {'_id':1, 'nom':1, 'prenom':1, 'email':1, 'civilite':1})
print(" ### local_survey_retval = ",local_survey_retval)
courrier_ref_interne = ""
if ( "survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "pos"):
courrier_ref_interne = "QUESTION_POSITIONNEMENT"
elif ( "survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "hot_eval"):
courrier_ref_interne = "EVAL_FORMATION"
elif ( "survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "cold_eval"):
courrier_ref_interne = "EVAL_FORMATION"
elif ("survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "human_eval"):
courrier_ref_interne = "EVAL_RESSOURCE_HUMAINE"
elif ("survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "autre_eval"):
courrier_ref_interne = "EVAL_AUTRE"
else :
mycommon.myprint(str(inspect.stack()[0][3]) + " Le type d'enquete est invalide ")
return False, " Le type d'enquete est invalide "
"""
Recuperation du modele de courrier (unique) associé à l'envoie des questions de positionnement
le principe est de prendre le module du partenaire. si pas de modele du partenaire on
prend le modele par default
"""
courrier_template_model = MYSY_GV.dbname['courrier_template'].count_documents(
{'ref_interne': str(courrier_ref_interne),
'valide': '1',
'locked': '0',
'type_doc': 'email',
'partner_owner_recid': str(my_partner['recid'])}
)
if (courrier_template_model == 1):
courrier_template_data = MYSY_GV.dbname['courrier_template'].find_one(
{'ref_interne': str(courrier_ref_interne),
'valide': '1',
'locked': '0',
'type_doc': 'email',
'partner_owner_recid': str(my_partner['recid'])}
)
else:
courrier_template_model = MYSY_GV.dbname['courrier_template'].count_documents(
{'ref_interne': str(courrier_ref_interne),
'valide': '1',
'locked': '0',
'type_doc': 'email',
'partner_owner_recid': "default"}
)
if (courrier_template_model == 1):
courrier_template_data = MYSY_GV.dbname['courrier_template'].find_one(
{'ref_interne': str(courrier_ref_interne),
'valide': '1',
'locked': '0',
'type_doc': 'email',
'partner_owner_recid': "default"}
)
else:
mycommon.myprint(str(
inspect.stack()[0][
3]) + " Aucun modèle de document configuré pour envoyer les demandes pour les questionnaires de positionnement ")
return False, " Aucun modèle de document configuré pour envoyer les demandes pour les questionnaires de positionnement "
if ("contenu_doc" not in courrier_template_data.keys() or str(
courrier_template_data['contenu_doc']).strip() == ""):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Le modèle de document ne contient pas de valeur 'contenu_doc' ")
return False, " Le modèle de document ne contient pas de valeur 'contenu_doc' "
contenu_doc_Template = jinja2.Template(str(courrier_template_data['contenu_doc']))
# Creation du dictionnaire d'information à utiliser pour la creation du doc
convention_dictionnary_data = {}
new_diction = {}
new_diction['token'] = diction['token']
is_apprenant = 0
new_diction['list_stagiaire_id'] = []
if( "inscription_id" in local_survey_retval.keys() and local_survey_retval['inscription_id']):
new_diction['list_stagiaire_id'].append(ObjectId(str( local_survey_retval['inscription_id'])))
new_diction['list_session_id'] = []
if ("session_id" in local_survey_retval.keys() and local_survey_retval['session_id']):
new_diction['list_session_id'].append(ObjectId(str(local_survey_retval['session_id'])))
new_diction['list_class_id'] = []
if ("class_id" in local_survey_retval.keys() and local_survey_retval['class_id']):
new_diction['list_class_id'].append(ObjectId(str(local_survey_retval['class_id'])))
new_diction['list_client_id'] = []
new_diction['list_apprenant_id'] = []
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
servey_url = MYSY_GV.CLIENT_URL_BASE + "Survey/" + str(local_survey_retval['_id'])
convention_dictionnary_data["servey_url"] = str(servey_url)
convention_dictionnary_data["servey_email"] = str(local_survey_retval['email'])
convention_dictionnary_data["servey_nom"] = str(local_survey_retval['nom'])
convention_dictionnary_data["servey_prenom"] =str(local_survey_retval['prenom'])
if (related_rh_data and "_id" in related_rh_data.keys()):
convention_dictionnary_data["related_rh_data_is_valide"] = "1"
else:
convention_dictionnary_data["related_rh_data_is_valide"] = "0"
if (related_rh_data and "civilite" in related_rh_data.keys()):
convention_dictionnary_data["related_rh_data_civilite"] = str(related_rh_data['civilite']).capitalize()
else:
convention_dictionnary_data["related_rh_data_is_valide"] = "neutre".capitalize()
if( related_rh_data and "email" in related_rh_data.keys()):
convention_dictionnary_data["related_rh_data_email"] = str(related_rh_data['email'])
else:
convention_dictionnary_data["related_rh_data_email"] = ""
if (related_rh_data and "prenom" in related_rh_data.keys()):
convention_dictionnary_data["related_rh_data_prenom"] = str(related_rh_data['prenom'])
else:
convention_dictionnary_data["related_rh_data_prenom"] = ""
if (related_rh_data and "nom" in related_rh_data.keys()):
convention_dictionnary_data["related_rh_data_nom"] = str(related_rh_data['nom'])
else:
convention_dictionnary_data["related_rh_data_nom"] = ""
body = {
"params": convention_dictionnary_data
}
html = contenu_doc_Template.render(params=body["params"])
subject_doc_Template = jinja2.Template(str(courrier_template_data['sujet']))
subject = subject_doc_Template.render(params=body["params"])
html_mime = MIMEText(html, 'html')
# Creation de l'email à enoyer
msg = MIMEMultipart("alternative")
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['Bcc'] = 'contact@mysy-training.com'
msg['Subject'] = str(subject)
# msg['to'] = "billardman01@hotmail.com"
msg['to'] = str(local_survey_retval['email'])
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['Bcc'] = 'contact@mysy-training.com'
msg['Subject'] = str(subject)
# msg['to'] = "billardman01@hotmail.com"
msg['to'] = str(local_survey_retval['email'])
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é " + str(val))
"""
Mettre à jour l'enquete pour dire que c'est envoyé
"""
now = str(datetime.now())
qry = {'_id': ObjectId(str(my_survey_id)), 'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0'}
result = MYSY_GV.dbname['survey'].find_one_and_update(qry,
{
'$unset': {
'user_response' : "",
'date_reponse' : ""
},
'$set':{'statut':'1', 'date_update':str(now),
'update_by':str(my_partner['_id']),
'date_envoi':now}
}
)
courrier_ref_interne = ""
if ("survey_type" in result.keys() and result['survey_type'] == "pos"):
courrier_ref_interne = "QUESTION_POSITIONNEMENT"
elif ("survey_type" in result.keys() and result['survey_type'] == "hot_eval"):
courrier_ref_interne = "EVAL_FORMATION"
elif ("survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "cold_eval"):
courrier_ref_interne = "EVAL_FORMATION"
elif ("survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "human_eval"):
courrier_ref_interne = "EVAL_RESSOURCE_HUMAINE"
elif ("survey_type" in local_survey_retval.keys() and local_survey_retval['survey_type'] == "autre_eval"):
courrier_ref_interne = "EVAL_AUTRE"
else:
mycommon.myprint(str(inspect.stack()[0][3]) + " Le type d'enquete est invalide ")
return False, " Le type d'enquete est invalide "
"""
Apres que l'envoie est fait, on va loguer le traitement dans la collection : courrier_template_tracking_history
"""
if( "related_collection_id" in local_survey_retval.keys() and 'related_collection' in local_survey_retval.keys()):
# Cette demande d'enquete concerne une inscription
local_inscription_id = local_survey_retval['related_collection_id']
qry = {'_id': ObjectId(str(local_inscription_id)),
'status': '1',
'partner_owner_recid': str(my_partner['recid'])}
#print(" ### qry 010101 = ", qry)
#print(" #### str(local_survey_retval['related_collection']) = ", str(local_survey_retval['related_collection']))
partner_owner_recid = "partner_owner_recid"
if( str(local_survey_retval['related_collection']) == "ressource_humaine"):
partner_owner_recid = "partner_recid"
""" print(" #### str(local_survey_retval['related_collection']) = ", str(local_survey_retval['related_collection']))
"""
"""print(" #### qry = ",
{'_id': ObjectId(str(local_inscription_id)),
'valide': '1',
str(partner_owner_recid): str(my_partner['recid'])})"""
inscription_data = MYSY_GV.dbname[str(local_survey_retval['related_collection'])].find_one({'_id': ObjectId(str(local_inscription_id)),
'valide': '1',
str(partner_owner_recid): str(my_partner['recid'])})
#print(" inscription_data = ", inscription_data)
if (inscription_data and "session_id" in inscription_data.keys() and inscription_data['session_id']):
local_status, local_retval = module_editique.Editic_Log_History_Action_From_courrier_template_type_document_ref_interne(
my_partner, str(courrier_ref_interne), str(inscription_data['session_id']), 'inscription',
str(local_inscription_id),"")
if (local_status is False):
mycommon.myprint(
" WARNING : Impossible de logguer l'historique pour l'inscrit (2) : " + str(local_inscription_id))
return True, " Les demandes d'enquete ont été envoyées "
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 demande "
"""
CRON :
Initialisation de la liste des attestation pour les inscrits pour lesquels aucune initialisation
n'a ete faite.
"""
def Cron_Init_Attestation_Formation_With_Template(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'session_id', '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]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
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 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 valididé du model de courrier
is_courrier_valide = MYSY_GV.dbname['courrier_template'].count_documents({'_id':ObjectId(str(diction['courrier_template_id'])),
'valide':'1',
'locked':'0',
'default_version':'1',
'partner_owner_recid':str(my_partner['recid'])})
if (is_courrier_valide != 1):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Le modèle de courrier '" + str(diction['courrier_template_id']) + "' n'est pas valide ")
return False, " Le modèle de courrier '" + str(diction['courrier_template_id']) + "' n'est pas valide "
# A présent les inscriptions sont ok, le modele de courrier ok, on peut proceder à l'initialisation des attestations
warning_message = " WARNING : "
is_warning_message = 0
is_courrier_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'])})
cpt = 0
for my_inscription in MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']),
'status': '1',
'partner_owner_recid': str(my_partner['recid'])}):
"""
01/03/2025 : Regarder si cet inscrit a déjà une initialisation d'attestation faite
ou qu'une atteste existe déjà dans la collection : 'attestation_formation'
"""
is_init_attestation_exist_count = MYSY_GV.dbname['attestation_formation'].count_documents(
{'inscription_id':str(my_inscription['_id']),
'partner_owner_recid':str(my_partner['recid']),
'session_id':str(diction['session_id']),
'valide':'1',
'locked':'0'})
if( is_init_attestation_exist_count == 0 ):
# Aucune attestation (ni initialisée, ni envoyé), on peut donc initialiser l'attestation
new_attestation_data = {}
new_attestation_data['date_update'] = str(datetime.now())
new_attestation_data['update_by'] = str(my_partner['_id'])
new_attestation_data['partner_owner_recid'] = str(my_partner['recid'])
new_attestation_data['valide'] = "1"
new_attestation_data['locked'] = "0"
new_attestation_data['statut'] = "0"
new_attestation_data['inscription_id'] = str(my_inscription['_id'])
new_attestation_data['session_id'] = str(my_inscription['session_id'])
new_attestation_data['courrier_template_id'] = str(is_courrier_data['_id'])
update_key = {}
update_key['inscription_id'] = str(my_inscription['_id'])
update_key['partner_owner_recid'] = str(my_partner['recid'])
ret_val2 = MYSY_GV.dbname['attestation_formation'].find_one_and_update(
update_key,
{"$set": new_attestation_data},
return_document=ReturnDocument.AFTER,
upsert=True,
)
if (ret_val2 is None):
mycommon.myprint(str(inspect.stack()[0][3]) + " - WARNING : Impossible d'initialiser l'attestation de formation pour l'inscription : "+str(my_inscription['_id']))
warning_message = warning_message +" Impossible d'initialiser l'attestation de formation pour l'inscription : "+str(my_inscription['_id']) +" - "
is_warning_message = 1
else:
cpt = cpt +1
if( is_warning_message == 1 ):
return True, str(cpt) + " Attestation(s) initialisée(s) : "+str(warning_message)
return True, str(cpt)+" Attestation(s) initialisée(s) "
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'initialiser les attestations"
"""
CRON :
Cette fonction prepare et envoi les attestation de formation qui n'a pas déjà été envoyé
a chaque """
def Cron_Send_Attestation_From_Session_By_Email_for_not_sent(Folder, diction):
try:
field_list_obligatoire = ['token', 'session_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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 "
is_warning_message = "0"
warning_message = "Les attestation ont été correctement envoyées par emails avec l'attention suivante : "
for attestation_formation_data in MYSY_GV.dbname['attestation_formation'].find({"session_id":str(diction['session_id']),
'partner_owner_recid':str(my_partner['recid']),
'valide':"1",
"locked":'0',
'statut':'0',
}):
inscrit_email = ""
if (attestation_formation_data and "inscription_id" in attestation_formation_data.keys() and
attestation_formation_data['inscription_id']):
inscrit_data = MYSY_GV.dbname['inscription'].find_one({'partner_owner_recid': str(my_partner['recid']),
'_id': ObjectId(str(attestation_formation_data[
'inscription_id']))})
if (inscrit_data and 'email' in inscrit_data.keys()):
inscrit_email = inscrit_data['email']
print(" ## envoyer l'attestation pour attestation_formation_data = ", attestation_formation_data)
# 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(attestation_formation_data['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"
#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(attestation_formation_data['inscription_id'])
new_diction_no_client['courrier_template_id'] = attestation_formation_data['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']
#print(" ##### new_diction_no_client = ", new_diction_no_client)
tab_saved_file_full_path = []
local_status, local_retval = Cron_Sent_Attestation_Stagiaire_By_Email(tab_saved_file_full_path, Folder, new_diction_no_client)
if (local_status is False):
mycommon.myprint(" WARNING impossible d'envoyer l'attestation de formation a l'apprenant : " + str(attestation_formation_data['inscription_id']) )
is_warning_message = "1"
warning_message = str(warning_message) + " -- " + str(
" WARNING impossible d'envoyer l'attestation de formation a l'apprenant : " + str(inscrit_email))
else:
"""
25/01/2024 : pour loger une action dans la collection ==> courrier_template_tracking_history
"""
local_status, local_retval = module_editique.Editic_Log_History_Action_From_courrier_template_type_document_ref_interne(
my_partner, "ATTESTATION_FORMATION", str(diction['session_id']), 'inscription',
str(attestation_formation_data['inscription_id']),
str(attestation_formation_data['courrier_template_id']))
# Mettre à jour avec la date d'envoi de l'attestation
updata_data = {}
updata_data['date_update'] = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
updata_data['update_by'] = str(my_partner['recid'])
updata_data['statut'] = "1"
updata_data['date_envoie'] = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
MYSY_GV.dbname['attestation_formation'].find_one_and_update({'_id':ObjectId(str(attestation_formation_data['_id'])),
'partner_owner_recid':str(my_partner['recid'])},
{'$set':updata_data})
if (is_warning_message == "1"):
return True, str(warning_message)
return True, " Les attestation 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 attestations par email "
"""
Envoi d'une attestation 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 Cron_Sent_Attestation_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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 = "Attestation_" + 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é " + str(val))
"""
# Ajout de l'evenement dans l'historique
"""
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'] = "Attestation de formation 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)
"""
print(" ### orig_file_name = ", orig_file_name)
print(" ### outputFilename = ", outputFilename)
if (orig_file_name and outputFilename):
new_file = {}
new_file['token'] = diction['token']
type_document = "Attestation"
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(" ## WARNINGGG Impossible de stocker le fichier d'attestation")
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 l'attestation de formation par email "
"""
Cette Fonction permet d'envoyer l'email de notification
au gestionnaire de formation du traitement automatique d'une session.
Par exemple, lorsque le système envoi automatiquement les convocations, les questionnaires de positionnement etc
le gestionnaire de l'organisme de formation doit etre notifié par email
L'email de notification est envoyé au contact principal du partenaire
"""
def Cron_Qualiopi_Automatic_Traitement_Notification(diction):
try:
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'session_id', 'tab_message']
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"
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
# Creation du dictionnaire d'information à utiliser pour la creation du doc
convention_dictionnary_data = {}
new_diction = {}
new_diction['token'] = str(diction['token'])
new_diction['list_stagiaire_id'] = []
new_diction['list_session_id'] = []
new_diction['list_class_id'] = []
new_diction['list_client_id'] = []
new_diction['list_apprenant_id'] = []
new_diction['list_session_id'].append(ObjectId(str(diction['session_id'])))
local_status, local_retval = mycommon.Get_Dictionnary_data_For_Template(new_diction)
if (local_status is False):
return local_status, local_retval
dictionnary_data = local_retval
dictionnary_data['tab_message'] = diction['tab_message']
nb_message = len(diction['tab_message'])
if (nb_message > 0):
"""
Recuperer le modèle de document
"""
local_diction = {}
local_diction['ref_interne'] = "QUALIOPI_AUTOMATIC_DOCUMENT_SENDING"
local_diction['type_doc'] = "email"
local_diction['partner_owner_recid'] = str(my_partner['recid'])
courrier_data_status, courrier_data_retval = mycommon.Get_Courrier_Template_Include_Default_Data(local_diction)
if (courrier_data_status is False):
return courrier_data_status, courrier_data_retval
if ("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 'QUALIOPI_AUTOMATIC_DOCUMENT_SENDING' n'est pas correctement configuré ")
return False, " Le modèle de courrier 'QUALIOPI_AUTOMATIC_DOCUMENT_SENDING' n'est pas correctement configuré "
# Recuperation des donnes smtp
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['recid'])
if (local_stpm_status is False):
return True, " WARNING : Impossible d'envoyer l'email de notification du traitement automatique (2) "
"""
Pour ajouter un statut au sujet du mail,
on va aller regarder dans l'ensemle des statuts des message
(diction_for_notification_email['global_status']).
La liste des valeurs possibles est :warning, true, false
- si on a aucune valeur, alors le statut global sera : "Aucune Action"
- si on a du 'warning' : alors le statut global sera : warning
- si on a du 'false' : alors le statut global sera : warning
- si on a que du 'true' : alors le statut global sera : OK
"""
traitement_global_status = ""
if (len(diction['global_status']) <= 0):
traitement_global_status = "Aucune Action"
elif ("warning" in diction['global_status']):
traitement_global_status = "Warning"
elif ("false" in diction['global_status']):
traitement_global_status = "Warning"
elif ("true" in diction['global_status']):
traitement_global_status = "Ok"
dictionnary_data['traitement_global_status'] = str(traitement_global_status)
body = {
"params": dictionnary_data,
}
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)
smtpserver.ehlo()
smtpserver.starttls()
smtpserver.login(MYSY_GV.O365_SMTP_COUNT_user, MYSY_GV.O365_SMTP_COUNT_password)
# Traitement du sujet du mail
sujet_mail_Template = jinja2.Template(str(courrier_data_retval['sujet']))
sujetHtml = sujet_mail_Template.render(params=body["params"])
# Traitement du corps du mail
contenu_doc_Template = jinja2.Template(str(courrier_data_retval['contenu_doc']))
sourceHtml = contenu_doc_Template.render(params=body["params"])
html_mime = MIMEText(sourceHtml, 'html')
# Creation de l'email à envoyer
msg = MIMEMultipart("alternative")
if (str(partner_own_smtp_value) == "1"):
msg.attach(html_mime)
msg['From'] = partner_SMTP_COUNT_From_User
msg['Bcc'] = 'contact@mysy-training.com'
msg['Subject'] = sujetHtml
tab_destination_mail = []
if ("societe_email" in dictionnary_data['company_data'].keys() and dictionnary_data['company_data'][
'societe_email']):
tab_destination_mail.append(str(dictionnary_data['company_data']['societe_email']))
if ("societe_contact_mail" in dictionnary_data['societe_contact_mail'].keys() and
dictionnary_data['company_data']['societe_contact_mail']):
tab_destination_mail.append(str(dictionnary_data['company_data']['societe_contact_mail']))
toaddrs = ",".join(tab_destination_mail)
dest_email = toaddrs
print(" ### toaddrs 01010 = ", dest_email)
msg['to'] = dest_email
val = smtpserver.send_message(msg)
print(" Email de notification envoyé " + str(val))
else:
msg.attach(html_mime)
msg['From'] = MYSY_GV.O365_SMTP_COUNT_From_User
msg['Bcc'] = 'contact@mysy-training.com'
msg['Subject'] = sujetHtml
tab_destination_mail = []
if( "societe_email" in dictionnary_data['company_data'].keys() and dictionnary_data['company_data']['societe_email']):
tab_destination_mail.append(str(dictionnary_data['company_data']['societe_email']))
if ("societe_contact_mail" in dictionnary_data['company_data'].keys() and dictionnary_data['company_data']['societe_contact_mail'] ):
tab_destination_mail.append(str(dictionnary_data['company_data']['societe_contact_mail']))
toaddrs = ",".join(tab_destination_mail)
dest_email = toaddrs
#print(" ### toaddrs 01010222 = ", dest_email)
msg['to'] = dest_email
val = smtpserver.send_message(msg)
print(" Email de notification envoyé " + str(val))
smtpserver.close()
return True, "L'email de notification 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 l'email de notification du traitement automatique"
"""
CRON :
Initialisation de la liste des emargements pour la paire inscrits / Sequence du jour pour lesquels aucune initialisation n'a pas déja été faite
n'a ete faite.
Important :
Dans le cas précis des emargements, on prend en compte le couple Inscrit / Sequence du jour .
Alog :
Le systeme va aller chercher les sequences du jour pour la session de formation
"""
def Cron_Init_Emargement_By_Date_With_Template(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'session_id', '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]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
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 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 valididé du model de courrier
is_courrier_valide = MYSY_GV.dbname['courrier_template'].count_documents(
{'_id': ObjectId(str(diction['courrier_template_id'])),
'valide': '1',
'locked': '0',
'default_version': '1',
'partner_owner_recid': str(my_partner['recid'])})
if (is_courrier_valide != 1):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Le modèle de courrier '" + str(
diction['courrier_template_id']) + "' n'est pas valide ")
return False, " Le modèle de courrier '" + str(diction['courrier_template_id']) + "' n'est pas valide "
# A présent les inscriptions sont ok, le modele de courrier ok, on peut proceder à l'initialisation des attestations
warning_message = " WARNING : "
is_warning_message = 0
is_courrier_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'])})
cpt = 0
"""
/!\ Cas particulier des émargements
Ici, on va aller chercher les séquences du jour
"""
todays_date = str(date.today().strftime("%d/%m/%Y"))
for session_sequence in MYSY_GV.dbname["session_formation_sequence"].find({"session_id":str(diction["session_id"]),
'valide':'1',
'locked':'0'}):
if( "sequence_start" in session_sequence.keys() ):
date_debut_sequence = str(session_sequence['sequence_start'])[0:10]
local_status = mycommon.CheckisDate(date_debut_sequence)
if (local_status is False):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " La date de debut de séquence " + str(
date_debut_sequence) + " n'est pas au format 'jj/mm/aaaa' ")
return False, " La date de debut de session " + str(
date_debut_sequence) + " n'est pas au format 'jj/mm/aaaa' "
#print(" ### session_sequence cmp de str(todays_date) = ", str(todays_date), " ET str(date_debut_sequence) = ", str(date_debut_sequence) )
if( str(todays_date) == str(date_debut_sequence)):
# Vu qu'il y a une séquence ce jour, on va aller faire le traitement pour chaque inscrit
for my_inscription in MYSY_GV.dbname['inscription'].find({'session_id': str(diction['session_id']),
'status': '1',
'partner_owner_recid': str(my_partner['recid'])}):
"""
01/03/2025 : Regarder si cet inscrit a déjà une initialisation d'emargement faite pour la meme séquence
"""
is_init_emargement_for_today_date_exist_count = MYSY_GV.dbname['emargement'].count_documents(
{'inscription_id':str(my_inscription['_id']),
'partner_owner_recid':str(my_partner['recid']),
'session_id':str(diction['session_id']),
'date': str(todays_date),
'sequence_start': str(session_sequence['sequence_start']),
'sequence_end': str(session_sequence['sequence_end']),
'valide':'1',
'locked':'0'})
if( is_init_emargement_for_today_date_exist_count == 0 ):
# Aucune attestation (ni initialisée, ni envoyé), on peut donc initialiser l'attestation
# On fait volontairement un update pour pallier tout eventuel pb.
query_key = {}
query_key['email'] = my_inscription['email']
query_key['session_id'] = my_inscription['session_id']
query_key['class_internal_url'] = my_inscription['class_internal_url']
query_key['date'] = str(todays_date)
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(my_inscription['_id'])
new_my_local_data['email'] = my_inscription['email']
new_my_local_data['nom'] = my_inscription['nom']
new_my_local_data['prenom'] = my_inscription['prenom']
new_my_local_data['session_id'] = my_inscription['session_id']
new_my_local_data['class_internal_url'] = my_inscription['class_internal_url']
new_my_local_data['date_update'] = str(datetime.now())
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['date'] = str(todays_date)
new_my_local_data['partner_owner_recid'] = str(my_partner['recid'])
ret_val2 = MYSY_GV.dbname['emargement'].find_one_and_update(
query_key,
{"$set": new_my_local_data},
upsert=True,
return_document=ReturnDocument.AFTER
)
if (ret_val2 is None):
mycommon.myprint(str(inspect.stack()[0][3]) + " - WARNING : Impossible d'initialiser l'émargement pour l'inscription : "+str(my_inscription['_id'])+" pour la date du "+str(todays_date))
warning_message = warning_message +" Impossible d'initialiser émargement pour l'inscription : "+str(my_inscription['_id']) +" pour la date du "+str(todays_date) +" - "
is_warning_message = 1
else:
cpt = cpt +1
if( is_warning_message == 1 ):
return True, str(cpt) + " Emargement(s) initialisé(s) : "+str(warning_message)
return True, str(cpt)+" Emargement(s) initialisé(s) "
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'initialiser les émargement"
"""
CRON :
Cette fonction prepare et envoi les demandes d'émargement du JOUR qui n'ont pas déjà été envoyées
/!\ : On envoie uniquement les demandes d'émargement du jours
"""
def Cron_Send_Emargement_By_Date_By_Email_for_not_sent(Folder, diction):
try:
field_list_obligatoire = ['token', 'session_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 la liste des arguments ")
return False, " La valeur '" + val + "' n'est pas presente dans la liste des arguments"
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 "
"""
/!\ Cas particulier des émargements
Ici, on envoyer uniquement les emargement du jour
"""
todays_date = str(date.today().strftime("%d/%m/%Y"))
tab_emargement_ids = []
is_warning_message = "0"
warning_message = "Les demandes d'émargement ont été correctement envoyées par emails avec l'attention suivante : "
# Recuperer la liste des emargement à envoyer pour le jour = str(todays_date)
for emargement_data in MYSY_GV.dbname['emargement'].find({"session_id":str(diction['session_id']),
'partner_owner_recid':str(my_partner['recid']),
'valide':"1",
"locked":'0',
'statut':'0',
"date":str(todays_date)
}):
tab_emargement_ids.append(str(emargement_data['_id']))
if(len(tab_emargement_ids) > 0 ):
tab_emargement_ids_str = ",".join(tab_emargement_ids)
local_payload = {}
local_payload['token'] = str(diction['token'])
local_payload['tab_emargement_ids'] = str(tab_emargement_ids_str)
local_payload['session_id'] = str(diction['session_id'])
status, retval = Inscription_mgt.Create_Emargement_Send_Email_From_Inscription(local_payload)
if( status is False ):
warning_message = str(retval)
is_warning_message = "1"
if (is_warning_message == "1"):
return True, str(warning_message)
return True, "Les demandes d'émargement envoyées par emails", str(len(tab_emargement_ids))
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 demandes d'émargement "