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 from unidecode import unidecode 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 ''' Dans le cadre de la recherche, l'expression saisit par l'utilisateur dans la search bar doit etre nettoyé, traité avant de rentrer dans le process. Puis la phrase est renvoyée en mode "unicode" ''' def Parse_Clean_Search_Text(sentence=None): try: if (len(str(sentence)) <= 0 ): return False, "" ''' /!\ : On supprime tous les caratère "spaciaux" et ponctuation EXCEPTE - le ":" dont on a besoin pour identifier les patterns et - le ' " ' dont on a besoin pour identifier les patterns ''' list_noises = ['...', '.', ';', ',', '!', '?', ')', '(', '[', ']', '\'', '’', '`', '©', '–', '{', '}', '-', '=', '°', '#', '-', '/', '~', '&', '\\', '.', '^', '$', '*', '+', '\\n', '?', '{', '}', '[', ']', '|', '(', ')', '-', '>', '<', '@', '®', '™', '«', '»'] for noise in list_noises: # print(" suppression de : '"+str(noise)+"' ") sentence = sentence.replace(str(noise), " ") unicode_sentence = unidecode(sentence) return True, unicode_sentence 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, ""