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'] AUTORIZED_SOURCE_IPV4=["127.0.0.1", "localhost", "88.170.110.220", "192.168.1.21", "192.168.1.48"] 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.count_documents({ 'token': str(token), 'valide': '1'}) 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.count_documents({ 'token': str(token), 'locked':'0', 'valide': '1'}) #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 d'email du user from token ''' def get_user_email_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_email = tmp_val[0]['email'] return user_email ''' recuperation d'email du user from recid ''' def get_user_email_from_recid(recid = ""): try: if len(str(recid)) <= 0 : myprint(" Le recid est vide") return False coll_token = dbname['user_account'] tmp_val = coll_token.find({'recid': str(recid), 'active': '1'}) if( tmp_val and tmp_val[0] and tmp_val[0]['email']): user_email = tmp_val[0]['email'] else: return False, "Impossible de supprimer le compte utilisateur" return user_email 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, "Impossible de supprimer le compte utilisateur" ''' 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 ''' Verification que le mot n'est pas stemisable à traver la tabla "word_not_stem ''' def Word_Not_Stemmize(word = None): try: coll_not_stem = dbname["word_not_stem"] val_tmp = coll_not_stem.count_documents({'mot': str(word)}) if (val_tmp > 0): return True else: return False 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, " Impossible de verifier Word_Not_Stemmize" ''' 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 ''' Securité : Ip source requester Cette fonction verifie si l'adresse IP de la source est autorisé ou pas. ''' def check_source_ipv4(source_ip=None): try: if source_ip in AUTORIZED_SOURCE_IPV4: myprint(" Security check : IP adresse '"+str(source_ip)+"' connected") return True else: myprint(" Security check : IP adresse '" + str(source_ip) + "' is not autorized") return False 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