Elyos_FI_Back_Office/survey_mgt.py

3600 lines
152 KiB
Python

"""
Ce document permet de gerer les "enquetes" avec utilisation des formulaires
"""
import ast
from zipfile import ZipFile
import bson
import pymongo
import xlsxwriter
from dateutil.relativedelta import relativedelta
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime, date
import module_editique
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
"""
Fontion pour recupérer les données d'une enquete données
"""
def Get_Given_Survey_Data_No_Token(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['survey_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 = [ 'survey_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']
RetObject = []
val_tmp = 1
qry = { 'statut':'1', 'valide': '1', 'locked': '0', '_id':ObjectId(str(diction['survey_id']))}
print(" qry == ", qry)
for New_retVal in MYSY_GV.dbname['survey'].find(qry).sort([("_id", pymongo.DESCENDING), ]):
user = New_retVal
user['id'] = str(val_tmp)
val_tmp = val_tmp + 1
class_title = ""
class_internal_url = ""
class_id = ""
session_code = ""
session_title = ""
session_code_date_debut = ""
session_code_date_fin = ""
if( "session_id" in New_retVal.keys() and New_retVal['session_id']):
# On va recuperer les données de la session
my_session_data = MYSY_GV.dbname['session_formation'].find_one( {"partner_owner_recid": str(New_retVal['partner_owner_recid']), 'valide': '1', '_id':ObjectId(str(New_retVal['session_id']))})
if( my_session_data and "class_internal_url" in my_session_data.keys() ):
my_class_data = MYSY_GV.dbname['myclass'].find_one(
{"partner_owner_recid":str(New_retVal['partner_owner_recid']), 'valide': '1',
'internal_url': str(my_session_data['class_internal_url'])})
if( my_class_data and "title" in my_class_data.keys() ):
class_title = my_class_data['title']
class_internal_url = my_class_data['internal_url']
class_id = str(my_class_data['_id'])
session_code = my_session_data['code_session']
session_title = my_session_data['titre']
session_code_date_debut = my_session_data['date_debut']
session_code_date_fin = my_session_data['date_fin']
user['class_title'] = class_title
user['class_internal_url'] = class_internal_url
user['class_id'] = class_id
user['session_code'] = session_code
user['session_title'] = session_title
user['session_code_date_debut'] = session_code_date_debut
user['session_code_date_fin'] = session_code_date_fin
"""
Recuperation des données du formulaire (question)
"""
list_questions = []
formulaire_message_introduction = ""
if( "formulaire_id" in New_retVal.keys() ):
my_formulaire_data = MYSY_GV.dbname['formulaire'].find_one( {"partner_owner_recid": str(New_retVal['partner_owner_recid']), 'valide': '1', 'locked':'0', '_id':ObjectId(str(New_retVal['formulaire_id']))})
if(my_formulaire_data and "list_questions" in my_formulaire_data.keys() ):
list_questions = my_formulaire_data['list_questions']
if( "message_introduction" in my_formulaire_data.keys() ):
formulaire_message_introduction = my_formulaire_data['message_introduction']
user['list_questions'] = list_questions
user['message_introduction'] = formulaire_message_introduction
"""
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 New_retVal.keys() and New_retVal['related_collection'] and New_retVal['related_collection'] == "ressource_humaine" and
'related_collection_id' in New_retVal.keys() and New_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(New_retVal['related_collection_id'])),
'valide': '1', 'locked': '0', 'partner_recid': str(New_retVal['partner_owner_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(New_retVal['related_collection_id'])),
'valide': '1', 'locked': '0', 'partner_recid': str(New_retVal['partner_owner_recid'])}, {'_id':1, 'nom':1, 'prenom':1, 'email':1, 'civilite':1})
if (related_rh_data and "_id" in related_rh_data.keys()):
user["related_rh_data_is_valide"] = "1"
else:
user["related_rh_data_is_valide"] = "0"
if (related_rh_data and "civilite" in related_rh_data.keys()):
user["related_rh_data_civilite"] = str(related_rh_data['civilite']).capitalize()
else:
user["related_rh_data_is_valide"] = "neutre".capitalize()
if( related_rh_data and "email" in related_rh_data.keys()):
user["related_rh_data_email"] = str(related_rh_data['email'])
else:
user["related_rh_data_email"] = ""
if (related_rh_data and "prenom" in related_rh_data.keys()):
user["related_rh_data_prenom"] = str(related_rh_data['prenom'])
else:
user["related_rh_data_prenom"] = ""
if (related_rh_data and "nom" in related_rh_data.keys()):
user["related_rh_data_nom"] = str(related_rh_data['nom'])
else:
user["related_rh_data_nom"] = ""
RetObject.append(mycommon.JSONEncoder().encode(user))
#print(" ### 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 les données "
"""
Cette fonction enregistrer le resultat d'un survey
"""
def Record_Survey_Data_No_Token(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['survey_id', "list_response" ]
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 = [ 'survey_id', "list_response" ]
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']
"""
Verifier que l'enquete est valide
"""
qry = {'statut': '1', 'valide': '1', 'locked': '0', '_id': ObjectId(str(diction['survey_id']))}
is_valide_survey = MYSY_GV.dbname['survey'].count_documents(qry)
if( is_valide_survey <= 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant de l'enquête est invalide ")
return False, " L'identifiant de l'enquête est invalide "
"""
Recuperation des données du formulaire
"""
survey_data = MYSY_GV.dbname['survey'].find_one(qry)
form_data_qry = { 'valide': '1', 'locked': '0', '_id': ObjectId(str(survey_data['formulaire_id'])),
'partner_owner_recid':str(survey_data['partner_owner_recid'])}
form_data = MYSY_GV.dbname['formulaire'].find_one(form_data_qry)
if( form_data is None ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant du formulaire est invalide ")
return False, " L'identifiant du formulaire est invalide "
if( "list_questions" not in form_data.keys() ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Le questionnaire du formulaire est invalide ")
return False, " Le questionnaire du formulaire est invalide "
form_list_question = form_data['list_questions']
JSON_user_response = ast.literal_eval(diction['list_response'])
list_keys = list(JSON_user_response.keys())
global_response = []
for my_key in list_keys:
question_id = my_key
response = JSON_user_response[my_key]
"""
On va recuperer la question ayant l'_id = question_id
"""
associated_question = [x for x in form_list_question if x['_id'] == question_id][0]['question']
node = {"question_id":question_id, "question":str(associated_question), "response":response}
"""
05/07/204 - On va aller recuperer l'info 'is_statistic' et le 'type' sur la question
"""
question_is_statistic = ""
question_type = ""
for local_question in form_list_question :
if( str(local_question['_id']) == str(question_id) and "is_statistic" in local_question.keys() ):
question_is_statistic = local_question['is_statistic']
if (str(local_question['_id']) == str(question_id) and "type" in local_question.keys()):
question_type = local_question['type']
node['is_statistic'] = question_is_statistic
node['type'] = question_type
global_response.append(node)
"""
On va enregister les reponse et changer le statut
"""
mydata = {}
mydata['date_update'] = str(datetime.now())
mydata['date_reponse'] = str(datetime.now())
mydata['statut'] = "2"
mydata['user_response'] = global_response
result = MYSY_GV.dbname['survey'].find_one_and_update(
{'statut': '1', 'valide': '1', 'locked': '0', '_id': ObjectId(str(diction['survey_id']))},
{"$set": mydata},
upsert=False,
return_document=ReturnDocument.AFTER
)
if (result is None or "_id" not in result.keys()):
mycommon.myprint(
" Impossible de mettre à jour le formulaire (2) ")
return False, " Impossible de mettre à jour le formulaire (2) "
return True, " Le formulaire a été correctement mis à jour"
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'enregister les informations "
"""
Survey for enduser : Cette fonction créer une enquete en prenant à partir d'une liste d'inscrit
/!\ : Cette fonction ne concerne uniquement et exclusibement les inscriptions validées (status = 1)
"""
def Add_Survey_Tab_Inscrit(diction):
try:
field_list_obligatoire = ['session_id', 'token', 'tab_ids', 'formulaire_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"
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
my_inscription_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
my_inscription_ids = diction['tab_ids']
tab_my_inscription_ids = str(my_inscription_ids).split(",")
for my_inscription_id in tab_my_inscription_ids:
# Verifier que les inscriptions sont valides pour cette session
qry = {'session_id': str(diction['session_id']),
'_id':ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(my_partner['recid']),
'valide':'1', 'status':'1'}
print(" qry2 === ", qry)
tmp_count = MYSY_GV.dbname['inscription'].count_documents({'session_id': str(diction['session_id']),
'_id':ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(my_partner['recid']),
'valide':'1', 'status':'1'})
if (tmp_count <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'apprenant est invalide ")
return False, " L'identifiant de l'apprenant est invalide "
"""
Verifier la valididé du formulaire
"""
is_valide_form = MYSY_GV.dbname['formulaire'].count_documents({'_id':ObjectId(str(diction['formulaire_id'])),
'valide':'1',
'locked':'0',
'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 preceder à la creation de l'enquete
"""
for my_inscription_id in tab_my_inscription_ids:
# Recuperation des données de l'inscript
req = {'session_id': str(diction['session_id']),
'_id':ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'status':'1'
}
#print(" ### req = ", req)
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 "
mytoday = datetime.today().strftime("%d/%m/%Y")
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['sending_date'] = str(mytoday)
local_status, local_retval = 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 "
"""
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
"""
def Init_Survey_Tab_For_All_Session_Inscrit(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 = []
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.append(str(my_inscription_id['_id']))
"""
Verifier la validité du formulaire
"""
is_valide_form = MYSY_GV.dbname['formulaire'].count_documents({'_id':ObjectId(str(diction['formulaire_id'])),
'valide':'1',
'locked':'0',
'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 - suppression des data existate et
2 - la creation de l'enquete
"""
MYSY_GV.dbname['survey'].delete_many({'partner_owner_recid':str(my_partner['recid']),
'session_id':str(diction['session_id']),
'survey_type':str(diction['survey_type'])})
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 = 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 "
"""
/!\ à l'image de la fonction "Init_Survey_Tab_For_All_Session_Inscrit",
cette fonction permet de créer une enquete pour un related_collection et une liste de related_collection_id
données.
"""
def Init_Survey_Tab_For_All_Except_Inscription(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', 'tab_related_collection_ids', '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 accepté ")
return False, " Les informations fournies sont incorrectes",
field_list_obligatoire = ['token', 'formulaire_id', 'survey_type', 'related_collection', 'tab_related_collection_ids']
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][
3]) + " La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " 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
"""
si la session est dans le dictionnaire, Verifier que la session est valide
"""
session_id = ""
if( "session_id" in diction and diction['session_id'] ):
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 "
session_id = str(diction['session_id'])
"""
si la class_id est dans le dictionnaire, Verifier que la formation est valide
"""
class_id = ""
if ("class_id" in diction and diction['class_id']):
is_valide_class = MYSY_GV.dbname['class_id'].count_documents(
{'_id': ObjectId(str(diction['class_id'])),
'valide': '1',
'partner_owner_recid': str(my_partner['recid'])})
if (is_valide_class <= 0):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de la formation est invalide ")
return False, " L'identifiant de la formation est invalide "
class_id = str(diction['class_id'])
tab_related_collection_ids = ""
if ("tab_related_collection_ids" in diction.keys() and diction['tab_related_collection_ids']):
tab_related_collection_ids = diction['tab_related_collection_ids']
tab_related_collection_ids_splited = str(tab_related_collection_ids).split(",")
"""
Verifier la validité des tab_related_collection_ids_splited
"""
qry_partner_owner_recid = "partner_owner_recid"
if( str(diction['related_collection']) == "ressource_humaine"):
qry_partner_owner_recid = "partner_recid"
for tmp_val in tab_related_collection_ids_splited:
is_tmp_val_valide = MYSY_GV.dbname[str(diction['related_collection'])].count_documents({'_id':ObjectId(str(tmp_val)), 'valide':'1', 'locked':'0',
str(qry_partner_owner_recid):str(my_partner['recid'])})
if( is_tmp_val_valide != 1 ):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de "+str(diction['related_collection']) +" est invalide ")
return False, " L'identifiant de "+str(diction['related_collection']) +" est invalide "
"""
Verifier la validité du formulaire
"""
is_valide_form = MYSY_GV.dbname['formulaire'].count_documents({'_id':ObjectId(str(diction['formulaire_id'])),
'valide':'1',
'locked':'0',
'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 - suppression des data existate et
2 - la creation de l'enquete
"""
for tmp_val in tab_related_collection_ids_splited:
MYSY_GV.dbname['survey'].delete_many({'partner_owner_recid':str(my_partner['recid']),
'related_collection':str(diction['related_collection']),
'related_collection_id':str(tmp_val),
'survey_type':str(diction['survey_type'])})
sending_date = ""
if( "sending_date" in diction ):
sending_date = diction['sending_date']
for tmp_val in tab_related_collection_ids_splited:
# Recuperation des données de l'inscript
personne_data = MYSY_GV.dbname[str(diction['related_collection'])].find_one(
{'_id': ObjectId(str(tmp_val)), 'valide': '1', 'locked': '0',
str(qry_partner_owner_recid): str(my_partner['recid'])})
if(personne_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 "
server_diction = {}
server_diction['token'] = diction['token']
server_diction['class_id'] = str(class_id)
server_diction['session_id'] = str(session_id)
server_diction['inscrit_id'] = ""
server_diction['formulaire_id'] = diction['formulaire_id']
server_diction['survey_type'] = diction['survey_type']
server_diction['sending_date'] = sending_date
server_diction['related_collection'] = str(diction['related_collection'])
server_diction['related_collection_id'] = str(tmp_val)
local_status, local_retval = Create_One_Survey_To_Except_Inscrit(server_diction)
if( local_status is False):
return local_status, local_retval
print(" ### local_status = ", local_status)
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 "
"""
Cette fonction envoi un formulaire pour inscrit
"""
def 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
"""
my_inscrit_data = MYSY_GV.dbname['inscription'].find_one({'_id': ObjectId(str(diction['inscrit_id'])),
'valide': '1',
'locked': '0',
'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'])
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 "
"""
Cette fonction crée un formulaire pour tous EXCEPTE les inscrit
"""
def Create_One_Survey_To_Except_Inscrit(diction):
try:
field_list_obligatoire = ['session_id', 'token', 'related_collection', 'related_collection_id', 'formulaire_id', 'survey_type',
'sending_date', 'class_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"
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
"""
if( diction['session_id'] ):
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
"""
if (diction['class_id']):
my_class_data = MYSY_GV.dbname['myclass'].find_one({'_id': ObjectId(str(diction['class_id'])),
'valide': '1',
'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
"""
qry_partner_owner_recid = "partner_owner_recid"
if (str(diction['related_collection']) == "ressource_humaine"):
qry_partner_owner_recid = "partner_recid"
my_inscrit_data = MYSY_GV.dbname[str(diction['related_collection'])].find_one({'_id': ObjectId(str(diction['related_collection_id'])),
'valide': '1',
'locked': '0',
str(qry_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 "
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(diction['class_id'])
new_survey_data['survey_type'] = str(diction['survey_type'])
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['related_collection'] = diction['related_collection']
new_survey_data['related_collection_id'] = diction['related_collection_id']
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(" ### new_survey_data = ", new_survey_data)
inserted_id = MYSY_GV.dbname['survey'].insert_one(new_survey_data).inserted_id
print(" ### new_survey_data inserted_id = ", 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 "
"""
Cette fonction permet d'envoyer la demande d'enquete à la personne concerné
/!\ : Cette fontion permet d'envoyer toutes les demande d'enquete, que soit :
un questionnaire de positionnement, une demande d'evaluation a chaud,
une demande d'evaluation à froid, ou tout autre.
Le type de courrier à utiliser, le 'courrier_template_type_document_ref_interne' à loger dans tracking history
depende du type d'enquete.
par exemple :
- si survey.survey_type == 'pos' ==> courrier_template_type_document_ref_interne = "QUESTION_POSITIONNEMENT" et
courrier_template.ref_interne = QUESTION_POSITIONNEMENT.
- si survey.survey_type == 'hot_eval' ==> courrier_template_type_document_ref_interne = "EVAL_FORMATION" et
courrier_template.ref_interne = EVAL_FORMATION.
/!\ : 21/07/2024 : Dans le cas d'un Renvoi, on va desactiver la question precedenmment envoyée.
"""
def Send_Survey_TabIds(diction):
try:
field_list_obligatoire = [ 'token', 'tab_ids']
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][
3]) + " La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " 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
my_survey_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
my_survey_ids = diction['tab_ids']
tab_my_survey_ids = str(my_survey_ids).split(",")
for my_survey_id in tab_my_survey_ids:
# Verifier que les survey sont valides
qry = { '_id':ObjectId(str(my_survey_id)), 'partner_owner_recid': str(my_partner['recid']),'valide':'1', 'locked':'0'}
tmp_count = MYSY_GV.dbname['survey'].count_documents(qry)
if (tmp_count <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'enquete "+str(my_survey_id)+" invalide ")
return False, " L'identifiant de l'enquete "+str(my_survey_id)+" invalide "
"""
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'}
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 "
"""
Fonction qui permet d'envoyer des demande d'enquete avec le job/cron
Ici on a pas besoin du cron, mais recuperer directement le 'partner_owner_recid'
de la ligne à traiter
"""
def Automatic_Send_Survey_TabIds(diction):
try:
field_list_obligatoire = [ 'partner_owner_recid', 'tab_ids']
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][
3]) + " La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes"
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
local_status, my_partner = mycommon.Get_Connected_User_Partner_Data_From_RecID(str(diction['partner_owner_recid']))
if (local_status is not True):
return local_status, my_partner
my_survey_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
my_survey_ids = diction['tab_ids']
tab_my_survey_ids = str(my_survey_ids).split(",")
for my_survey_id in tab_my_survey_ids:
# Verifier que les survey sont valides
qry = { '_id':ObjectId(str(my_survey_id)), 'partner_owner_recid': str(my_partner['recid']),'valide':'1', 'locked':'0'}
print(" qry2 === ", qry)
tmp_count = MYSY_GV.dbname['survey'].count_documents(qry)
if (tmp_count <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'enquete "+str(my_survey_id)+" invalide ")
return False, " L'identifiant de l'enquete "+str(my_survey_id)+" invalide "
"""
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'}
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 "
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'] = my_partner['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'])
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,
{'$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 result.keys() and result['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( "inscription_id" in local_survey_retval.keys() ):
# Cette demande d'enquete concerne une inscription
local_inscription_id = local_survey_retval['inscription_id']
qry = {'_id': ObjectId(str(local_inscription_id)),
'status': '1',
'partner_owner_recid': str(my_partner['recid'])}
#print(" ### qry 010101 = ", qry)
inscription_data = MYSY_GV.dbname['inscription'].find_one({'_id': ObjectId(str(local_inscription_id)),
'status': '1',
'partner_owner_recid': str(my_partner['recid'])})
if( inscription_data is None or "session_id" not in inscription_data.keys() ):
mycommon.myprint(
" WARNING : Impossible de logguer l'historique pour l'inscrit : " + str(local_inscription_id))
#print(" inscription_data = ", inscription_data)
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 "
"""
Fonction pour Recuperer la liste des enquete avec les filtre suivant :
- session_id,
- formulaire_type - ['pos', 'hot_eval', 'cold_eval', 'human_eval', 'autre_eval']
"""
def Get_List_Survey_with_filter(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'session_id', 'formulaire_type', '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"
"""
Verification des champs obligatoires
"""
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 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
filt_session_id = {}
if ("session_id" in diction.keys()):
filt_session_id = {'session_id': str(diction['session_id'])}
filt_related_collection = {}
if ("related_collection" in diction.keys()):
filt_related_collection = {'related_collection': str(diction['related_collection'])}
filt_related_collection_id = {}
if ("related_collection_id" in diction.keys()):
related_collection_id = {'related_collection_id': str(diction['related_collection_id'])}
RetObject = []
val_tmp = 0
qry = {"partner_owner_recid": str(my_partner['recid']), 'valide': '1', 'locked': '0'}
query_with_filter = {'$and': [qry, filt_session_id, filt_related_collection, filt_related_collection_id]}
pipe_qry = ([
{'$match': query_with_filter },
{'$project': { 'valide': 0, 'locked': 0, }},
{'$lookup': {
'from': 'formulaire',
"let": {'formulaire_id': "$formulaire_id",
'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [
{'$match':
{'$expr':
{'$and':
[
{'$eq': ["$_id", {'$convert': {
'input': "$$formulaire_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
{'$eq': ["$valide", "1"]},
{'$eq': ["$type", str(diction['formulaire_type'])]},
{'$eq': ["$partner_owner_recid", '$$partner_owner_recid']}
]
}
}
},
],
'as': 'formulaire'
}
},
{
'$unwind': '$formulaire'
}
])
#print(" ### pipe_qry = ", pipe_qry)
for New_retVal in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
user = New_retVal
user['id'] = str(val_tmp)
# recuperer les données de la fromation
class_title = ""
class_external_code = ""
if( "class_id" in New_retVal.keys() ):
class_data = MYSY_GV.dbname['myclass'].find_one({'_id':ObjectId(New_retVal['class_id']),
'valide':'1',
'partner_owner_recid':my_partner['recid']},
{'title':1, 'external_code':1})
if( class_data and 'title' in class_data.keys() and 'external_code' in class_data.keys()):
class_title = class_data['title']
class_external_code = class_data['external_code']
user['class_title'] = class_title
user['class_external_code'] = class_external_code
# Recuperer les données de la session
session_title = ""
session_code = ""
if ("session_id" in New_retVal.keys()):
session_data = MYSY_GV.dbname['session_formation'].find_one({'_id': ObjectId(New_retVal['session_id']),
'valide': '1',
'partner_owner_recid': my_partner['recid']},
{'titre': 1, 'code_session': 1})
if (session_data and 'titre' in session_data.keys() and 'code_session' in session_data.keys()):
session_title = session_data['titre']
session_code = session_data['code_session']
user['session_title'] = session_title
user['session_code'] = session_code
"""
Recuperation des info de la ressource humaine
"""
rh_nom = ""
rh_prenom = ""
if( "survey_type" in New_retVal.keys() and New_retVal["survey_type"] == "human_eval" and
"related_collection" in New_retVal.keys() and New_retVal["related_collection"] == "ressource_humaine" and
"related_collection_id" in New_retVal.keys() and New_retVal["related_collection_id"] ):
rh_data = MYSY_GV.dbname['ressource_humaine'].find_one({'_id':ObjectId(New_retVal["related_collection_id"]),
'valide':'1',
'locked':'0',
'partner_recid':str(my_partner['recid'])}, {'nom':1, "prenom":1})
if(rh_data and "nom" in rh_data.keys()):
rh_nom = rh_data['nom']
if (rh_data and "prenom" in rh_data.keys()):
rh_prenom = rh_data['prenom']
user['rh_nom'] = rh_nom
user['rh_prenom'] = rh_prenom
val_tmp = val_tmp + 1
RetObject.append(mycommon.JSONEncoder().encode(user))
return True, RetObject
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False, " Impossible de récupérer la liste des formulaires "
"""
Cette fonction pour exporter les resultat d'une enquete dans une fichier excel
"""
def Export_To_Excel_Survey_with_filter(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'session_id', 'formulaire_type' , '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"
"""
Verification des champs obligatoires
"""
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 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
filt_session_id = {}
if ("session_id" in diction.keys()):
filt_session_id = {'session_id': str(diction['session_id'])}
filt_related_collection = {}
if ("related_collection" in diction.keys()):
filt_related_collection = {'related_collection': str(diction['related_collection'])}
filt_related_collection_id = {}
if ("related_collection_id" in diction.keys()):
related_collection_id = {'related_collection_id': str(diction['related_collection_id'])}
"""
Recuperer les 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 de formation est invalide ")
return False, " L'identifiant de la session de formation est invalide "
"""
Recuperer les données de la formation
"""
my_class_data = MYSY_GV.dbname['myclass'].find_one({'internal_url': str(my_session_data['class_internal_url']),
'valide': '1',
'partner_owner_recid': str(
my_partner['recid'])}, {'title':1})
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 "
RetObject = []
val_tmp = 0
qry = {"partner_owner_recid": str(my_partner['recid']), 'valide': '1', 'locked': '0'}
query_with_filter = {'$and': [qry, filt_session_id, filt_related_collection, filt_related_collection_id]}
pipe_qry = ([
{'$match': query_with_filter },
{'$project': { 'valide': 0, 'locked': 0, }},
{'$lookup': {
'from': 'formulaire',
"let": {'formulaire_id': "$formulaire_id",
'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [
{'$match':
{'$expr':
{'$and':
[
{'$eq': ["$_id", {'$convert': {
'input': "$$formulaire_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
{'$eq': ["$valide", "1"]},
{'$eq': ["$type", str(diction['formulaire_type'])]},
{'$eq': ["$partner_owner_recid", '$$partner_owner_recid']}
]
}
}
},
],
'as': 'formulaire'
}
},
{
'$unwind': '$formulaire'
}
])
#print(" ### pipe_qry = ", pipe_qry)
ts = datetime.now().timestamp()
ts = str(ts).replace(".", "").replace(",", "")[-5:]
orig_file_name = "Export_Reponse_" + str(ts) + ".xlsx"
outputFilename = str(MYSY_GV.TEMPORARY_DIRECTORY) + "/" + str(orig_file_name)
tab_exported_fields_header = ["formation", "code_session", "date_debut_session",
"date_fin_session", "email", "nom", "prenom", "type questionnaire", "date_envoi",
"date_reponse", "question", "reponse"]
tab_exported_fields = ["formation", "code_session", "date_debut_session",
"date_fin_session", "email", "nom", "prenom", "date_envoi",
"date_reponse", "question", "reponse"]
# Create a workbook and add a worksheet.
workbook = xlsxwriter.Workbook(outputFilename)
worksheet = workbook.add_worksheet()
row = 0
column = 0
"""
Creation de l'entete du fichier excel
"""
for header_item in tab_exported_fields_header:
worksheet.write(row, column, header_item)
column += 1
"""
Creation des data du fichier excel
"""
for Survey_Data in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
column = 0
row = row + 1
# Champ : Titre de la formation
no_html_title = ""
if ("title" in my_class_data.keys()):
no_html_title = my_class_data['title']
worksheet.write(row, column, no_html_title)
column += 1
# Champ : Code session
code_session = ""
if ("code_session" in my_session_data.keys()):
code_session = my_session_data['code_session']
worksheet.write(row, column, code_session)
column += 1
# Champ : Date debut session
date_debut_session = ""
if ("code_session" in my_session_data.keys()):
date_debut_session = my_session_data['date_debut']
worksheet.write(row, column, date_debut_session)
column += 1
# Champ : Date fin session
date_fin_session = ""
if ("date_fin" in my_session_data.keys()):
date_fin_session = my_session_data['date_fin']
worksheet.write(row, column, date_fin_session)
column += 1
# Champ : Email
email = ""
if ("email" in Survey_Data.keys()):
email = Survey_Data['email']
worksheet.write(row, column, email)
column += 1
# Champ : nom
nom = ""
if ("nom" in Survey_Data.keys()):
nom = Survey_Data['nom']
worksheet.write(row, column, nom)
column += 1
# Champ : prenom
prenom = ""
if ("prenom" in Survey_Data.keys()):
prenom = Survey_Data['prenom']
worksheet.write(row, column, prenom)
column += 1
# Champ : survey_type
worksheet.write(row, column, str(diction['formulaire_type']))
column += 1
# Champ : date_envoi
date_envoi = ""
if ("date_envoi" in Survey_Data.keys()):
date_envoi = str(Survey_Data['date_envoi'])[0:16]
worksheet.write(row, column, date_envoi)
column += 1
# Champ : date_reponse
date_reponse = ""
if ("date_reponse" in Survey_Data.keys()):
date_reponse = str(Survey_Data['date_reponse'])[0:16]
worksheet.write(row, column, date_reponse)
column += 1
# Gestion des question et reponse
if( "user_response" in Survey_Data.keys() ):
question_num = 1
for user_response in Survey_Data['user_response']:
question = ""
if ("question" in user_response.keys()):
question = str(user_response['question'])
#worksheet.write(0, column, question)
#worksheet.write(row, column, question)
column += 1
#worksheet.write(0, column, "response_"+str(question_num))
worksheet.write(0, column, question)
response = ""
if ("response" in user_response.keys()):
response = str(user_response['response'])
worksheet.write(row, column, response)
column += 1
question_num += 1
else:
#question = "---"
#worksheet.write(row, column, question)
column += 1
response = "---"
worksheet.write(row, column, response)
column += 1
workbook.close()
if os.path.exists(outputFilename):
# print(" ### ok os.path.exists(outputFilename) "+str(outputFilename))
return True, send_file(outputFilename, as_attachment=True)
return False, "Impossible de générer l'export (2) "
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 l'export "
"""
Cette fonction pour exporter les resultat d'une enquete dans une fichier excel :
POUR UNE LISTE _ID
"""
def Export_To_Excel_Survey_with_filter_Tab_Survey_Id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'tab_ids', 'type' ]
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_ids', '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"
"""
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
my_survey_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
my_survey_ids = diction['tab_ids']
my_survey_type = ""
if( "type" in diction.keys() ):
my_survey_type = diction['type']
my_survey_ids = my_survey_ids.replace(".", ",")
tab_my_survey_ids = str(my_survey_ids).split(",")
tab_my_survey_ids = [s.strip() for s in tab_my_survey_ids]
"""
Verifier la validité des survey
"""
for my_survey_id in tab_my_survey_ids:
# Verifier qui la formation n'a pas deja été evaluée
tmp_count = MYSY_GV.dbname['survey'].count_documents({'_id':ObjectId(str(my_survey_id)),
'partner_owner_recid': str(my_partner['recid']),
'valide':'1',
'locked':'0'
})
if (tmp_count != 1):
mycommon.myprint(str(inspect.stack()[0][3]) + " L'identifiant de l'enquête "+str(my_survey_id) +" est invalide ")
return False, " L'identifiant de l'enquête "+str(my_survey_id) +" est invalide "
filt_session_id = {}
if ("session_id" in diction.keys()):
filt_session_id = {'session_id': str(diction['session_id'])}
filt_related_collection = {}
if ("related_collection" in diction.keys()):
filt_related_collection = {'related_collection': str(diction['related_collection'])}
filt_related_collection_id = {}
if ("related_collection_id" in diction.keys()):
related_collection_id = {'related_collection_id': str(diction['related_collection_id'])}
RetObject = []
val_tmp = 0
ts = datetime.now().timestamp()
ts = str(ts).replace(".", "").replace(",", "")[-5:]
orig_file_name = "Export_Reponse_" + str(ts) + ".xlsx"
outputFilename = str(MYSY_GV.TEMPORARY_DIRECTORY) + "/" + str(orig_file_name)
tab_exported_fields_header = ["formation", "code_session", "date_debut_session",
"date_fin_session", "email", "nom", "prenom", "type questionnaire", "date_envoi",
"date_reponse", "question", "reponse"]
tab_exported_fields = ["formation", "code_session", "date_debut_session",
"date_fin_session", "email", "nom", "prenom", "date_envoi",
"date_reponse", "question", "reponse"]
""""
Pour créer le document PDF, on va recuperer les données du Get_Dictionnary_data_For_Template
"""
new_diction = {}
new_diction['token'] = diction['token']
new_diction['list_stagiaire_id'] = []
new_diction['list_session_id'] = []
new_diction['list_class_id'] = []
new_diction['list_client_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
"""
Recuperer le modèle de document
"""
local_diction = {}
local_diction['ref_interne'] = "EVALUATION_MODEL_PDF"
local_diction['type_doc'] = "pdf"
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 'EVALUATION_MODEL_PDF' n'est pas correctement configuré ")
return False, " Le modèle de courrier 'EVALUATION_MODEL_PDF' n'est pas correctement configuré "
# Stokage des nom de fichier à zipper
list_file_name_to_zip = []
"""
Creation des data du fichier excel
"""
for my_survey_id in tab_my_survey_ids:
qry = {"partner_owner_recid": str(my_partner['recid']), 'valide': '1', 'locked': '0',
'_id':ObjectId(str(my_survey_id))}
query_with_filter = {'$and': [qry, filt_session_id, filt_related_collection, filt_related_collection_id]}
#print(" ## qry = ", qry)
#print(" ## my_survey_id = ", my_survey_id)
pipe_qry = ([
{'$match': query_with_filter},
{'$project': {'valide': 0, 'locked': 0, }},
{'$lookup': {
'from': 'formulaire',
"let": {'formulaire_id': "$formulaire_id",
'partner_owner_recid': '$partner_owner_recid'},
'pipeline': [
{'$match':
{'$expr':
{'$and':
[
{'$eq': ["$_id", {'$convert': {
'input': "$$formulaire_id",
'to': "objectId",
'onError': {'error': 'true'},
'onNull': {'isnull': 'true'}
}}]},
{'$eq': ["$valide", "1"]},
{'$eq': ["$partner_owner_recid", '$$partner_owner_recid']}
]
}
}
},
],
'as': 'formulaire'
}
},
{
'$unwind': '$formulaire'
}
])
#print(" ### Export_To_Excel_Survey_with_filter_Tab_Survey_Id pipe_qry = ", pipe_qry)
for Survey_Data in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
param_survey_data = {}
"""
Recuperer les données de la session
"""
my_session_data = MYSY_GV.dbname['session_formation'].find_one(
{'_id': ObjectId(str(Survey_Data['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 de formation est invalide ")
return False, " L'identifiant de la session de formation est invalide "
"""
11/02/2025 -
Si c'est l'evaluation d'un enseignant par un apprenant, il faut recuprer les données de l'enseignant pour l'afficher dans la fiche
"""
evaluation_formateur = "0"
formateur_nom = ""
formateur_prenom = ""
formateur_email = ""
if( Survey_Data['survey_type'] == "human_eval" and Survey_Data['related_collection'] == "ressource_humaine" and "related_collection_id" in Survey_Data.keys()
and Survey_Data['related_collection_id']):
evaluation_formateur = "1"
formateur_data = MYSY_GV.dbname['ressource_humaine'].find_one(
{'_id': ObjectId(str(Survey_Data['related_collection_id'])),
'valide': '1',
'partner_recid': str(
my_partner['recid'])}, {'nom': 1, 'prenom':1, 'email':1})
if( formateur_data and 'nom' in formateur_data.keys() ):
formateur_nom = formateur_data['nom']
if (formateur_data and 'prenom' in formateur_data.keys()):
formateur_prenom = formateur_data['prenom']
if (formateur_data and 'email' in formateur_data.keys()):
formateur_email = formateur_data['email']
param_survey_data['evaluation_formateur'] = evaluation_formateur
param_survey_data['formateur_evalue_nom'] = formateur_nom
param_survey_data['formateur_evalue_prenom'] = formateur_prenom
param_survey_data['formateur_evalue_email'] = formateur_email
"""
Recuperer les données de la formation
"""
my_class_data = MYSY_GV.dbname['myclass'].find_one(
{'_id': ObjectId(str(Survey_Data['class_id'])),
'valide': '1',
'partner_owner_recid': str(
my_partner['recid'])}, {'title': 1})
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 "
# Champ : Titre de la formation
no_html_title = ""
if ("title" in my_class_data.keys()):
no_html_title = my_class_data['title']
param_survey_data['class_title'] = my_class_data['title']
# Champ : Code session
code_session = ""
if ("code_session" in my_session_data.keys()):
code_session = my_session_data['code_session']
param_survey_data['code_session'] = my_session_data['code_session']
# Champ : Date debut session
date_debut_session = ""
if ("code_session" in my_session_data.keys()):
date_debut_session = my_session_data['date_debut']
param_survey_data['session_date_debut'] = my_session_data['date_debut']
# Champ : Date fin session
date_fin_session = ""
if ("date_fin" in my_session_data.keys()):
date_fin_session = my_session_data['date_fin']
param_survey_data['session_date_fin'] = my_session_data['date_fin']
# Champ : Email
email = ""
if ("email" in Survey_Data.keys()):
email = Survey_Data['email']
param_survey_data['email'] = Survey_Data['email']
# Champ : nom
nom = ""
if ("nom" in Survey_Data.keys()):
nom = Survey_Data['nom']
param_survey_data['nom'] = Survey_Data['nom']
# Champ : prenom
prenom = ""
if ("prenom" in Survey_Data.keys()):
prenom = Survey_Data['prenom']
param_survey_data['prenom'] = Survey_Data['prenom']
# Champ : date_envoi
date_envoi = ""
if ("date_envoi" in Survey_Data.keys()):
date_envoi = str(Survey_Data['date_envoi'])[0:16]
param_survey_data['date_envoi'] = str(date_envoi)
# Type de questionnaire
param_survey_data['survey_type'] = str(my_survey_type)
print(" param_survey_data = ", param_survey_data)
# Champ : date_reponse
date_reponse = ""
if ("date_reponse" in Survey_Data.keys()):
date_reponse = str(Survey_Data['date_reponse'])[0:16]
param_survey_data['date_reponse'] = str(date_reponse)
# Gestion des question et reponse
param_survey_data['question_reponse'] = []
if( "user_response" in Survey_Data.keys() ):
question_num = 1
for user_response in Survey_Data['user_response']:
node_question_reponse = {}
question = ""
if ("question" in user_response.keys()):
question = str(user_response['question'])
node_question_reponse['question'] = question
response = ""
if ("response" in user_response.keys()):
response = str(user_response['response'])
node_question_reponse['response'] = response
question_num += 1
# recup type de question pr pdf
question_type = ""
if ("type" in user_response.keys()):
question_type = str(user_response['type'])
node_question_reponse['question_type'] = question_type
param_survey_data['question_reponse'].append(node_question_reponse)
else:
node_question_reponse = {}
node_question_reponse['response'] = "---"
response = "---"
node_question_reponse['response'] = response
# recup type de question pr pdf
question_type = "---"
node_question_reponse['question_type'] = question_type
param_survey_data['question_reponse'].append(node_question_reponse)
convention_dictionnary_data['param_survey_data'] = param_survey_data
body = {
"params": convention_dictionnary_data,
}
#print(" ### param_survey_data - body = ", body)
"""
Creation du fichier PDF
"""
contenu_doc_Template = jinja2.Template(str(courrier_data_retval['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(",", "")[-3:]
tmp_file_name = str(param_survey_data['code_session'])+"_"+str(param_survey_data['nom'])
if(len(str(tmp_file_name)) > 30 ):
tmp_file_name = str(tmp_file_name)[0:30]
orig_file_name = "Evaluation_" + str(tmp_file_name)+ "_" + 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()
list_file_name_to_zip.append(outputFilename)
# Create a ZipFile Object
todays_date = str(date.today().strftime("%d/%m/%Y"))
ts = datetime.now().timestamp()
ts = str(ts).replace(".", "").replace(",", "")[-5:]
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_Evaluation_"+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):
return True, send_file(zip_file_name, as_attachment=True)
return False, "Impossible de générer l'export (2) "
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 l'export "
"""
Cette fonction permet de faire des statistiques ou de projections graphiques sur
sur des reponses, par exemple la moyenne de satisfaction par rapport
a un formateur ou a une formation ( en effet dans la collection "survey",
on le champ "class_id"
"""
def Get_Survey_Stat_Data_For_Given_related_collection_id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'periode_start_date', 'periode_end_date', 'related_collection', 'related_collection_id',
'question_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"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'related_collection', 'related_collection_id', 'question_id',
'periode_start_date', 'periode_end_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"
"""
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
"""
Recuperer les données de la quesion
"""
is_question_formulaire_exist = MYSY_GV.dbname['formulaire'].count_documents(
{'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0',
'list_questions._id': str(diction['question_id'])})
print(" qry = ",{'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0',
'list_questions._id': str(diction['question_id'])})
if (is_question_formulaire_exist != 1):
mycommon.myprint(" L'identifiant de la question est invalide ")
return False, " L'identifiant de la question est invalide "
is_question_formulaire = MYSY_GV.dbname['formulaire'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0',
'list_questions._id': str(diction['question_id'])})
is_question_data = None
for tmp in is_question_formulaire['list_questions']:
if( tmp['_id'] == str(diction['question_id'])):
is_question_data = tmp
if( is_question_data is None ):
mycommon.myprint(" L'identifiant de la question est invalide (2) ")
return False, " L'identifiant de la question est invalide (2) "
if( "is_statistic" not in is_question_data.keys() or str(is_question_data['is_statistic']) != "1"):
mycommon.myprint(" La question n'est pas soumises à une analyse statistique ")
return False, " La question n'est pas soumises à une analyse statistique "
question_analyse_type = ""
if( is_question_data['type'] in ['note', 'entier']):
question_analyse_type = "moyenne"
elif (is_question_data['type'] in ['bool']):
question_analyse_type = "somme"
else :
mycommon.myprint(" Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données ")
return False, " Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données "
"""
Si on a des class_id alors on cree les filtre
"""
filt_class_id = {}
if( "class_id" in diction.keys() and diction['class_id'] ):
filt_class_id = { 'class_id':str(diction['class_id'])}
"""
Si on a des related_collection, 'related_collection_id' on cree les filtre
"""
filt_related_collection_and_related_collection_id = {}
if ("related_collection" in diction.keys() and diction[
'related_collection'] and "related_collection_id" in diction.keys() and diction['related_collection_id']):
filt_related_collection_and_related_collection_id = {
'related_collection': str(diction['related_collection']),
'related_collection_id': str(diction['related_collection_id'])}
filt_periode_start_date = ""
if ("periode_start_date" in diction.keys() and diction['periode_start_date']):
filt_periode_start_date = str(diction['periode_start_date'])[0:10]
local_status = mycommon.CheckisDate(filt_periode_start_date)
if (local_status is False):
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa.")
return False, " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa."
"""
Si on a des dates, 'date_from', 'date_to', verifier qu'elles sont valides
"""
filt_periode_start_date = ""
if ("periode_start_date" in diction.keys() and diction['periode_start_date']):
filt_periode_start_date = str(diction['periode_start_date'])[0:10]
local_status = mycommon.CheckisDate(filt_periode_start_date)
if (local_status is False):
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa.")
return False, " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa."
filt_periode_end_date = ""
if ("periode_end_date" in diction.keys() and diction['periode_end_date']):
filt_periode_end_date = str(diction['periode_end_date'])[0:10]
local_status = mycommon.CheckisDate(filt_periode_end_date)
if (local_status is False):
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le filtre : 'date de fin' n'est pas au format jj/mm/aaaa.")
return False, " Le filtre : 'date de fin' n'est pas au format jj/mm/aaaa."
if( filt_periode_start_date and filt_periode_end_date ):
filt_periode_start_date_ISODATE = datetime.strptime(str(filt_periode_start_date), '%d/%m/%Y')
filt_periode_end_date_ISODATE = datetime.strptime(str(filt_periode_end_date), '%d/%m/%Y')
else:
mycommon.myprint(str(
inspect.stack()[0][3]) + " Les filtred : les dates de debut et fin ne sont pas valides.")
return False, " Les filtred : les dates de debut et fin ne sont pas valides."
"""
Creation de la range des mois entre filt_periode_start_date_ISODATE et
filt_periode_end_date_ISODATE
"""
range_date_month = []
start = filt_periode_start_date_ISODATE
end = filt_periode_end_date_ISODATE
while start <= end:
node = {}
node['month_year'] = '{:02d}'.format(start.month) + "_" + str(start.year)
node['label'] = '{:02d}'.format(start.month) + "_" + str(start.year)
node['TotalAmount'] = 0
node['value'] = 0
node['count'] = 0
range_date_month.append(node)
start += relativedelta(months=1)
#print(" ### range_date_month = ", range_date_month)
qery_match = {'$and': [{"partner_owner_recid": str(my_partner['recid'])},
{"valide": '1', 'locked':'0', 'statut':'2',
"date_reponse": {"$exists": True},
'user_response.question_id': str(diction['question_id']),
},
filt_related_collection_and_related_collection_id,
filt_class_id,
{
'mysy_survey_date_reponse': {'$gte': filt_periode_start_date_ISODATE,
'$lte': filt_periode_end_date_ISODATE}}, ]}
pipe_qry = ([
{"$addFields": {
"mysy_survey_date_reponse": {
'$dateFromString': {
'dateString': { '$substr': [ "$date_reponse", 0, 10 ] },
'format': "%Y-%m-%d"
}
}
}
},
{'$match': qery_match},
])
print(" ### Get_Survey_Stat_Data ici pipe_qry = ", pipe_qry)
RetObject = []
val_tmp = 0
raw_reponse_data = []
for retval in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
for tmp in retval['user_response']:
if (tmp['question_id'] == str(diction['question_id'])):
node = {}
node['related_collection'] = str(diction['related_collection'])
node['related_collection_id'] = str(diction['related_collection_id'])
node['question_id'] = str(diction['question_id'])
node['question_question'] = str(is_question_data['question'])
node['question_type'] = str(is_question_data['type'])
node['question_response'] = str(tmp['response'])
node['question_date_reponse'] = str(retval['date_reponse'])
node['question_date_reponse_mois_annee'] = str(retval['date_reponse'])[5:7]+"_"+str(retval['date_reponse'])[0:4]
raw_reponse_data.append(node)
print(" ### les reponses brutes sont : ", raw_reponse_data)
global_nb_question_response = 0
global_somme_question_response = 0
global_moyenne_question_response = 0
for tmp in range_date_month:
nb_question_response = 0
somme_question_response = 0
for local_data in raw_reponse_data:
if (str(local_data['question_date_reponse_mois_annee']) == str(tmp['month_year'])):
nb_question_response = nb_question_response + 1
if( question_analyse_type == "somme" and str(local_data['question_response']).lower().strip() == "oui"):
somme_question_response = somme_question_response + 1
elif(question_analyse_type == "moyenne" ):
somme_question_response = somme_question_response + mycommon.tryFloat(str(local_data['question_response']).lower().strip())
if( nb_question_response > 0 ):
"""print(" ## calcul moyenne : ")
print(" ## calcul somme_question_response : ", somme_question_response)
print(" ## calcul nb_question_response : ", nb_question_response)
"""
moyenne_question_response = round(somme_question_response / nb_question_response, 2)
tmp['nb_question_response'] = mycommon.tryFloat(str(nb_question_response))
tmp['somme_question_response'] = mycommon.tryFloat(str(somme_question_response))
tmp['moyenne_question_response'] = mycommon.tryFloat(str(moyenne_question_response))
tmp['count'] = mycommon.tryFloat(str(nb_question_response))
if( str(question_analyse_type) == "moyenne"):
tmp['value'] = mycommon.tryFloat(str(moyenne_question_response))
tmp['type_question'] = "moyenne"
elif( str(question_analyse_type) == "somme"):
tmp['value'] = mycommon.tryFloat(str(somme_question_response))
tmp['type_question'] = "somme"
else:
tmp['nb_question_response'] = 0
tmp['somme_question_response'] = 0
tmp['moyenne_question_response'] = 0
tmp['count'] = 0
tmp['value'] = 0
global_nb_question_response = global_nb_question_response + mycommon.tryFloat(str( tmp['nb_question_response']))
global_somme_question_response = global_somme_question_response + mycommon.tryFloat(str( tmp['somme_question_response']))
if( global_nb_question_response > 0 ):
global_moyenne_question_response = round(global_somme_question_response / global_nb_question_response, 2)
node = {}
node['global_nb_question_response'] = global_nb_question_response
node['global_somme_question_response'] = global_somme_question_response
node['global_moyenne_question_response'] = global_moyenne_question_response
final_data = {}
final_data['data'] = range_date_month
final_data['global'] = node
print(" ### les reponses final_data: ", final_data)
RetObject = []
RetObject.append(mycommon.JSONEncoder().encode(final_data))
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 les données statistiques "
"""
Export Excel des resultat du Get_Survey_Stat_Data_For_Given_related_collection_id
"""
def Export_Excel_Survey_Stat_Data_For_Given_related_collection_id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'periode_start_date', 'periode_end_date', 'related_collection', 'related_collection_id',
'question_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"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'related_collection', 'related_collection_id', 'question_id',
'periode_start_date', 'periode_end_date', 'class_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
"""
Recuperer les données de la quesion
"""
is_question_formulaire_exist = MYSY_GV.dbname['formulaire'].count_documents(
{'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0',
'list_questions._id': str(diction['question_id'])})
if (is_question_formulaire_exist != 1):
mycommon.myprint(" L'identifiant de la question est invalide ")
return False, " L'identifiant de la question est invalide "
is_question_formulaire = MYSY_GV.dbname['formulaire'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'locked': '0',
'list_questions._id': str(diction['question_id'])})
is_question_data = None
for tmp in is_question_formulaire['list_questions']:
if( tmp['_id'] == str(diction['question_id'])):
is_question_data = tmp
if( is_question_data is None ):
mycommon.myprint(" L'identifiant de la question est invalide (2) ")
return False, " L'identifiant de la question est invalide (2) "
if( "is_statistic" not in is_question_data.keys() or str(is_question_data['is_statistic']) != "1"):
mycommon.myprint(" La question n'est pas soumises à une analyse statistique ")
return False, " La question n'est pas soumises à une analyse statistique "
question_analyse_type = ""
if( is_question_data['type'] in ['note', 'entier']):
question_analyse_type = "moyenne"
elif (is_question_data['type'] in ['bool']):
question_analyse_type = "somme"
else :
mycommon.myprint(" Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données ")
return False, " Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données "
"""
Si on a des class_id alors on cree les filtre
"""
filt_class_id = {}
if ("class_id" in diction.keys() and diction['class_id']):
filt_class_id = {'class_id': str(diction['class_id'])}
"""
Si on a des related_collection, 'related_collection_id' on cree les filtre
"""
filt_related_collection_and_related_collection_id = {}
if ("related_collection" in diction.keys() and diction[
'related_collection'] and "related_collection_id" in diction.keys() and diction['related_collection_id']):
filt_related_collection_and_related_collection_id = {
'related_collection': str(diction['related_collection']),
'related_collection_id': str(diction['related_collection_id'])}
"""
Si on a des dates, 'date_from', 'date_to', verifier qu'elles sont valides
"""
filt_periode_start_date = ""
if ("periode_start_date" in diction.keys() and diction['periode_start_date']):
filt_periode_start_date = str(diction['periode_start_date'])[0:10]
local_status = mycommon.CheckisDate(filt_periode_start_date)
if (local_status is False):
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa.")
return False, " Le filtre : 'date de debut' n'est pas au format jj/mm/aaaa."
filt_periode_end_date = ""
if ("periode_end_date" in diction.keys() and diction['periode_end_date']):
filt_periode_end_date = str(diction['periode_end_date'])[0:10]
local_status = mycommon.CheckisDate(filt_periode_end_date)
if (local_status is False):
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le filtre : 'date de fin' n'est pas au format jj/mm/aaaa.")
return False, " Le filtre : 'date de fin' n'est pas au format jj/mm/aaaa."
if( filt_periode_start_date and filt_periode_end_date ):
filt_periode_start_date_ISODATE = datetime.strptime(str(filt_periode_start_date), '%d/%m/%Y')
filt_periode_end_date_ISODATE = datetime.strptime(str(filt_periode_end_date), '%d/%m/%Y')
else:
mycommon.myprint(str(
inspect.stack()[0][3]) + " Les filtred : les dates de debut et fin ne sont pas valides.")
return False, " Les filtred : les dates de debut et fin ne sont pas valides."
"""
Creation de la range des mois entre filt_periode_start_date_ISODATE et
filt_periode_end_date_ISODATE
"""
range_date_month = []
start = filt_periode_start_date_ISODATE
end = filt_periode_end_date_ISODATE
while start <= end:
node = {}
node['month_year'] = '{:02d}'.format(start.month) + "_" + str(start.year)
node['label'] = '{:02d}'.format(start.month) + "_" + str(start.year)
node['TotalAmount'] = 0
node['value'] = 0
node['count'] = 0
range_date_month.append(node)
start += relativedelta(months=1)
#print(" ### range_date_month = ", range_date_month)
qery_match = {'$and': [{"partner_owner_recid": str(my_partner['recid'])},
{"valide": '1', 'locked':'0', 'statut':'2',
"date_reponse": {"$exists": True},
'user_response.question_id': str(diction['question_id']),
},
filt_class_id,
filt_related_collection_and_related_collection_id,
{
'mysy_survey_date_reponse': {'$gte': filt_periode_start_date_ISODATE,
'$lte': filt_periode_end_date_ISODATE}}, ]}
pipe_qry = ([
{"$addFields": {
"mysy_survey_date_reponse": {
'$dateFromString': {
'dateString': { '$substr': [ "$date_reponse", 0, 10 ] },
'format': "%Y-%m-%d"
}
}
}
},
{'$match': qery_match},
])
#print(" ### Get_Survey_Stat_Data ici pipe_qry = ", pipe_qry)
RetObject = []
val_tmp = 0
raw_reponse_data = []
for retval in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
for tmp in retval['user_response']:
if (tmp['question_id'] == str(diction['question_id'])):
node = {}
node['related_collection'] = str(diction['related_collection'])
node['related_collection_id'] = str(diction['related_collection_id'])
node['question_id'] = str(diction['question_id'])
node['question_question'] = str(is_question_data['question'])
node['question_type'] = str(is_question_data['type'])
node['question_response'] = str(tmp['response'])
node['question_date_reponse'] = str(retval['date_reponse'])
node['question_date_reponse_mois_annee'] = str(retval['date_reponse'])[5:7]+"_"+str(retval['date_reponse'])[0:4]
raw_reponse_data.append(node)
print(" ### les reponses brutes sont : ", raw_reponse_data)
global_nb_question_response = 0
global_somme_question_response = 0
global_moyenne_question_response = 0
for tmp in range_date_month:
nb_question_response = 0
somme_question_response = 0
for local_data in raw_reponse_data:
if (str(local_data['question_date_reponse_mois_annee']) == str(tmp['month_year'])):
nb_question_response = nb_question_response + 1
if( question_analyse_type == "somme" and str(local_data['question_response']).lower().strip() == "oui"):
somme_question_response = somme_question_response + 1
elif(question_analyse_type == "moyenne" ):
somme_question_response = somme_question_response + mycommon.tryFloat(str(local_data['question_response']).lower().strip())
if( nb_question_response > 0 ):
"""print(" ## calcul moyenne : ")
print(" ## calcul somme_question_response : ", somme_question_response)
print(" ## calcul nb_question_response : ", nb_question_response)
"""
moyenne_question_response = round(somme_question_response / nb_question_response, 2)
tmp['nb_question_response'] = mycommon.tryFloat(str(nb_question_response))
tmp['somme_question_response'] = mycommon.tryFloat(str(somme_question_response))
tmp['moyenne_question_response'] = mycommon.tryFloat(str(moyenne_question_response))
tmp['count'] = mycommon.tryFloat(str(nb_question_response))
if( str(question_analyse_type) == "moyenne"):
tmp['value'] = mycommon.tryFloat(str(moyenne_question_response))
tmp['type_question'] = "moyenne"
elif( str(question_analyse_type) == "somme"):
tmp['value'] = mycommon.tryFloat(str(somme_question_response))
tmp['type_question'] = "somme"
else:
tmp['nb_question_response'] = 0
tmp['somme_question_response'] = 0
tmp['moyenne_question_response'] = 0
tmp['count'] = 0
tmp['value'] = 0
global_nb_question_response = global_nb_question_response + mycommon.tryFloat(str( tmp['nb_question_response']))
global_somme_question_response = global_somme_question_response + mycommon.tryFloat(str( tmp['somme_question_response']))
if( global_nb_question_response > 0 ):
global_moyenne_question_response = round(global_somme_question_response / global_nb_question_response, 2)
node = {}
node['global_nb_question_response'] = global_nb_question_response
node['global_somme_question_response'] = global_somme_question_response
node['global_moyenne_question_response'] = global_moyenne_question_response
final_data = {}
final_data['data'] = range_date_month
final_data['global'] = node
print(" ### les reponses final_data: ", final_data)
"""
Preparation des exports Excels à faire
"""
todays_date = str(date.today().strftime("%d/%m/%Y"))
ts = datetime.now().timestamp()
ts = str(ts).replace(".", "").replace(",", "")[-5:]
orig_file_name = "Export_Excel_Report_" + str(my_partner['recid']) + "_" + str(ts) + ".xlsx"
outputFilename = str(MYSY_GV.TEMPORARY_DIRECTORY) + "/" + str(orig_file_name)
tab_exported_fields = [ 'date_export', 'filtre_date_debut', 'filtre_date_fin', 'question', 'type_question', 'periode', 'nb_question_response',
'somme_question_response', 'moyenne_question_response']
# Create a workbook and add a worksheet.
workbook = xlsxwriter.Workbook(outputFilename)
worksheet = workbook.add_worksheet()
row = 0
column = 0
# Ecrire l'entete
for header_item in tab_exported_fields:
worksheet.write(row, column, header_item)
column += 1
# Ecrire les lignes du fichier excel
for val_tmp in range_date_month:
column = 0
row = row + 1
worksheet.write(row, column, str(todays_date))
column += 1
worksheet.write(row, column, str(diction['periode_start_date']))
column += 1
worksheet.write(row, column, str(diction['periode_end_date']))
column += 1
worksheet.write(row, column, str(is_question_data['question']))
column += 1
worksheet.write(row, column, str(is_question_data['type']))
column += 1
if( "month_year" in val_tmp.keys() ):
worksheet.write(row, column, str(val_tmp['month_year']))
column += 1
else:
worksheet.write(row, column, "")
column += 1
if ("nb_question_response" in val_tmp.keys()):
worksheet.write(row, column, str(val_tmp['nb_question_response']))
column += 1
else:
worksheet.write(row, column, "")
column += 1
if ("somme_question_response" in val_tmp.keys()):
worksheet.write(row, column, str(val_tmp['somme_question_response']))
column += 1
else:
worksheet.write(row, column, "")
column += 1
if ("moyenne_question_response" in val_tmp.keys()):
worksheet.write(row, column, str(val_tmp['moyenne_question_response']))
column += 1
else:
worksheet.write(row, column, "")
column += 1
workbook.close()
if os.path.exists(outputFilename):
# print(" ### ok os.path.exists(outputFilename) "+str(outputFilename))
return True, send_file(outputFilename, as_attachment=True)
return False, " Impossible de générer l'export csv (2) "
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 l'export csv "
"""
Recuperer la liste des questions d'un 'related_collection' et 'related_collection_id'
avec statistique
"""
def Get_Entity_Question_List_With_Stat_For_Given_related_collection_id(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', '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"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'related_collection', 'related_collection_id', 'class_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
"""
Si on a des class_id alors on cree les filtre
/!\ : quand utilise le filtre "class_id", on verrouille "related_collection" sur la valeur : "inscription"
"""
filt_class_id = {}
if ("class_id" in diction.keys() and diction['class_id']):
filt_class_id = {'class_id': str(diction['class_id']),
'related_collection':'inscription'}
"""
Si on a des related_collection, 'related_collection_id' on cree les filtre
"""
filt_related_collection_and_related_collection_id = {}
if ("related_collection" in diction.keys() and diction[
'related_collection'] and "related_collection_id" in diction.keys() and diction['related_collection_id']):
filt_related_collection_and_related_collection_id = {
'related_collection': str(diction['related_collection']),
'related_collection_id': str(diction['related_collection_id'])}
#print(" ### range_date_month = ", range_date_month)
qery_match = {'$and': [{"partner_owner_recid": str(my_partner['recid'])},
{"valide": '1', 'locked':'0', 'statut':'2',
"date_reponse": {"$exists": True},
'user_response.is_statistic': "1",
},
filt_class_id,
filt_related_collection_and_related_collection_id,
]}
pipe_qry = ([
{'$match': qery_match},
])
print(" ### Get_Survey_Stat_Data ici pipe_qry = ", pipe_qry)
RetObject = []
val_tmp = 0
RetObject = []
raw_question_id = []
cpt = 0
for retval in MYSY_GV.dbname['survey'].aggregate(pipe_qry):
for tmp in retval['user_response']:
if (tmp['is_statistic'] == "1" and str(tmp['question_id']) not in raw_question_id):
raw_question_id.append(str(tmp['question_id']))
node = {}
node['id'] = str(cpt)
cpt = cpt + 1
node['question_question'] = str(tmp['question'])
node['related_collection'] = str(diction['related_collection'])
node['related_collection_id'] = str(diction['related_collection_id'])
node['question_id'] = str(tmp['question_id'])
node['question_question'] = str(tmp['question'])
node['class_id'] = str(retval['class_id'])
if (str(tmp['type']) in ['note', 'entier']):
node['question_type'] = "Moyenne"
elif (str(tmp['type'])in ['bool']):
node['question_type'] = "Somme"
else:
mycommon.myprint(
" Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données ")
return False, " Le paramétrage de la question ne permet pas de déterminer la nature statistiques des données "
RetObject.append(mycommon.JSONEncoder().encode(node))
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 questions "