Elyos_FI_Back_Office/survey_mgt.py

1734 lines
71 KiB
Python

"""
Ce document permet de gerer les "enquetes" avec utilisation des formulaires
"""
import ast
import bson
import pymongo
import xlsxwriter
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime
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 liste ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
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
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 liste ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
"""
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}
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 liste ")
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,
"""
def Init_Survey_Tab_For_All_Session_Inscrit(diction):
try:
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 liste ")
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
req = {'session_id': str(diction['session_id']),
'_id':ObjectId(str(my_inscription_id)),
'partner_owner_recid': str(my_partner['recid']),
'valide': '1', 'status':'1'
}
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']
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 envoi un formulaire pour inscrit
"""
def Create_One_Survey_To_Inscrit(diction):
try:
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 liste ")
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 "
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['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 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.
"""
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 liste ")
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'}
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 "
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"
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'])
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 local_survey_retval.keys() and local_survey_retval['survey_type'] == "cold_eval"):
courrier_ref_interne = "EVAL_FORMATION"
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 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 liste ")
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"
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"
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']
"""
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' ]
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 liste ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
filt_session_id = {}
if ("session_id" in diction.keys()):
filt_session_id = {'session_id': str(diction['session_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]}
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)
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' ]
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 liste ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
filt_session_id = {}
if ("session_id" in diction.keys()):
filt_session_id = {'session_id': str(diction['session_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]}
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 "