Elyos_FI_Back_Office/emargement.py

573 lines
23 KiB
Python

"""
Ce ficher gerer l'emargement à une session de formation
"""
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
from datetime import timedelta
from xhtml2pdf import pisa
import jinja2
import ftplib
import pysftp
from flask import send_file
"""
Creation du tableau d'emargement
Au debut de la session, le formateur clique sur un bouton qui va
aller créer le tableau d'emargement.
la finction prend en argement :
- url_formation
- session_id
"""
def CreateTableauEmargement(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = ['class_internal_url', 'session_id', 'token']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - Creation partner account : Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée")
return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables"
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = ['class_internal_url', 'session_id', '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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
mydata = {}
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
# Verifier la validité du token
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
return "Err_Connexion", " La session de connexion n'est pas valide"
class_internal_url = ""
if ("class_internal_url" in diction.keys()):
if diction['class_internal_url']:
class_internal_url = diction['class_internal_url']
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if (partner_recid is False):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de créer la liste d'emargement ")
return False, "Impossible de créer la liste d'emargement "
session_id = ""
if ("session_id" in diction.keys()):
if diction['session_id']:
mydata['session_id'] = diction['session_id']
session_id = diction['session_id']
"""
Recuperation des info de la session de formation
"""
date_du = ""
date_au = ""
ville = ""
code_postal = ""
adresse = ""
qry = {"_id":ObjectId(str(session_id)), "valide": "1"}
session_formation_count = MYSY_GV.dbname['session_formation'].count_documents({"_id":ObjectId(str(session_id)), "valide": "1"})
if( session_formation_count <= 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - Aucune session de formation valide ")
return False, "Aucune session de formation valide "
for tmp_val in MYSY_GV.dbname['session_formation'].find({"_id":ObjectId(str(session_id)),
"valide": "1"}):
date_du = str(tmp_val['date_debut'])[0:10]
date_au = str(tmp_val['date_fin'])[0:10]
ville = ""
if ("ville" in tmp_val.keys() and tmp_val['ville']):
ville = tmp_val['ville']
code_postal = ""
if ("code_postal" in tmp_val.keys() and tmp_val['code_postal']):
code_postal = tmp_val['code_postal']
adresse = ""
if ("adresse" in tmp_val.keys() and tmp_val['adresse']):
adresse = tmp_val['adresse']
is_date_du = mycommon.CheckisDate(date_du)
is_date_au = mycommon.CheckisDate(date_au)
if( is_date_du is False or is_date_au is False):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - Impossible de créer la liste d'emargement, Les dates de la session sont incorrectes ")
return False, " Impossible de créer la liste d'emargement, Les dates de la session sont incorrectes "
date_emargement = datetime.strptime(str(date_du), '%d/%m/%Y')
i = 0 # compteur de securité
# On vide la collection emargement pour la session = str(session_id)
local_myquery = {"session_id": str(session_id)}
x = MYSY_GV.dbname['emargement'].delete_many(local_myquery)
mycommon.myprint(
str(inspect.stack()[0][
3]) + " - INFO : "+str(x.deleted_count)+" Documents supprimés de la collection : emargement : session = "+str(session_id) )
liste_date = []
#print(" Date du = "+str(date_du)+" CMP Date au ="+str(date_au))
while( date_emargement < datetime.strptime(str(date_au), '%d/%m/%Y') and i < 360):
date_10 = str(date_emargement)
liste_date.append(date_10[0:10])
i = i + 1
date_emargement = date_emargement + timedelta(days=1)
#print(" ### date_emargement = "+str(date_emargement) + " < "+str(datetime.strptime(str(date_au), '%d/%m/%Y') ))
#print("#### Liste Date = "+str(liste_date))
coll_emargement = MYSY_GV.dbname['emargement']
#Recuperation de la liste des participants dont l'inscription est validée
for tmp_val in MYSY_GV.dbname['inscription'].find({'session_id':str(session_id),
'class_internal_url':str(class_internal_url),
'status':'1'} ):
my_local_data = {}
my_local_data['email'] = tmp_val['email']
my_local_data['nom'] = tmp_val['nom']
my_local_data['prenom'] = tmp_val['prenom']
my_local_data['session_id'] = tmp_val['session_id']
my_local_data['class_internal_url'] = tmp_val['class_internal_url']
my_local_data['date_update'] = str(datetime.now())
for local_date in liste_date :
query_key = {}
query_key['email'] = tmp_val['email']
query_key['session_id'] = tmp_val['session_id']
query_key['class_internal_url'] = tmp_val['class_internal_url']
query_key['date'] = str(local_date)
new_my_local_data = {}
new_my_local_data['email'] = tmp_val['email']
new_my_local_data['nom'] = tmp_val['nom']
new_my_local_data['prenom'] = tmp_val['prenom']
new_my_local_data['session_id'] = tmp_val['session_id']
new_my_local_data['class_internal_url'] = tmp_val['class_internal_url']
new_my_local_data['date_update'] = str(datetime.now())
new_my_local_data['date'] = str(local_date)
new_my_local_data['matin'] = False
new_my_local_data['apresmidi'] = False
#print("### emargement insertion de : "+str(new_my_local_data))
local_ret_val = coll_emargement.find_one_and_update(
query_key,
{"$set": new_my_local_data},
upsert=True,
return_document=ReturnDocument.AFTER
)
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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de créer le tableau d'emargement"
"""
Cette fonction récupérer la liste d'emargement
"""
def GetTableauEmargement(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = ['class_internal_url', 'session_id', 'token']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - Creation partner account : Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée")
return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables"
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = ['class_internal_url', 'session_id', '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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
mydata = {}
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
# Verifier la validité du token
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
return "Err_Connexion", " La session de connexion n'est pas valide"
class_internal_url = ""
if ("class_internal_url" in diction.keys()):
if diction['class_internal_url']:
class_internal_url = diction['class_internal_url']
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if (partner_recid is False):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de créer la liste d'emargement ")
return False, "Impossible de créer la liste d'emargement "
session_id = ""
if ("session_id" in diction.keys()):
if diction['session_id']:
mydata['session_id'] = diction['session_id']
session_id = diction['session_id']
val_tmp = 1
RetObject = []
for retval in MYSY_GV.dbname['emargement'].find({'session_id':str(session_id),
'class_internal_url':str(class_internal_url)}).sort([("date",pymongo.ASCENDING),]):
user = retval
user['id'] = str(val_tmp)
RetObject.append(mycommon.JSONEncoder().encode(user))
val_tmp = val_tmp + 1
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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de récupérer le tableau d'emargement"
"""
Fonction de mise à jour de l'emargement d'une personne sur une date
"""
def UpdateUserEmargementDate(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = [ 'session_id', 'token', 'email', 'date', 'matin', 'apresmidi', 'class_internal_url']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - Creation partner account : Le champ '" + val + "' n'est pas autorisé, Creation partenaire annulée")
return False, "Impossible de créer la liste d'emargement. Toutes les informations fournies ne sont pas valables"
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = [ 'session_id', 'token', 'email', '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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
mydata = {}
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
# Verifier la validité du token
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
return "Err_Connexion", " La session de connexion n'est pas valide"
my_email = ""
if ("email" in diction.keys()):
if diction['email']:
my_email = diction['email']
mydata['email'] = diction['email']
my_date = ""
if ("date" in diction.keys()):
if diction['date']:
my_date = diction['date']
mydata['date'] = diction['date']
my_matin = ""
if ("matin" in diction.keys()):
if diction['matin']:
my_matin = diction['matin']
my_apresmidi = ""
if ("apresmidi" in diction.keys()):
if diction['apresmidi']:
my_apresmidi = diction['apresmidi']
class_internal_url = ""
if ("class_internal_url" in diction.keys()):
if diction['class_internal_url']:
class_internal_url = diction['class_internal_url']
mydata['class_internal_url'] = class_internal_url
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if (partner_recid is False):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de valider l'emargement ")
return False, "Impossible de valider l'emargement "
session_id = ""
if ("session_id" in diction.keys()):
if diction['session_id']:
mydata['session_id'] = diction['session_id']
session_id = diction['session_id']
mydata['date_update'] = str(datetime.now())
if( my_matin == "1"):
mydata['matin'] = True
elif (my_matin == "0"):
mydata['matin'] = False
if (my_apresmidi == "1"):
mydata['apresmidi'] = True
elif (my_apresmidi == "0"):
mydata['apresmidi'] = False
query_key = {}
query_key['email'] = mydata['email']
query_key['session_id'] = mydata['session_id']
query_key['date'] = str(my_date)
query_key['class_internal_url'] = str(class_internal_url)
coll_emargement = MYSY_GV.dbname['emargement']
local_ret_val = coll_emargement.find_one_and_update(
query_key,
{"$set": mydata},
upsert=False,
return_document=ReturnDocument.AFTER
)
mycommon.myprint(str(inspect.stack()[0][3]) + " OK, la liste d'emargement de la session "+str( mydata['session_id'])+" a été mise à jour")
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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de valider l'émargement"
"""
Cette fonction genere un fichier pdf pour les liste d'emargement
"""
def GerneratePDFEmargementList(diction):
try:
field_list = ['session_id', 'email', 'date', 'token', 'class_internal_url']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans liste ")
return False, " Impossible d'imprimer la liste d'emargement"
"""
Verification de la liste des champs obligatoires
"""
field_list_obligatoire = ['session_id', 'token', 'class_internal_url']
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, " Impossible d'imprimer la liste d'emargement"
query_get_data = {}
mytoken = ""
if ("token" in diction.keys()):
if diction['token']:
mytoken = diction['token']
partner_recid = mycommon.get_parnter_recid_from_token(mytoken)
if (partner_recid is False):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible d'imprimer la liste d'emargement ")
return False, " Impossible d'imprimer la liste d'emargement"
myclass_internal_url = ""
if ("class_internal_url" in diction.keys()):
if diction['class_internal_url']:
myclass_internal_url = diction['class_internal_url']
query_get_data['class_internal_url'] = diction['class_internal_url']
mysession_id = ""
if ("session_id" in diction.keys()):
if diction['session_id']:
mysession_id = diction['session_id']
query_get_data['session_id'] = diction['session_id']
myemail = ""
if ("email" in diction.keys()):
if diction['email']:
myemail = diction['email']
query_get_data['email'] = diction['email']
mydate = ""
if ("date" in diction.keys()):
if diction['date']:
mydate = diction['date']
query_get_data['date'] = diction['date']
## Recuperation des données
coll_emargement = MYSY_GV.dbname['emargement']
print("### query_get_data = " + str(query_get_data))
tab_users = []
for val_tmp in coll_emargement.find(query_get_data).sort([("date", pymongo.ASCENDING)]):
local_tmp = {}
local_tmp['email'] = val_tmp['email']
local_tmp['nom'] = val_tmp['nom']
local_tmp['prenom'] = val_tmp['prenom']
local_tmp['matin'] = val_tmp['matin']
local_tmp['apresmidi'] = val_tmp['apresmidi']
local_tmp['date'] = val_tmp['date']
tab_users.append(local_tmp)
#print(' ### tab_users = ' + str(tab_users))
## Recuperation des info de la formation
class_title = ""
for val_tmp in MYSY_GV.dbname['myclass'].find({'internal_url': str(myclass_internal_url)}):
class_title = val_tmp['title']
## Recuperation des info de la sessop,
code_session = ""
for val_tmp_2 in MYSY_GV.dbname['session_formation'].find({'_id': ObjectId(str(mysession_id))}):
code_session = val_tmp_2['code_session']
mytoday = datetime.today().strftime("%d/%m/%Y")
#print(' ### class_title = ' + str(class_title))
templateLoader = jinja2.FileSystemLoader(searchpath="./")
templateEnv = jinja2.Environment(loader=templateLoader)
TEMPLATE_FILE = "Template/tpl_emargement.html"
template = templateEnv.get_template(TEMPLATE_FILE)
# This data can come from database query
body = {
"data": {
"title": str(class_title),
"date": str(mytoday),
"code_session":str(code_session),
},
"user": tab_users
}
#print(" ### body = " + str(body))
sourceHtml = template.render(json_data=body["data"], users=body["user"])
orig_file_name = "test_emargement.pdf"
outputFilename = str(MYSY_GV.EMARGEMENT_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()
#print(" ### outputFilename = "+str(outputFilename))
if os.path.exists(outputFilename):
#print(" ### ok os.path.exists(outputFilename) "+str(outputFilename))
return True, send_file(outputFilename, as_attachment=True)
# myprint(str(inspect.stack()[0][3]) +" debut envoie de la factureeee "+diction['invoice_id'])
# email.SendInvoiceEmail(str(diction['invoice_email']), diction)
# On deplace la facture vers le serveur ftp
'''myprint(
str(inspect.stack()[0][3]) + " deplacement de la facture vers " + str(
outputFilename))
'''
"""
cnopts = pysftp.CnOpts()
cnopts.hostkeys = None
with pysftp.Connection(host=MYSY_GV.MYSY_FTP_HOST, username=MYSY_GV.MYSY_FTP_LOGIN,
password=MYSY_GV.MYSY_FTP_PWD, cnopts=cnopts) as session:
print("Connection successfully established ... ")
localFilePath = outputFilename
remoteFilePath = str(MYSY_GV.INVOICE_FTP_LOCAL_STORAGE_DIRECTORY) + str(orig_file_name)
# print(" DEPLACEMENT DE " + str(localFilePath) + " VERS " + str(remoteFilePath) + " AVANTTT TRAITEMENT")
# Use put method to upload a file
session.put(localFilePath, remoteFilePath)
# Switch to a remote directory
myprint(
str(inspect.stack()[0][3]) + " DEPLACEMENT DE " + str(localFilePath) + " VERS " + str(
remoteFilePath) + " EST OKKKK")
"""
# return True on success and False on errors
print(pisaStatus.err, type(pisaStatus.err))
return True, " le fichier generé "
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, False