3005 lines
140 KiB
Python
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 "
|