Elyos_FI_Back_Office/site_formation.py

538 lines
21 KiB
Python

"""
Ce fichier permet de gérer les differents sites d'un établissement de formation.
1) Ajouter la notion de site de formation
Un site géré dans les paramètres
Un site est un lieu physique où virtuel de formation. A moyen et long terme il sera utilisé pour gérer une école avec plusieurs lieux de formation.
Un site est defini par : "code_site", "nom_site", "site_adr", "site_cp", "site_ville", "site_pays", "telephone", "email", "site_rattachement_id"
la clé est le "code_site"
"""
import ast
import xlsxwriter
import jinja2
import pymongo
from flask import send_file
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime, date
from xhtml2pdf import pisa
import attached_file_mgt
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 Inscription_mgt as Inscription_mgt
import Session_Formation as sf
"""
Ajout d'un site de formation
"""
def Add_Site_Formation(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', "code_site", "nom_site", "site_adr", "site_cp", "site_ville", "site_pays",
"telephone", "email", "site_rattachement_id", "description"
]
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', "code_site", "nom_site",]
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes",
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
# Verifier s'il n'y pas un site avec le meme code
is_code_site_valide = MYSY_GV.dbname['site_formation'].count_documents({'code_site':str(diction['code_site']),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(my_partner['recid'])})
if( is_code_site_valide != 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Un site avec le même code existe déjà ")
return False, " Un site avec le même code existe déjà "
# Si un te site de rattachement est envoyé, verifier sa validité
site_rattachement_id = ""
if( "site_rattachement_id" in diction.keys() and diction['site_rattachement_id']):
site_rattachement_id = diction['site_rattachement_id']
is_site_rattachement_id_valide = MYSY_GV.dbname['site_formation'].count_documents({'_id':ObjectId(str(site_rattachement_id)),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(my_partner['recid'])})
if( is_site_rattachement_id_valide != 1 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant du site de rattachement est invalide ")
return False, " L'identifiant du site de rattachement est invalide "
mydata = diction
del mydata['token']
"""
# Initialisation et remplissage des champs non envoyés à vide
"""
for val in field_list:
if val not in diction.keys():
mydata[str(val)] = ""
mydata['code_site'] = str(diction['code_site']).lower()
mydata['date_update'] = str(datetime.now())
mydata['update_by'] = str(my_partner['_id'])
mydata['partner_owner_recid'] = str(my_partner['recid'])
mydata['valide'] = "1"
mydata['locked'] = "0"
print(" ### site à créer = ", mydata)
inserted_id = MYSY_GV.dbname['site_formation'].insert_one(mydata).inserted_id
if (not inserted_id):
mycommon.myprint(" Impossible de créer le site de formation (2) ")
return False, " Impossible de créer le site de formation (2) "
return True, "Le site a été correctement ajouté "
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 site de formation "
"""
Mettre à jour un site de formation
"""
def Update_Site_Formation(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', "site_id", "code_site", "nom_site", "site_adr", "site_cp", "site_ville", "site_pays", "telephone",
"email", "site_rattachement_id", "description"
]
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', "site_id",]
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes",
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
# Verifier que site existe
is_code_site_valide = MYSY_GV.dbname['site_formation'].count_documents({'_id':ObjectId(str(diction['site_id'])),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(my_partner['recid'])})
if( is_code_site_valide != 1 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant du site est invalide ")
return False, " L'identifiant du site est invalide "
# Si un te site de rattachement est envoyé, verifier sa validité
site_rattachement_id = ""
if( "site_rattachement_id" in diction.keys() and diction['site_rattachement_id']):
site_rattachement_id = diction['site_rattachement_id']
is_site_rattachement_id_valide = MYSY_GV.dbname['site_formation'].count_documents({'_id':ObjectId(str(site_rattachement_id)),
'valide':'1',
'locked':'0',
'partner_owner_recid':str(my_partner['recid'])})
if( is_site_rattachement_id_valide != 1 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant du site de rattachement est invalide ")
return False, " L'identifiant du site de rattachement est invalide "
my_site_id = diction['site_id']
mydata = diction
del mydata['token']
del mydata['site_id']
mydata['date_update'] = str(datetime.now())
mydata['update_by'] = str(my_partner['_id'])
print(" ### mydata = ", mydata)
qry = {'_id': ObjectId(str(my_site_id)),
'partner_owner_recid': str(my_partner['recid'])}
print(" ### qry = ", qry)
result = MYSY_GV.dbname['site_formation'].find_one_and_update(
{'_id': ObjectId(str(my_site_id)),
'partner_owner_recid': str(my_partner['recid'])},
{"$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 site de formation (2) ")
return False, " Impossible de mettre à jour le site de formation (2) "
return True, "Le site 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 de mettre à jour le site de formation "
"""
Recuperation de liste des sites de formation
"""
def Get_List_Site_Formation_with_filter(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', "code_site", "nom_site", ]
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list and val.startswith('my_') is False:
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token',]
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
filt_code_site = {}
if ("code_site" in diction.keys()):
filt_code_site = {'code_site': {'$regex': str(diction['code_site']), "$options": "i"}}
filt_nom_site = {}
if ("nom_site" in diction.keys()):
filt_nom_site = {'nom_site': {'$regex': str(diction['nom_site']), "$options": "i"}}
RetObject = []
val_tmp = 0
qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0'}
query_with_filter = {'$and': [qry, filt_code_site, filt_nom_site ]}
print(" ### Get_List_Site_Formation_with_filter query_with_filter = ", query_with_filter)
for New_retVal in MYSY_GV.dbname['site_formation'].find(query_with_filter):
user = New_retVal
user['id'] = str(val_tmp)
val_tmp = val_tmp + 1
site_rattachement_nom = ""
site_rattachement_code = ""
if( "site_rattachement_id" in New_retVal.keys() and New_retVal['site_rattachement_id']):
site_rattachement_data = MYSY_GV.dbname['site_formation'].find_one(
{'_id': ObjectId(str(New_retVal['site_rattachement_id'])), 'valide': '1', 'locked': '0',
'partner_owner_recid': str(my_partner['recid'])})
if( "code_site" in site_rattachement_data.keys()):
site_rattachement_code = site_rattachement_data['code_site']
if ("nom_site" in site_rattachement_data.keys()):
site_rattachement_nom = site_rattachement_data['nom_site']
else:
user['site_rattachement_id'] = ""
user['site_rattachement_nom'] = site_rattachement_nom
user['site_rattachement_code'] = site_rattachement_code
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 sites de formation "
"""
Recuperer les données d'un site
"""
def Get_Given_Site_Formation(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', "site_id", ]
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list and val.startswith('my_') is False:
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token','site_id']
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
RetObject = []
val_tmp = 0
qry = {"partner_owner_recid":str(my_partner['recid']), 'valide':'1', 'locked':'0','_id':ObjectId(str(diction['site_id']))}
for New_retVal in MYSY_GV.dbname['site_formation'].find(qry):
user = New_retVal
user['id'] = str(val_tmp)
val_tmp = val_tmp + 1
site_rattachement_nom = ""
site_rattachement_code = ""
if( "site_rattachement_id" in New_retVal.keys() and New_retVal['site_rattachement_id']):
site_rattachement_data = MYSY_GV.dbname['site_formation'].find_one(
{'_id': ObjectId(str(New_retVal['site_rattachement_id'])), 'valide': '1', 'locked': '0',
'partner_owner_recid': str(my_partner['recid'])})
if( "code_site" in site_rattachement_data.keys()):
site_rattachement_code = site_rattachement_data['code_site']
if ("nom_site" in site_rattachement_data.keys()):
site_rattachement_nom = site_rattachement_data['nom_site']
user['site_rattachement_nom'] = site_rattachement_nom
user['site_rattachement_code'] = site_rattachement_code
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 du site de formation "
"""
Supprimer un site de formation
regle :
Un site n'est supprimable que s'il n'est pas utilisé dans les collections :
- ressource_humaine
- ressource_materielle
- session_formation
"""
def Delete_Given_Site_Formation(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', "site_id", ]
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list and val.startswith('my_') is False:
mycommon.myprint(str(
inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token','site_id']
for val in field_list_obligatoire:
if val not in diction:
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ")
return False, " Les informations fournies sont incorrectes"
"""
Verification de l'identité et autorisation de l'entité qui
appelle cette API
"""
token = ""
if ("token" in diction.keys()):
if diction['token']:
token = diction['token']
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
"""
Verifier que la site exist
"""
is_site_existe = MYSY_GV.dbname['site_formation'].count_documents({'_id':ObjectId(str(diction['site_id'])),
'partner_owner_recid':str(my_partner['recid'])})
if( is_site_existe != 1 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " L'identifiant du site est invalide ")
return False, " L'identifiant du site est invalide "
"""
Verifier que le site n'est pas utilisé dans les collections suivantes :
- ressource_humaine
- ressource_materielle
- session_formation
"""
list_collection_to_check = ['ressource_humaine', 'ressource_materielle']
for local_collection in list_collection_to_check:
is_site_id_exist = MYSY_GV.dbname[str(local_collection)].count_documents({'site_formation_id':str(diction['site_id']),
'partner_recid':str(my_partner['recid']),
'valide':'1'})
if( is_site_id_exist != 0 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Ce site de formation est utilisé "+str(is_site_id_exist)+" fois dans "+str(local_collection))
return False, " Ce site de formation est utilisé "+str(is_site_id_exist)+" fois dans "+str(local_collection)
list_collection_to_check = ['partner_owner_recid']
for local_collection in list_collection_to_check:
is_site_id_exist = MYSY_GV.dbname[str(local_collection)].count_documents(
{'site_formation_id': str(diction['site_id']),
'partner_recid': str(my_partner['recid']),
'valide': '1'})
if (is_site_id_exist != 0):
mycommon.myprint(
str(inspect.stack()[0][3]) + " Ce site de formation est utilisé " + str(
is_site_id_exist) + " fois dans " + str(local_collection))
return False, " Ce site de formation est utilisé " + str(is_site_id_exist) + " fois dans " + str(
local_collection)
RetObject = []
val_tmp = 0
qry = {"partner_owner_recid":str(my_partner['recid']), '_id':ObjectId(str(diction['site_id']))}
ret_site_formation = MYSY_GV.dbname['site_formation'].delete_many(qry)
return True, "Le site de formation a été correctement supprimé"
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 supprimer les données du site de formation "