Elyos_FI_Back_Office/prj_common.py

349 lines
9.3 KiB
Python

from difflib import SequenceMatcher
import textdistance
from datetime import datetime
import logging
import secrets
from pymongo import MongoClient
import inspect
from werkzeug.utils import secure_filename
import time
import os
import csv
import inspect
import sys
from datetime import datetime
from pymongo import ReturnDocument
TOKEN_SIZE = 25
CONNECTION_STRING = "mongodb://localhost/cherifdb"
client = MongoClient(CONNECTION_STRING)
dbname = client['cherifdb']
def myprint(message = ""):
logging.info(str(datetime.now()) + " : "+str(message) )
print(str(datetime.now()) + " : " + str(message))
def create_token_urlsafe():
return secrets.token_urlsafe(TOKEN_SIZE)
def create_user_recid():
return secrets.token_hex(TOKEN_SIZE)
'''
Cette fonction recupere et enregistrer un
fichier dans le Dossier "./Data/".
La fonction retourne le nom du fichier
'''
def Upload_Save_CSV_File(file=None, Folder=None):
try:
basename = os.path.basename(file.filename)
basename2 = basename.split(".")
'''
Verification qu'il s'agit bien d'un fichier csv, dont le nom ne comporte pas de "."
'''
if(len(basename2) != 2 ):
myprint(str(inspect.stack()[0][3]) + " - : Le nom du fichier est incorret")
return False, None
if( str(basename2[1]).lower() != "csv" ):
myprint(str(inspect.stack()[0][3]) + " - : Ce n'est pas un fichier csv")
return False, None
timestr = time.strftime("%Y%m%d%H%M%S")
new_file_name = str(basename2[0]) + "_" + str(timestr) + "." + str(basename2[1])
file.filename = new_file_name
file.save(os.path.join(str(Folder), secure_filename(file.filename))) # t
Global_file_name = "./Data/"+file.filename
return True, Global_file_name
except Exception as e :
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False, None
'''
Cette fonction prend un email et un token
puis verifie si la validité du trile (email, token, statut)
'''
def check_token_validity(email="", token=""):
try :
coll_token = dbname['user_token']
tmp_count = coll_token.find(
{ 'token': str(token), 'valide': '1'}).count()
if (tmp_count <= 0):
myprint("Le token n'est pas valide")
return False
return True
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False
'''
Cette fonction prend un email et un token
puis verifie si le compte utilisateur est actif
'''
def check_user_validity(email="", token=""):
try :
coll_token = dbname['user_account']
message = {}
ret = True
for retVal in coll_token.find({ 'token': str(token)}):
user = retVal
if( user['valide'] == '0'):
print(" le compte avec le token : "+token+" n'est pas valide")
message['valide'] = '0'
ret = False
if (user['locked'] == '1'):
print(" le compte avec le token : " + token + " est verrouillé")
message['locked'] = '1'
ret = False
return ret, message
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False, message
'''
Cette fonction prend un token
puis verifie si la validité du trile (email, token, statut) EXCLUSIVEMENT PR LES PARTNER, d
donc la table partner_token
'''
def check_partner_token_validity(email="", token=""):
try :
coll_token = dbname['partner_token']
tmp_count = coll_token.find(
{ 'token': str(token), 'locked':'0', 'valide': '1'}).count()
if (tmp_count <= 0):
myprint("Le token du partenaire n'est pas valide")
return False
return True
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False
'''
recuperation du recid du user
'''
def get_user_recid_from_token(token = ""):
if len(str(token)) <= 0 :
myprint(" Le token est vide")
return False
coll_token = dbname['user_token']
tmp_val = coll_token.find({'token': str(token), 'valide': '1'})
user_recid = tmp_val[0]['recid']
return user_recid
'''
recuperation du recid du partner
'''
def get_parnter_recid_from_token(token = ""):
if len(str(token)) <= 0 :
myprint(" Le token partner est vide")
return False
coll_token = dbname['partner_token']
tmp_val = coll_token.find({'token': str(token), 'valide': '1', 'locked':'0'})
user_recid = tmp_val[0]['recid']
return user_recid
def get_user_recid_from_email(email = ""):
if len(str(email)) <= 0 :
myprint(" L'email est vide")
return False
coll_user = dbname['user_account']
tmp_val = coll_user.find({'email': str(email), 'active': '1'})
user_recid = tmp_val[0]['recid']
return user_recid
'''
Cette fonction créer la reference interne d'une formation
'''
def Create_internal_call_ref():
retval = None
now = datetime.now()
retval = "Mysy_"+now
return retval
def textdist():
val = textdistance.mra("doe", "dough")
print(" mra = "+str(val))
val2 = textdistance.editex("doe", "dough")
print(" editex = " + str(val2))
def similaire():
mots = ["Durand est present", "Meyer", "Dupond", "Dopon", "DUPON", "Nguyen", "Toto"]
ratio = 0.8
for mot in mots:
print(" CMP de 'Dupont' et '"+mot+"'" )
my_ratio = SequenceMatcher(None, "Dupont", mot).ratio()
print(" ## RATION = "+str(my_ratio))
#resultat = [mot for mot in mots if SequenceMatcher(None, "Dupont", mot).ratio() >= ratio]
#print(resultat)
return
def levenshtein(mot1,mot2):
# ligne_i est un tableau tel que tout au long de l'algorithme,
# ligne_i[k] contienne la distance de levenshtein entre les k premières lettres de mot1
# et les i premières lettres de mot2
# Au début, i=0, et la distance entre les k premières lettres de mot1 et la chaîne vide
# vaut bien sûr k. (il faut faire k suppressions pour passer des k premières lettres de mot1
# à la chaîne vide)
ligne_i = [ k for k in range(len(mot1)+1) ]
# i va ensuite varier de 1 à len(mot2)
for i in range(1, len(mot2) + 1):
# i vient d'être incrémenté. On stocke dans ligne_prec la valeur de la ligne numéro i-1
ligne_prec = ligne_i
# On crée la nouvelle ligne, dont le premier élément (l'élement numéro 0) doit être
# la distance de levenshtein entre la chaîne vide ("") et les i premières lettres de mot2, soit i
# (il faut faire i additions pour passer de la chaîne vide aux i premières lettres de mot2)
ligne_i = [i]*(len(mot1)+1)
# On va ensuite remplir le reste de la ligne i, c'est-à-dire calculer ligne_i[k] pour k allant de 1 à len(mot1)
for k in range(1,len(ligne_i)):
# La variable cout vaut 0 si la kième lettre de mot1 est la même que la ième lettre de mot2, et 1 sinon
#La kième lettre de mot1 s'obtient avec mot1[k-1], les indices commencent à 0
cout = int(mot1[k-1] != mot2[i-1])
#Voilà enfin le sel de l'algorithme, le calcul de ligne_i[k] pour i et k quelconques,
# connaissant ligne_prec[k-1], ligne_prec[k] et ligne_i[k-1]
ligne_i[k] = min(ligne_i[k-1] + 1, ligne_prec[k] + 1, ligne_prec[k-1] + cout)
# Lorsque l'on sort de la boucle, i vaut len(mot2)
#Ce que l'on cherche est la distance de levenshtein entre les len(mot1) premières lettres de mot1
# et les len(mot2) premières lettres de mot2, qui est stockée dans ligne_i[len(mot1)]
return ligne_i[len(mot1)]
'''
Check if string is float and convert.
0 if error
'''
def tryFloat(val):
try:
val = str(val).replace(",",".")
myfloat = float(val)
return myfloat
except ValueError:
return 0
'''
Check if string is Int and convert.
0 if error
'''
def tryInt(val):
try:
val = str(val).replace(",",".")
tab_val = val.split(".")
myfloat = int(tab_val[0])
return myfloat
except ValueError:
return 0
'''
Cette fonction verifie si un mot est dans le dictionnaire français - une table interne
si non, le mot est enregistré dans une table pour traitement utérieur.
Utilisé dans le cas de l'indexation d'une formation
'''
def check_word_in_fr_dict(mot=None):
try:
print("#### analyse du mot "+str(mot))
col_name = dbname["list_mots_fr"]
col_name_not_fr = dbname["list_mots_not_fr"]
mydata = {}
val_tmp = col_name.count_documents({'mot': str(mot)})
if (val_tmp <= 0):
myprint(" Le mot '" + mot + "' n'existe pas dans le dictionnaire")
mydata['mot'] = mot
mydata['treated'] = int("0")
mydata['update_date'] = datetime.now()
ret_val = col_name_not_fr.find_one_and_update(
{'mot': str(mot) }, {"$set": mydata},upsert=True,
return_document=ReturnDocument.AFTER
)
if ( ret_val['_id'] is False):
print(" Impossible d'enregistrer le '" + mot + "'")
return False
return False
return True
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False
'''
Cette fonction recherche à savoir si un mot est dans le dictionnaire dans le cadre de la re
recherche d'un utilisateur '''
def recherche_check_word_in_fr_dict(mot=None):
try:
print("#### analyse du mot "+str(mot))
col_name = dbname["list_mots_fr"]
val_tmp = col_name.count_documents({'mot': str(mot)})
if (val_tmp <= 0):
myprint(" Le mot '" + mot + "' n'existe pas dans le dictionnaire")
return False
return True
except Exception as e:
exc_type, exc_obj, exc_tb = sys.exc_info()
myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno))
return False