""" Ce document permet de gerer les attestations de formation. Dans le processus d'utilisation, - On initialiser les attestation d'une session - On peut visualiser les attestion - on peut envoyer et renvoyer / télécharger des attestations Pour permettre d'avoir des attestation differentes sur une session, on prendre entrée le tab_ids des inscriptions concernée - Envoi - recuperation des attestation d'une session - etc """ import bson import pymongo from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import prj_common as mycommon import secrets import inspect import sys, os import csv import pandas as pd from pymongo import ReturnDocument import GlobalVariable as MYSY_GV from math import isnan import GlobalVariable as MYSY_GV 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 """ Initialisation de la liste des attestation avec une version du courrier """ def Init_Attestation_Formation_With_Template(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'tab_inscriptions_ids', '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', 'tab_inscriptions_ids', 'courrier_template_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " 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 tab_inscriptions_ids = "" if ("tab_inscriptions_ids" in diction.keys()): if diction['tab_inscriptions_ids']: tab_inscriptions_ids = diction['tab_inscriptions_ids'] tab_inscriptions_ids_splited = str(tab_inscriptions_ids).split(",") # Controle de validité de toutes info avant de commencer à initialiser les attesations for my_inscription in tab_inscriptions_ids_splited: # Verifier que l'inscription est valide my_inscription_is_valide = MYSY_GV.dbname['inscription'].count_documents({'_id':ObjectId(str(my_inscription)), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])}) if( my_inscription_is_valide != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'inscription_id '" + my_inscription + "' n'est pas valide ") return False, " L'inscription_id '" + my_inscription + "' n'est pas valide " # 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', '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 tab_inscriptions_ids_splited: # Verifier que l'inscription est valide my_inscription_data = MYSY_GV.dbname['inscription'].find_one({'_id':ObjectId(str(my_inscription)), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])}) 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['date_envoie'] = "" new_attestation_data['inscription_id'] = str(my_inscription_data['_id']) new_attestation_data['session_id'] = str(my_inscription_data['session_id']) new_attestation_data['courrier_template_id'] = str(is_courrier_data['_id']) update_key = {} update_key['inscription_id'] = str(my_inscription_data['_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_data['_id'])) warning_message = warning_message +" Impossible d'initialiser l'attestation de formation pour l'inscription : "+str(my_inscription_data['_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" """ Initialisation de la liste des attestation pour toutes les inscription d'une session. ici on ne prend pas une inscription specifique """ def Init_Attestation_Formation_With_Template_For_All_Inscription(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 liste ") 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', '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'])}): 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" """ Récuperer les attestatioon d'une session de formation """ def Get_Attestion_By_Session(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'session_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'existe pas") return False, " Les informations fournies sont incorrectes", """ Verification des champs obligatoires """ field_list_obligatoire = field_list = ['token', 'session_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " 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 RetObject = [] val_tmp = 1 for New_retVal in MYSY_GV.dbname['attestation_formation'].find({'session_id':str(diction['session_id']), 'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0'}): user = New_retVal user['id'] = str(val_tmp) val_tmp = val_tmp + 1 # Recuperation des données de l'apprenant concernée qry_filter = {'session_id': str(New_retVal['session_id']), '_id': ObjectId(str(New_retVal['inscription_id'])),'partner_owner_recid': str(New_retVal['partner_owner_recid'])} pipe_qry = ([{'$match': qry_filter}, {'$lookup': {'from': 'apprenant', 'let': {'apprenant_id': '$apprenant_id', 'partner_owner_recid': '$partner_owner_recid'}, 'pipeline': [{'$match': {'$expr': {'$and': [{'$eq': ['$valide', '1']}, {'$eq': ['$_id', {'$convert': { 'input': '$$apprenant_id', 'to': 'objectId', 'onError': {'error': 'true'}, 'onNull': { 'isnull': 'true'}}}]}]}}}], 'as': 'apprenant_collection'}} ]) #print(" ### Get_Attestion_By_Session ici pipe_qry = ", pipe_qry) nom = "" prenom = "" email = "" for local_Insc_retval in MYSY_GV.dbname['inscription'].aggregate(pipe_qry): if ('apprenant_collection' in local_Insc_retval.keys() and len(local_Insc_retval['apprenant_collection']) > 0): nom = str(local_Insc_retval['apprenant_collection'][0]['nom']) prenom = str(local_Insc_retval['apprenant_collection'][0]['prenom']) email = str(local_Insc_retval['apprenant_collection'][0]['email']) else: nom = str(local_Insc_retval['nom']) prenom = str(local_Insc_retval['prenom']) email = str(local_Insc_retval['email']) user['nom'] = nom user['prenom'] = prenom user['email'] = email if( "date_envoie" not in user.keys() ): user['date_envoie'] = "" if ("courrier_template_id" not in user.keys()): user['courrier_template_id'] = "" user['courrier_template_nom'] = "" else: model_courrier_data = MYSY_GV.dbname['courrier_template'].find_one({'_id':ObjectId(str(user['courrier_template_id'])), 'partner_owner_recid':str(my_partner['recid'])}) if("nom" in model_courrier_data.keys() ): user['courrier_template_nom'] = str(model_courrier_data['nom']) else: user['courrier_template_nom'] = "" RetObject.append(mycommon.JSONEncoder().encode(user)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des attestations " """ Cette fonction recupere les differentes modele d'attestation de formation de stagiaire avec des option comme : - ref_interne - nom - type_doc On accepte plusieurs versions du meme doc """ def Get_List_Modele_Attestion_Formation_With_Filter(diction): try: field_list = ['token', 'ref_interne','nom', 'type_doc', 'courrier_template_type_document_ref_interne' ] 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, "Le champ '" + val + "' n'est pas autorisé" field_list_obligatoire = ['token', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][ 3]) + " La valeur '" + val + "' n'est pas presente dans liste ") return False, " La valeur '" + val + "' n'est pas presente dans liste" my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner # Recuperation des option de filter filt_type_doc = {} if ("type_doc" in diction.keys()): filt_type_doc = {'type_doc': str(diction['type_doc'])} filt_nom = {} if ("nom" in diction.keys()): filt_nom = {'nom': str(diction['nom'])} filt_ref_interne = {} if ("ref_interne" in diction.keys()): filt_ref_interne = {'ref_interne': str(diction['ref_interne'])} filt_courrier_template_type_document_ref_interne = {} if ("courrier_template_type_document_ref_interne" in diction.keys()): filt_courrier_template_type_document_ref_interne = {'courrier_template_type_document_ref_interne': str(diction['courrier_template_type_document_ref_interne'])} RetObject = [] val_tmp = 0 """ # Recuperation des documents (collection : courrier_template) de ce partenaire avec 'ref_interne' = 'ATTESTATION_FORMATION' 1 - On regarde le partenaire à des conventions qui sont personnalisées, si non 2 - On va sortir les conventions par defaut de MySy. /!\ : On ne melange pas les 2. c'est l'un ou l'autre """ qry = {'$and': [{'ref_interne': 'ATTESTATION_FORMATION', 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}, filt_courrier_template_type_document_ref_interne, filt_type_doc, filt_nom, filt_ref_interne ]} for retval in MYSY_GV.dbname['courrier_template'].find({'$and': [{'ref_interne': 'ATTESTATION_FORMATION', 'valide': '1', 'locked': '0', 'partner_owner_recid': str(my_partner['recid'])}, filt_courrier_template_type_document_ref_interne, filt_type_doc, filt_nom, filt_ref_interne ]}): user = retval val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) # Si aucune convention personnalisée, on va chercher les conventions mysy if (val_tmp == 0): for retval in MYSY_GV.dbname['courrier_template'].find({'$and': [{'ref_interne': 'ATTESTATION_FORMATION', 'valide': '1', 'locked': '0', 'partner_owner_recid': 'default'}, filt_courrier_template_type_document_ref_interne, filt_type_doc, filt_nom, filt_ref_interne ]}): user = retval val_tmp = val_tmp + 1 RetObject.append(mycommon.JSONEncoder().encode(user)) #print(" ### Get_List_Convocations_Stagiaire_With_Filter RetObject = ", RetObject) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste des attestations de formation" """ Envoi des attestation de formation """ def Send_Attestation_Formation_With_Template(diction): try: diction = mycommon.strip_dictionary(diction) """ Verification des input acceptés """ field_list = ['token', 'tab_inscriptions_ids', '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', 'tab_inscriptions_ids', 'courrier_template_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ") return False, " 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 tab_inscriptions_ids = "" if ("tab_inscriptions_ids" in diction.keys()): if diction['tab_inscriptions_ids']: tab_inscriptions_ids = diction['tab_inscriptions_ids'] tab_inscriptions_ids_splited = str(tab_inscriptions_ids).split(",") # Controle de validité de toutes info avant de commencer à initialiser les attesations for my_inscription in tab_inscriptions_ids_splited: # Verifier que l'inscription est valide my_inscription_is_valide = MYSY_GV.dbname['inscription'].count_documents({'_id':ObjectId(str(my_inscription)), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])}) if( my_inscription_is_valide != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'inscription_id '" + my_inscription + "' n'est pas valide ") return False, " L'inscription_id '" + my_inscription + "' n'est pas valide " # 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', '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 tab_inscriptions_ids_splited: # Verifier que l'inscription est valide my_inscription_data = MYSY_GV.dbname['inscription'].find_one({'_id':ObjectId(str(my_inscription)), 'status':'1', 'partner_owner_recid':str(my_partner['recid'])}) 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_data['_id']) new_attestation_data['session_id'] = str(my_inscription_data['session_id']) new_attestation_data['courrier_template_id'] = str(is_courrier_data['_id']) update_key = {} update_key['inscription_id'] = str(my_inscription_data['_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_data['_id'])) warning_message = warning_message +" Impossible d'initialiser l'attestation de formation pour l'inscription : "+str(my_inscription_data['_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"