Elyos_FI_Back_Office/emargement.py

1127 lines
46 KiB
Python

"""
Ce ficher gerer l'emargement à une session de formation
"""
import ast
import pymongo
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime
import partners
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 and val.startswith('my_') is False:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - 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 la liste des arguments ")
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"
"""
23/11/2023 : Evolution en prenant les sequences.
Evolution de la fonction de création du tableau d'emargement.
Dans cette evolution, on plutot recuperer les sequence déjà créesq
31/01/2024 :
le statut de l'émargement :
0 ==> Initial
1 ==> Envoyé
2 ==> Validé
"""
def CreateTableauEmargement_From_Sequence(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 and val.startswith('my_') is False:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - 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 la liste des arguments ")
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']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
partner_recid = my_partner['recid']
class_internal_url = ""
if ("class_internal_url" in diction.keys()):
if diction['class_internal_url']:
class_internal_url = diction['class_internal_url']
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", 'partner_owner_recid':str(partner_recid)}
# Verifier que la session existe
session_formation_count = MYSY_GV.dbname['session_formation'].count_documents(qry)
if( session_formation_count <= 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - Aucune session de formation valide ")
return False, "Aucune session de formation valide "
session_data = MYSY_GV.dbname['session_formation'].find_one({"_id":ObjectId(str(session_id)), "valide": "1"})
# Verifier qu'on a bien des sequences valides pour cette session de formation
is_existe_sequence_session_valide_count = MYSY_GV.dbname['session_formation_sequence'].count_documents({'session_id':str(diction['session_id']),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(partner_recid)})
if( is_existe_sequence_session_valide_count <= 0):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - Aucune sequence valide pour cette session de formation ")
return False, "Aucune sequence valide pour cette session de formation "
# Recuperation de la liste des sequences associées à la session
i = 0 # compteur de securité
# On vide la collection emargement pour la session = str(session_id)
local_myquery = {"session_id": str(session_id), 'partner_owner_recid':str(partner_recid)}
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) )
# Verification qu'il y a bien des séquences de formation pour cette session
session_sequence_count = MYSY_GV.dbname['session_formation_sequence'].count_documents({'session_id': str(session_id),
'locked': '0',
'valide': '1',
'partner_owner_recid': str(
partner_recid)})
if( session_sequence_count <= 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - Cette session de formation n'a aucune séquence valide ")
return False, " Cette session de formation n'a aucune séquence valide "
#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',
'partner_owner_recid':str(partner_recid)} ):
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())
# Recuperation de liste des sequences
for session_sequence in MYSY_GV.dbname['session_formation_sequence'].find({'session_id':str(session_id),
'locked':'0',
'valide':'1',
'partner_owner_recid':str(partner_recid)} ):
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(session_sequence['sequence_start'])[0:10]
query_key['sequence_start'] = str(session_sequence['sequence_start'])
query_key['sequence_end'] = str(session_sequence['sequence_end'])
new_my_local_data = {}
new_my_local_data['inscription_id'] = str(tmp_val['_id'])
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())
query_key['date'] = str(session_sequence['sequence_start'])[0:10]
new_my_local_data['is_present'] = False
new_my_local_data['matin'] = False
new_my_local_data['apresmidi'] = False
new_my_local_data['statut'] = "0"
new_my_local_data['date_envoi'] = ""
new_my_local_data['date_update'] = str(datetime.now())
new_my_local_data['update_by'] = str(my_partner['recid'])
new_my_local_data['valide'] = "1"
new_my_local_data['locked'] = "0"
new_my_local_data['partner_owner_recid'] = str(str(partner_recid))
#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, " Le tableau d'émargement a été correctement initialisé "
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 initialiser de créer le tableau d'émargement "
"""
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 and val.startswith('my_') is False:
mycommon.myprint(str(inspect.stack()[0][
3]) + " - Le champ '" + val + "' n'est pas autorisé")
return False, "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 la liste des arguments ")
return False, "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), ("sequence_start",pymongo.ASCENDING), ("sequence_end",pymongo.ASCENDING),]):
user = retval
if( "statut" not in retval.keys() ):
user['statut'] = ""
if ("date_envoi" not in retval.keys()):
user['date_envoi'] = ""
if ("date_emargement" not in retval.keys()):
user['date_emargement'] = ""
if ("mysy_manual_signature_img" not in retval.keys()):
user['mysy_manual_signature_img'] = ""
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', 'is_present', '_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é, 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', '_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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
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
partner_recid = str(my_partner['recid'])
# 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"
"""
Verifier que l'emargement est valide
"""
is_valide_emargement = MYSY_GV.dbname['emargement'].count_documents({'_id':ObjectId(str(diction['_id'])), 'valide':'1',
'locked':'0', 'partner_owner_recid':str(partner_recid)})
if( is_valide_emargement != 1):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant de l'emargement est invalide ")
return False, " L'identifiant de l'emargement est invalide "
is_valide_emargement_data = MYSY_GV.dbname['emargement'].find_one(
{'_id': ObjectId(str(diction['_id'])), 'valide': '1',
'locked': '0', 'partner_owner_recid': str(partner_recid)})
is_present = ""
if ("is_present" in diction.keys()):
if diction['is_present']:
is_present = diction['is_present']
mydata = {}
now = str(datetime.now().strftime("%d/%m/%Y %H:%M:%S"))
mydata['date_update'] = now
if (is_present == "1"):
mydata['is_present'] = True
elif (is_present == "0"):
mydata['is_present'] = False
mydata['statut'] = "2"
mydata['date_emargement'] = now
mydata['update_by'] = str(my_partner['_id'])
query_key = {}
query_key['_id'] = ObjectId(str(diction['_id']))
query_key['partner_owner_recid'] = str(partner_recid)
#print(" #### query_key = ", query_key)
#print(" #### mydata = ", mydata)
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
)
return True, "La mise à jour a été correctement faite"
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"
"""
04/07/2024 -
Mise à jour d'une liste d'emargement
"""
def Update_Given_Emargement_Data(diction):
try:
diction = mycommon.strip_dictionary(diction)
'''
# 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', 'is_present', 'tab_ids', 'comment'
]
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é, 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 = [ '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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
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
partner_recid = str(my_partner['recid'])
# 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"
tab_ids = ""
if ("tab_ids" in diction.keys()):
if diction['tab_ids']:
tab_ids = diction['tab_ids']
tab_ids_splited = str(tab_ids).split(",")
tab_ids_splited_Objectid = []
for my_id in tab_ids_splited:
# Verifier que les survey sont valides
qry = {'_id': ObjectId(str(my_id)), 'partner_owner_recid': str(my_partner['recid']), 'valide': '1',
'locked': '0'}
print(" qry2 === ", qry)
tmp_count = MYSY_GV.dbname['emargement'].count_documents(qry)
if (tmp_count <= 0):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant de l'emargement " + str(my_id) + " est invalide ")
return False, " L'identifiant de l'emargement " + str(my_id) + " est invalide "
tab_ids_splited_Objectid.append(ObjectId(str(my_id)))
is_present = ""
if ("is_present" in diction.keys()):
if diction['is_present']:
is_present = diction['is_present']
comment = ""
if ("comment" in diction.keys()):
if diction['comment']:
comment = str(diction['comment']).strip()
mydata = {}
mydata['comment'] = str(comment)
mydata['date_update'] = str(datetime.now())
mydata['update_by'] = str(my_partner['_id'])
query_key = {}
query_key['_id'] ={'$in':tab_ids_splited_Objectid}
query_key['partner_owner_recid'] = str(partner_recid)
query_key['valide'] = "1"
query_key['locked'] = "0"
print(" #### query_key = ", query_key)
print(" #### mydata = ", mydata)
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
)
return True, "La mise à jour a été correctement faite"
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 mettre à jour l'émargement"
"""
Recuperer les données d'un emargemend donnée
"""
def Get_Given_Emargement_Data(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 = [ 'token', '_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é, 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 = [ 'token', '_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, "Impossible de créer la liste d'emargement, Toutes les informations necessaires n'ont pas été fournies"
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
partner_recid = str(my_partner['recid'])
# 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"
RetObject = []
val_tmp = 0
for New_retVal in MYSY_GV.dbname['survey'].find_one({'_id':ObjectId(str(diction['_id'])),
'valide':'1', 'locked':'0',
'partner_owner_recid':my_partner['recid']}):
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) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno))
return False, "Impossible de récuperer les données de 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', 'courrier_template_id']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list and val.startswith('my_') is False:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
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 la liste des arguments ")
return False, " Impossible d'imprimer la liste d'emargement"
query_get_data = {}
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
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"
"""
28/03/2024 :
Verifier que le modele de courrier est valide
"""
courrier_template_id = ""
if( "courrier_template_id" in diction.keys() ):
courrier_template_id = diction['courrier_template_id']
if( courrier_template_id == ""):
mycommon.myprint(str(inspect.stack()[0][3]) + " Le modèle de courrier est invalide ")
return False, " Le modèle de courrier est invalide "
courrier_template_id_count = MYSY_GV.dbname['courrier_template'].count_documents({"_id":ObjectId(courrier_template_id),
'partner_owner_recid':my_partner['recid'],
'valide':'1',
'locked':'0'})
if( courrier_template_id_count != 1):
mycommon.myprint(str(inspect.stack()[0][3]) + " Le modèle de courrier est invalide (1) ")
return False, " Le modèle de courrier est invalide (1)"
courrier_template_id_data = MYSY_GV.dbname['courrier_template'].find_one(
{"_id": ObjectId(courrier_template_id),
'partner_owner_recid': my_partner['recid'],
'valide': '1',
'locked': '0'})
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), ("sequence_start", pymongo.ASCENDING), ("sequence_end", pymongo.ASCENDING)]):
local_tmp = {}
local_tmp['email'] = val_tmp['email']
local_tmp['nom'] = val_tmp['nom']
local_tmp['prenom'] = val_tmp['prenom']
local_tmp['date'] = val_tmp['date']
local_tmp['sequence_start'] = val_tmp['sequence_start']
local_tmp['sequence_end'] = val_tmp['sequence_end']
local_tmp['is_present'] = val_tmp['is_present']
mysy_manual_signature_img = ""
if( "mysy_manual_signature_img" in val_tmp.keys() ):
mysy_manual_signature_img = val_tmp['mysy_manual_signature_img']
local_tmp['mysy_manual_signature_img'] = mysy_manual_signature_img
tab_users.append(local_tmp)
#print(' ### tab_users = ' + str(tab_users))
date_for_tmplate = {}
# Recuperation des données du partenaire
date_for_tmplate['partner_name'] = my_partner['nom']
date_for_tmplate['partner_adresse'] = ""
if( "partner_adresse" in my_partner.keys () ):
date_for_tmplate['partner_adresse'] = my_partner['adr_street']
date_for_tmplate['partner_ville'] = ""
if ("adr_city" in my_partner.keys()):
date_for_tmplate['partner_ville'] = my_partner['adr_city']
date_for_tmplate['partner_code_postal'] = ""
if ("invoice_adr_zip" in my_partner.keys()):
date_for_tmplate['partner_code_postal'] = my_partner['invoice_adr_zip']
date_for_tmplate['partner_pays'] = ""
if ("adr_country" in my_partner.keys()):
date_for_tmplate['partner_pays'] = my_partner['adr_country']
date_for_tmplate['partner_phone'] = ""
if ("telephone" in my_partner.keys()):
date_for_tmplate['partner_phone'] = my_partner['telephone']
date_for_tmplate['partner_mail'] = ""
if ("email" in my_partner.keys()):
date_for_tmplate['partner_mail'] = my_partner['email']
if( "website" in my_partner.keys() ):
date_for_tmplate['partner_website'] = my_partner['website']
else:
date_for_tmplate['partner_website'] = ""
mytoday = str(datetime.today().strftime("%d/%m/%Y"))
date_for_tmplate['today_date'] = str(mytoday)
## 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 = ""
if("title" in val_tmp):
class_title = val_tmp['title']
date_for_tmplate['title'] = class_title
## Recuperation des info de la session
code_session = ""
for val_tmp_2 in MYSY_GV.dbname['session_formation'].find({'_id': ObjectId(str(mysession_id))}):
code_session = ""
if ("code_session" in val_tmp_2):
code_session = val_tmp_2['code_session']
date_for_tmplate['code_session'] = code_session
session_date_debut = ""
if ("date_debut" in val_tmp_2):
session_date_debut = val_tmp_2['date_debut']
date_for_tmplate['session_date_debut'] = session_date_debut
session_date_fin = ""
if ("date_fin" in val_tmp_2):
session_date_fin = val_tmp_2['date_fin']
date_for_tmplate['session_date_fin'] = session_date_fin
session_titre = ""
if ("titre" in val_tmp_2):
session_titre = val_tmp_2['titre']
date_for_tmplate['session_titre'] = session_titre
session_distantiel = ""
if ("distantiel" in val_tmp_2):
session_distantiel = val_tmp_2['distantiel']
date_for_tmplate['session_distantiel'] = session_distantiel
session_presentiel = ""
if ("presentiel" in val_tmp_2):
session_presentiel = val_tmp_2['presentiel']
date_for_tmplate['session_presentiel'] = session_presentiel
#print(json.dumps(date_for_tmplate, indent=1))
# Recuperation des données du partenaire associé à l'utilisateur connecté
local_status, local_retval = mycommon.Get_Connected_User_Partner_Data_From_RecID(my_partner['recid'])
if (local_status is False):
return local_status, local_retval
local_company_data = local_retval
# Recuperation des données de la société
company_data = {}
company_liste_champ = ['nom', 'email', 'telephone', 'num_nda', 'website', 'adr_street', 'adr_zip', 'adr_city',
'adr_country']
for champ in company_liste_champ:
if (champ in local_company_data):
new_champ_name = "societe_" + str(champ)
new_field = {new_champ_name: str(local_company_data[champ])}
company_data.update(new_field)
else:
new_champ_name = "societe_" + str(champ)
new_field = {new_champ_name: ""}
company_data.update(new_field)
# Recuperation du logo et du cachet de la société si il y en a
local_part_status, local_part_imgs = partners.getRecodedParnterImage_from_front(
{'token': str(diction['token'])})
if (local_part_status is False):
mycommon.myprint(
str(inspect.stack()[0][3]) + " WARNING : Impossible de récuperer le logo et le cachet du partenaire ")
new_field_logo = {'societe_logo': ''}
new_field_cachet = {'societe_cachet': ''}
company_data.update(new_field_logo)
company_data.update(new_field_cachet)
else:
local_JSON = ast.literal_eval(local_part_imgs[0])
new_field_logo = {'societe_logo': "data:image/png;base64," + local_JSON['logo_img']}
new_field_cachet = {'societe_cachet': "data:image/png;base64," + local_JSON['cachet_img']}
company_data.update(new_field_logo)
company_data.update(new_field_cachet)
#print(" -------- ")
#print(json.dumps(tab_users, indent=1))
# This data can come from database query
body = {
"data": date_for_tmplate,
"user": tab_users,
"company_data" :company_data
}
### Recuperation du modèle de courrier pdf pour l'emargement
"""
data_doc = {}
data_doc['document_ref_intere'] = "EMARGEMENT"
data_doc['document_type'] = "pdf"
data_doc['partner_owner_recid'] = str(partner_recid)
local_retval_status, local_retval_message = mycommon.Get_Personnalized_Document_From_courrier_template(data_doc)
if( local_retval_status is False):
return local_retval_status, local_retval_message
"""
partner_document_data = courrier_template_id_data
contenu_doc_Template = jinja2.Template(str(partner_document_data['contenu_doc']))
sourceHtml = contenu_doc_Template.render(json_data=body["data"], users=body["user"], company_data=body["company_data"])
orig_file_name = "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