Elyos_FI_Back_Office/survey_mgt.py

530 lines
21 KiB
Python

"""
Ce document permet de gerer les "enquetes" avec utilisation des formulaires
"""
import ast
import bson
import pymongo
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime
import prj_common as mycommon
import secrets
import inspect
import sys, os
import csv
import pandas as pd
from pymongo import ReturnDocument
import GlobalVariable as MYSY_GV
from math import isnan
import GlobalVariable as MYSY_GV
import ela_index_bdd_classes as eibdd
import email_mgt as email
import jinja2
from flask import send_file
from xhtml2pdf import pisa
from email.message import EmailMessage
from email.mime.text import MIMEText
from email import encoders
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
"""
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 = 0
qry = { 'statut':'0', 'valide': '1', 'locked': '0', '_id':ObjectId(str(diction['survey_id']))}
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 = ""
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']
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['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 = []
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']
user['list_questions'] = list_questions
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 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': '0', '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 "
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]
node = {"question_id":question_id, "response":response}
global_response.append(node)
"""
On va enregister les reponse et changer le statut
"""
mydata = {}
mydata['date_update'] = str(datetime.now())
mydata['update_by'] = ""
mydata['statut'] = "1"
mydata['user_response'] = global_response
result = MYSY_GV.dbname['survey'].find_one_and_update(
{'statut': '0', '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'inscri
"""
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
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'})
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
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'
})
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']
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 Send_One_Survey_To_Inscrit(diction):
try:
field_list_obligatoire = ['session_id', 'token', 'inscrit_id', '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
"""
Recuperer les données du formulaire
"""
my_formulaire_data = MYSY_GV.dbname['formulaire'].find_one({'_id':ObjectId(str(diction['formulaire_id'])),
'valide':'1',
'locked':'1',
'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',
'locked': '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 "
"""
Recuperer les données de l'inscrit
"""
my_inscrit_data = MYSY_GV.dbname['session_formation'].find_one({'_id': ObjectId(str(diction['inscrit_id'])),
'valide': '1',
'locked': '1',
'partner_owner_recid': str(
my_partner['recid'])})
if (my_inscrit_data is None):
mycommon.myprint(
str(inspect.stack()[0][
3]) + " L'identifiant de l'inscrit est invalide ")
return False, " L'identifiant de l'inscrit est invalide "
new_survey_data = {}
new_survey_data['email'] = my_inscrit_data['email']
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['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) "
servey_url = MYSY_GV.CLIENT_URL_BASE+"Survey/"+str(inserted_id)
"""
A présent l'enquete est créée, il faut donc envoyer
le mail avec l'url à la personne qui doit remplir l'enquete
"""
"""
Recuperation des parametre SMTP du partner si le client a decidé d'utiliser son propre smpt
"""
partner_own_smtp_value = "0"
partner_own_smtp = MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'partner_smtp',
'valide': '1',
'locked': '0'})
if (partner_own_smtp and "config_value" in partner_own_smtp.keys()):
partner_own_smtp_value = partner_own_smtp['config_value']
if (str(partner_own_smtp_value) == "1"):
partner_SMTP_COUNT_password = str(MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'smtp_user_pwd',
'valide': '1',
'locked': '0'}, {'config_value': 1})['config_value'])
partner_SMTP_COUNT_smtpsrv = str(MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'smtp_server',
'valide': '1',
'locked': '0'}, {'config_value': 1})['config_value'])
partner_SMTP_COUNT_user = str(MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'smtp_user',
'valide': '1',
'locked': '0'}, {'config_value': 1})['config_value'])
partner_SMTP_COUNT_From_User = str(MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'smtp_count_from_name',
'valide': '1',
'locked': '0'}, {'config_value': 1})['config_value'])
partner_SMTP_COUNT_port = str(MYSY_GV.dbname['base_partner_setup'].find_one(
{'partner_owner_recid': str(my_partner['recid']),
'config_name': 'smtp_count_port',
'valide': '1',
'locked': '0'}, {'config_value': 1})['config_value'])
if (str(partner_own_smtp_value) == "1"):
smtpserver = smtplib.SMTP(partner_SMTP_COUNT_smtpsrv, partner_SMTP_COUNT_port)
else:
smtpserver = smtplib.SMTP(MYSY_GV.O365_SMTP_COUNT_smtpsrv, MYSY_GV.O365_SMTP_COUNT_port)
html = """\
<html>
<body>
<p>Hi,<br>
Merci de suivre l'enquete :"""+str(servey_url)+""" </p>
<p><a href="https://blog.mailtrap.io/2018/09/27/cloud-or-local-smtp-server">SMTP Server for Testing: Cloud-based or Local?</a></p>
<p> Feel free to <strong>let us</strong> know what content would be useful for you!</p>
</body>
</html>
"""
html_mime = MIMEText(html, 'html')
# Creation de l'email à enoyer
msg = MIMEMultipart("alternative")
if (str(partner_own_smtp_value) == "1"):
msg.attach(html_mime)
msg['From'] = partner_SMTP_COUNT_From_User
msg['Bcc'] = 'contact@mysy-training.com'
msg['Subject'] = " Demande d'enquete "
# msg['to'] = "billardman01@hotmail.com"
msg['to'] = str(my_inscrit_data['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'] = " Demande d'enquete "
# msg['to'] = "billardman01@hotmail.com"
msg['to'] = str(my_inscrit_data['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))
return True, "OK"
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False, " Impossible d'envoyer l'enquete unitaire "
"""
Survey for customer : Cette fonction créer une enquete en prenant à partir d'une liste d'inscri
"""