627 lines
17 KiB
Python
627 lines
17 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
|
||
from unidecode import unidecode
|
||
import GlobalVariable as MYSY_GV
|
||
from serpapi import GoogleSearch
|
||
import prj_common as mycommon
|
||
import re
|
||
|
||
|
||
def myprint(message = ""):
|
||
logging.info(str(datetime.now()) + " : "+str(message) )
|
||
print(str(datetime.now()) + " : " + str(message))
|
||
|
||
|
||
def create_token_urlsafe():
|
||
return secrets.token_urlsafe(MYSY_GV.TOKEN_SIZE)
|
||
|
||
|
||
def create_user_recid():
|
||
return secrets.token_hex(MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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 = MYSY_GV.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):
|
||
try:
|
||
# 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)]
|
||
|
||
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, None
|
||
|
||
|
||
'''
|
||
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 = MYSY_GV.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 = MYSY_GV.dbname["list_mots_fr"]
|
||
col_name_not_fr = MYSY_GV.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 = MYSY_GV.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 MYSY_GV.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, ""
|
||
|
||
|
||
'''
|
||
Cette fonction va aller cherche la recherche edendue en
|
||
utilisant l'api de "serpapi"
|
||
'''
|
||
def Get_Extended_Result(sentence=None):
|
||
try:
|
||
list_extended = []
|
||
external_code_prefixe = str(datetime.now().timestamp()).replace(".", "")
|
||
print("external_code_prefixe = " + str(external_code_prefixe))
|
||
|
||
|
||
status = RunSearchAPI(sentence, external_code_prefixe)
|
||
if( status is False):
|
||
return False, list_extended
|
||
|
||
return True, external_code_prefixe
|
||
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 lance l'API et enregistre le resultat en base
|
||
'''
|
||
def RunSearchAPI(search_text=None, external_code_prefixe=None):
|
||
try:
|
||
if( len(str(search_text).strip()) < 0 ):
|
||
return True
|
||
|
||
params = {
|
||
"q": str(search_text),
|
||
"hl": "fr",
|
||
"gl": "fr",
|
||
"num":"20",
|
||
"safe": "active",
|
||
"google_domain": "google.com",
|
||
"api_key": "596cb9a468f8292fcefa6f297444db9c12478685d8734b52efdf8aa53c54fd55"
|
||
}
|
||
|
||
mycommon.myprint("######## PARAM RunSearchAPI = "+str(params))
|
||
|
||
'''
|
||
return false prematuré pour eviter de consommer les credits de l'API
|
||
'''
|
||
return False
|
||
|
||
|
||
search = GoogleSearch(params)
|
||
results = search.get_dict()
|
||
organic_results = results['organic_results']
|
||
|
||
my_collection = MYSY_GV.YTUBES_dbname['mysyserpapi']
|
||
|
||
mycommon.myprint("resutlat 1 organic_results ")
|
||
cmpt = 0
|
||
for val in organic_results:
|
||
cmpt = cmpt +1
|
||
mydata = {}
|
||
'''
|
||
print("External_code = " + str(val['title']))
|
||
print("Title = "+str(val['title']))
|
||
print("url = " + str(val['link']))
|
||
print("description = " + str(val['snippet']))
|
||
|
||
if ("snippet_highlighted_words" in val.keys()):
|
||
if val['snippet_highlighted_words']:
|
||
print("mot_cles = " + str(val['snippet_highlighted_words']))
|
||
'''
|
||
|
||
mydata['external_code'] = external_code_prefixe+"_"+str(cmpt)
|
||
mydata['title'] = str(val['title'])
|
||
mydata['url'] = str(val['link'])
|
||
|
||
if ("snippet" in val.keys()):
|
||
if val['snippet']:
|
||
mydata['description'] = str(val['snippet'])
|
||
|
||
if ("snippet_highlighted_words" in val.keys()):
|
||
if val['snippet_highlighted_words']:
|
||
mydata['tags'] = str(val['snippet_highlighted_words'])
|
||
|
||
mydata['update_date'] = str(datetime.now())
|
||
|
||
if ("position" in val.keys()):
|
||
if val['position']:
|
||
mydata['rang'] = str(val['position'])
|
||
|
||
mydata['orign_search_text'] = str(search_text)
|
||
mydata['valide'] = "1"
|
||
mydata['treated'] = "0"
|
||
|
||
ret_val = my_collection.find_one_and_update({'url': str(mydata['url'])},
|
||
{"$set": mydata},
|
||
upsert=True,
|
||
return_document=ReturnDocument.AFTER
|
||
)
|
||
|
||
if ret_val and ret_val['_id']:
|
||
mycommon.myprint(" Le document de la recherche étendu a bien été ajouté = " + str(ret_val['_id']))
|
||
|
||
else:
|
||
mycommon.myprint(" WARNING : Impossible d'ajouter le document de la recherche étentue " + str(mydata['url']))
|
||
|
||
return True
|
||
except Exception as e:
|
||
exc_type, exc_obj, exc_tb = sys.exc_info()
|
||
mycommon.myprint(str(e) + " - Line : " + str(exc_tb.tb_lineno))
|
||
return False
|
||
|
||
|
||
|
||
'''
|
||
Cette fonction prends un mot,
|
||
et retourne sont masculin / singulier
|
||
Cette fonction va s'etoffer au fur et à mesure
|
||
|
||
/!\ Cette fonction prend un pot, pas une phrase
|
||
'''
|
||
def GetMasculinSingulier(word=None):
|
||
try:
|
||
mot = str(word).lower() # mettre les mots en minuscule
|
||
# Retirons les caractères spéciaux :
|
||
|
||
# mots terminés par "es"
|
||
patter2 = re.compile(r"\w+(ees)+$")
|
||
|
||
# mots terminés par "s"
|
||
patter3 = re.compile(r"\w+(s)+$")
|
||
|
||
# mots terminés par "x"
|
||
patter4 = re.compile(r"\w+(x)+$")
|
||
|
||
|
||
if (len(str(mot)) > 3):
|
||
if( re.match(patter2, str(mot))):
|
||
neword = mot[:-2]
|
||
return True, neword
|
||
|
||
if (re.match(patter3, str(mot))):
|
||
neword = mot[:-1]
|
||
return True, neword
|
||
|
||
if (re.match(patter4, str(mot))):
|
||
neword = mot[:-1]
|
||
return True, neword
|
||
|
||
|
||
return True, str(mot)
|
||
|
||
|
||
except Exception as e:
|
||
exc_type, exc_obj, exc_tb = sys.exc_info()
|
||
mycommon.myprint(str(e) + " - Line : " + str(exc_tb.tb_lineno))
|
||
return False, ""
|