Elyos_FI_Back_Office/Collection_Historique.py

254 lines
10 KiB
Python

"""
Ce fichier permet de gerer les historique des evenements pour une collection.
L'idée est qu'a travers ce module on puisse voir toutes les actions loggué pour une collection id.
Ce module permet de logguer et de recuperer des event.
"""
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
"""
/!\ : 15/12/2024 : A faire
Ajouter le log vers celui qui a éffectué l'action.
Ceci pour tracer les action d'une pesonne comme la suppression d'un objet.
Idée :
Si Suppression : Mettre l'_id, nom, prenom (ou numero de commande, code) de l'objet supprimé.
"""
"""
Pour un ajout, sont pris en compte :
- related_collection ==> Collection concercée
- l'id de l'entité dans la collection
- attached_doc_id (si une pièce jointe est enregistrée dans la collection : download_files)
- action_date : Date à la quelle l'action a été faite. La date doit etre au format : d/%m/%Y %H:%M:%S
/!\ : Il n'y pas d'update. Pour corriger un event, il faut logguer un autre event
/!\ 20/04/25 : on ajoute un champ 'technical_comment' qui permet de logger des detail d'ordre technique qui permet de
mieux logger les evenements critiques comme la modification des droits d'acces, les mots de passe, etc
"""
def Add_Historique_Event(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'related_collection', 'related_collection_recid', 'attached_doc_id', 'action_date',
'action_description', 'technical_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é")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
field_list_obligatoire =['token', 'related_collection', 'related_collection_recid', 'action_date',
'action_description']
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
# Verifier que l'action_date est bien au format: %d/%m/%Y %H:%M:%S
if( mycommon.CheckisDate_Hours_Second(str(diction['action_date'])) is False):
mycommon.myprint(
str(inspect.stack()[0][3]) + " La date de l'action doit être au format jj/mm/aaaa hh:mm:ss")
return False, " La date de l'action doit être au format jj/mm/aaaa hh:mm:ss"
# Verifier que 'related_collection_recid' existe et est valide dans la collection 'related_collection'
is_existe_valide_related_collection_recid = MYSY_GV.dbname[str(diction['related_collection'])].count_documents({'_id':ObjectId(str(diction['related_collection_recid'])),
'partner_owner_recid':str(my_partner['recid']),
})
if( is_existe_valide_related_collection_recid <= 0):
# On verifie si on a pas plutot un 'partner_recid' au lieu d'un 'partner_owner_recid'
is_existe_valide_related_collection_recid2 = MYSY_GV.dbname[
str(diction['related_collection'])].count_documents(
{'_id': ObjectId(str(diction['related_collection_recid'])),
'partner_recid': str(my_partner['recid']),
})
if( is_existe_valide_related_collection_recid2 is None ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - L'identifiant du 'related_collection_recid' n'est pas valide ")
return False, " L'identifiant du 'related_collection_recid' n'est pas valide "
# Verifier l'existance de la pièce jointe (si pièce jointe) : attached_doc_id
if( 'attached_doc_id' in diction.keys() and diction['attached_doc_id']):
is_existe_valide_attached_doc_id = MYSY_GV.dbname['download_files'].count_documents(
{'_id': ObjectId(str(diction['attached_doc_id'])),
'partner_recid': str(my_partner['recid']),
'valide': '1',
'locked': '0'})
if (is_existe_valide_attached_doc_id <= 0):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - L'identifiant du 'is_existe_valide_attached_doc_id' n'est pas valide ")
return False, " L'identifiant du 'is_existe_valide_attached_doc_id' n'est pas valide "
# La description ne doit pas faire plus de 255 caractère
action_description = ""
if('action_description' in diction.keys() and diction['action_description']):
action_description = diction['action_description']
if(len(str(action_description)) > 255 ):
mycommon.myprint(
str(inspect.stack()[0][3]) + " - La description de l'historique d'évènement fait plus de 255 caractères ")
return False, " La description de l'historique d'évènement fait plus de 255 caractères "
mydata = diction
if 'token' in mydata:
del mydata['token']
mydata['partner_owner_recid'] = my_partner['recid']
mydata['connected_user_id'] = str(my_partner['_id'])
mydata['valide'] = "1"
mydata['locked'] = "0"
mydata['date_update'] = str(datetime.now().strftime("%d/%m/%Y %H:%M:%S"))
mydata['update_by'] = str(my_partner['_id'])
inserted_id = ""
inserted_id = MYSY_GV.dbname['historique_action'].insert_one(mydata).inserted_id
if (not inserted_id):
mycommon.myprint(
" Impossible d'ajouter l'historique d'évènement (1) ")
return False, " Impossible d'ajouter l'historique d'évènement (1) "
return True, " L'événement a été ajouté à l'historique"
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'ajouter l'historique d'évènement "
"""
Recupération de l'historique des actions associé à un element.
Un element est defini par :
- related_collection
- related_collection_recid
"""
def Get_Action_Historique_List(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = ['token', 'related_collection', 'related_collection_recid']
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'existe pas")
return False, " Les informations fournies sont incorrectes",
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'related_collection', 'related_collection_recid']
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
RetObject = []
val_tmp = 1
for New_retVal in MYSY_GV.dbname['historique_action'].find({'related_collection':str(diction['related_collection']), 'related_collection_recid':str(diction['related_collection_recid']),
'partner_owner_recid':str(my_partner['recid']), 'valide':'1', 'locked':'0'}).limit(MYSY_GV.HISTORIQUE_MAX_EVENT).sort([("_id", pymongo.DESCENDING)]):
user = New_retVal
user['id'] = str(val_tmp)
# Recuperation de l'email du user qui fait l'action
connected_user_data = MYSY_GV.dbname['partnair_account'].find_one({'_id':ObjectId(str(New_retVal['connected_user_id'])), 'recid':str(my_partner['recid'])}, {'nom':1,'email':1 })
if( connected_user_data ):
user['connected_user_nom'] = connected_user_data['nom']
user['connected_user_email'] = connected_user_data['email']
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 évènements "