''' Cette class gere la help page (articles et video de tuto en ligne) . ''' import pysftp from pymongo import MongoClient import pymongo import hashlib import json from bson import ObjectId from unidecode import unidecode import ela_index_bdd_classes import ela_index_bdd_classes as ela_index import logging from datetime import datetime, date import prj_common as mycommon import secrets from pymongo import ReturnDocument import inspect import sys, os import GlobalVariable as MYSY_GV import user_message_mgt as umm import Ela_Spacy as ES import GlobalVariable as MYSY_GV import re class JSONEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, ObjectId): return str(o) print(self) print(o) return json.JSONEncoder.default(self, o) ''' Recuperation des tous les help articles, quelque soit le statut (publié et non publié) ''' def get_all_help_article(diction): try: # Dictionnaire des champs utilisables field_list = ['token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'search_text'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer les articles et avis" val_tmp = 0 insertObject = [] todays_date = str(date.today().strftime("%Y-%m-%d")) filt_date_today_ISODATE = datetime.strptime(str(todays_date), '%Y-%m-%d') coll_name = MYSY_GV.dbname['articles_help'] val_tmp = 1 insertObject = [] for x in coll_name.find({'valide': '1', 'locked': '0'}, { "indexed": 0, "indexed_desc": 0, "indexed_obj": 0, "indexed_title": 0, "valide": 0, "locked": 0, }).sort( [ ("date_avis",pymongo.DESCENDING), ("title_formation",pymongo.ASCENDING),]): # mycommon.myprint("AVANT ==> "+str(x['description'])) if( "qualite" in x.keys() ): val = x['qualite'] if (len(x['qualite']) > MYSY_GV.MAX_CARACT): x['qualite'] = val[:MYSY_GV.MAX_CARACT] + " ..." else: x['qualite'] = val[:MYSY_GV.MAX_CARACT] else : x['qualite'] = "--" val = x['avis'] CLEANR = re.compile('<.*?>') clean_avis = re.sub(CLEANR, '', str(val)) clean_avis = str(clean_avis).replace(" ", " "); #print(" ### CLEANRCLEANR = "+str(clean_avis)) if (len(clean_avis) > MYSY_GV.MAX_CARACT): x['avis'] = clean_avis[:MYSY_GV.MAX_CARACT] + " ..." else: x['avis'] = str(clean_avis) # mycommon.myprint("APRES ==> " + str(x['description'])) # gestion date publication is_published = "0" if( "date_publication" in x.keys() and x['date_publication']): if (datetime.strptime(str(x['date_publication']).strip()[0:10], '%Y-%m-%d') <= datetime.strptime(str(date.today().strftime("%Y-%m-%d")).strip(), '%Y-%m-%d')): is_published = "1" x['is_published'] = str(is_published) if("support_type" not in x.keys() ): x['support_type'] = "" user = x user['id'] = str(val_tmp) insertObject.append(mycommon.JSONEncoder().encode(user)) val_tmp = val_tmp + 1 #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les articles et avis" """ Recuperation des help_article publiés, c'est à dire valide et dont la date de publication est arrivée ou passée """ def get_all_published_help_article(diction): try: # Dictionnaire des champs utilisables field_list = ['token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'search_text'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer les articles et avis" todays_date = str(date.today().strftime("%Y-%m-%d")) filt_date_today_ISODATE = datetime.strptime(str(todays_date), '%Y-%m-%d') qery_match_published = {'$and': [ {'valide': '1', 'locked': '0'}, { 'mysy_article_date_publication': {'$lte': filt_date_today_ISODATE}}, ]} query_pipe_published = ([ {"$addFields": { "mysy_article_date_publication": { '$dateFromString': { 'dateString': {"$substr": ["$date_publication", 0, 10]}, "format": "%Y-%m-%d" } } } }, {'$match': qery_match_published }, { '$sort': {'mysy_article_date_publication': -1} }, ]) print(" ### query_pipe = ", query_pipe_published ) val_tmp = 1 insertObject = [] for x in MYSY_GV.dbname['articles_help'].aggregate(query_pipe_published): if( "qualite" in x.keys() ): val = x['qualite'] if (len(x['qualite']) > MYSY_GV.MAX_CARACT): x['qualite'] = val[:MYSY_GV.MAX_CARACT] + " ..." else: x['qualite'] = val[:MYSY_GV.MAX_CARACT] else : x['qualite'] = "--" val = x['avis'] CLEANR = re.compile('<.*?>') clean_avis = re.sub(CLEANR, '', str(val)) clean_avis = str(clean_avis).replace(" ", " "); #print(" ### CLEANRCLEANR = "+str(clean_avis)) if (len(clean_avis) > MYSY_GV.MAX_CARACT): x['avis'] = clean_avis[:MYSY_GV.MAX_CARACT] + " ..." else: x['avis'] = str(clean_avis) # mycommon.myprint("APRES ==> " + str(x['description'])) x['is_published'] = "1" if ("support_type" not in x.keys()): x['support_type'] = "" user = x user['id'] = str(val_tmp) insertObject.append(mycommon.JSONEncoder().encode(user)) val_tmp = val_tmp + 1 #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les articles et avis" """ Recuperation des help_article non publiés, c'est à dire valide et dont la date de publication N'EST PAS arrivée ou passée """ def get_all_non_published_help_article(diction): try: # Dictionnaire des champs utilisables field_list = ['token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'search_text'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer les articles et avis" todays_date = str(date.today().strftime("%Y-%m-%d")) filt_date_today_ISODATE = datetime.strptime(str(todays_date), '%Y-%m-%d') qery_match_non_published = {'$and': [ {'valide': '1', 'locked': '0'}, { 'mysy_article_date_publication': {'$gt': filt_date_today_ISODATE}}, ]} query_pipe_non_published = ([ {"$addFields": { "mysy_article_date_publication": { '$dateFromString': { 'dateString': {"$substr": ["$date_publication", 0, 10]}, "format": "%Y-%m-%d" } } } }, {'$match': qery_match_non_published }, { '$sort': {'mysy_article_date_publication': -1} }, ]) print(" ### query_pipe_non_published = ", query_pipe_non_published) val_tmp = 1 insertObject = [] for x in MYSY_GV.dbname['articles_help'].aggregate(query_pipe_non_published): if ("qualite" in x.keys()): val = x['qualite'] if (len(x['qualite']) > MYSY_GV.MAX_CARACT): x['qualite'] = val[:MYSY_GV.MAX_CARACT] + " ..." else: x['qualite'] = val[:MYSY_GV.MAX_CARACT] else: x['qualite'] = "--" val = x['avis'] CLEANR = re.compile('<.*?>') clean_avis = re.sub(CLEANR, '', str(val)) clean_avis = str(clean_avis).replace(" ", " "); # print(" ### CLEANRCLEANR = "+str(clean_avis)) if (len(clean_avis) > MYSY_GV.MAX_CARACT): x['avis'] = clean_avis[:MYSY_GV.MAX_CARACT] + " ..." else: x['avis'] = str(clean_avis) # mycommon.myprint("APRES ==> " + str(x['description'])) x['is_published'] = "0" if ("support_type" not in x.keys()): x['support_type'] = "" user = x user['id'] = str(val_tmp) insertObject.append(mycommon.JSONEncoder().encode(user)) val_tmp = val_tmp + 1 # print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les articles et avis" ''' Recuperation des 3 derniers help_article tous avis et commentaires difference de celui en argument (les 3 derniers autre que celui en argument) ''' def get_last_3_help_article(diction): try: # Dictionnaire des champs utilisables field_list = ['token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'internal_url'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer les articles et avis" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', "internal_url", ] 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 la liste des arguments ") return False, " Les informations fournies sont incorrectes" coll_name = MYSY_GV.dbname['articles_help'] qry = {'internal_url' :{'$ne':str(diction['internal_url'])}, 'valide': '1', 'locked': '0'} #print(" ### qry = ", qry) val_tmp = 0 insertObject = [] for x in coll_name.find(qry, { "indexed": 0, "indexed_desc": 0, "indexed_obj": 0, "indexed_title": 0, "valide": 0, "locked": 0, }).limit(3).sort( [ ("date_avis",pymongo.DESCENDING), ("title_formation",pymongo.ASCENDING),]): # mycommon.myprint("AVANT ==> "+str(x['description'])) if( "qualite" in x.keys() ): val = x['qualite'] if (len(x['qualite']) > MYSY_GV.MAX_CARACT_DEDUIT): x['qualite'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] + " ..." else: x['qualite'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] val = x['avis'] CLEANR = re.compile('<.*?>') clean_avis = re.sub(CLEANR, '', str(val)) clean_avis = str(clean_avis).replace(" ", " "); #print(" ### CLEANRCLEANR = "+str(clean_avis)) if (len(clean_avis) > MYSY_GV.MAX_CARACT_DEDUIT): x['avis'] = clean_avis[:MYSY_GV.MAX_CARACT_DEDUIT] + " ..." else: x['avis'] = str(clean_avis) # mycommon.myprint("APRES ==> " + str(x['description'])) if ("support_type" not in x.keys()): x['support_type'] = "" user = x user['id'] = str(val_tmp) insertObject.append(mycommon.JSONEncoder().encode(user)) val_tmp = val_tmp + 1 #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer les articles et avis" ''' Cette API recupere et retourne un help_article ''' def get_help_article(diction): try: # Dictionnaire des champs utilisables field_list = ['internal_url', 'token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city','user_postal', 'user_latitude', 'user_longitude', 'user_state'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer l'articles/avis" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['internal_url', 'token'] 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 la liste des arguments ") return False, " Impossible de récupérer l'articles/avis" # Recuperation des parametres mydata = {} if ("user_ip" in diction.keys()): if diction['user_ip']: mydata['user_ip'] = diction['user_ip'] if ("user_country_code" in diction.keys()): if diction['user_country_code']: mydata['user_country_code'] = diction['user_country_code'] if ("user_country_name" in diction.keys()): if diction['user_country_name']: mydata['user_country_name'] = diction['user_country_name'] if ("user_city" in diction.keys()): if diction['user_city']: mydata['user_city'] = diction['user_city'] if ("user_postal" in diction.keys()): if diction['user_postal']: mydata['user_postal'] = diction['user_postal'] if ("user_latitude" in diction.keys()): if diction['user_latitude']: mydata['user_latitude'] = diction['user_latitude'] if ("user_longitude" in diction.keys()): if diction['user_longitude']: mydata['user_longitude'] = diction['user_longitude'] if ("user_state" in diction.keys()): if diction['user_state']: mydata['user_state'] = diction['user_state'] my_title_formation = '' if ("title_formation" in diction.keys()): if diction['title_formation']: mydata['title_formation'] = diction['title_formation'] my_title_formation = diction['title_formation'] my_internal_id = '' if ("internal_url" in diction.keys()): if diction['internal_url']: my_internal_id = diction['internal_url'] #print(" My Data =" + str(mydata)) coll_name = MYSY_GV.dbname['articles_help'] insertObject = [] for x in coll_name.find({'valide': '1', 'locked': '0', 'internal_url':my_internal_id}, { "indexed": 0, "indexed_desc": 0, "indexed_obj": 0, "indexed_title": 0, "valide": 0, "locked": 0, }).sort([("title_formation",pymongo.ASCENDING), ("date_avis",pymongo.ASCENDING), ("note", pymongo.DESCENDING), ]) : if ("support_type" not in x.keys()): x['support_type'] = "" user = x insertObject.append(mycommon.JSONEncoder().encode(user)) #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer l'article/avis" ''' Cette API recupere et retourne un help_article en partant de l'_id ''' def get_help_article_from_id(diction): try: # Dictionnaire des champs utilisables field_list = ['internal_url', 'token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city','user_postal', 'user_latitude', 'user_longitude', 'user_state', '_id'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer l'articles/avis" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['_id',] 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 la liste des arguments ") return False, " Impossible de récupérer l'articles/avis" #print(" My Data =" + str(mydata)) coll_name = MYSY_GV.dbname['articles_help'] insertObject = [] cpt = 0 for x in coll_name.find({'valide': '1', 'locked': '0', '_id':ObjectId(str(diction['_id']))}, { "indexed": 0, "indexed_desc": 0, "indexed_obj": 0, "indexed_title": 0, "valide": 0, "locked": 0, }).sort([("title_formation",pymongo.ASCENDING), ("date_avis",pymongo.ASCENDING), ("note", pymongo.DESCENDING), ]) : cpt = cpt + 1 if ("support_type" not in x.keys()): x['support_type'] = "" user = x insertObject.append(mycommon.JSONEncoder().encode(user)) if( cpt == 0 ): return False, "Identifiant de l'article est invalide" #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer l'article/avis" """ Suppression d'un help_article """ def delete_help_article_from_id(diction): try: # Dictionnaire des champs utilisables field_list = ['token', '_id'] incom_keys = diction.keys() ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' for val in incom_keys: if str(val).lower() not in str(field_list).lower(): mycommon.myprint(str(inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Impossible de récupérer l'articles/avis" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['_id',] 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 la liste des arguments ") return False, " Impossible de récupérer l'articles/avis" """ 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 la validité de l'article à supprimer """ is_valide_article_avis_count = MYSY_GV.dbname['articles_help'].count_documents( {'_id': ObjectId(str(diction['_id'])), 'valide': '1'}) if (is_valide_article_avis_count != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'article n'est pas valide ") return False, " L'identifiant de l'article n'est pas valide " delete = MYSY_GV.dbname['articles_help'].delete_one({'_id': ObjectId(str(diction['_id']))}) #print(" getObject = ", str(insertObject)) return True, "L'article a été correctement supprimé" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de supprimer l'article/avis" ''' Cette fonction crée un articles avis, avec les element de traàabilité : - user_recid, - IP, - etc Une fois que l'avis est bien ajouté, une notification est envoyer dans la collection 'user_message' à destination de l'adresse de contact du site. Important : : 30/12/2024 : on va reconstituer le champs : 'img_path' en faisant img_path = MYSY_GV.BLOG_IMG_URL+img_file_name ''' def add_help_article(diction): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token', 'title_formation', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'type_formation', 'qualite', 'avis', 'note', 'qualityrating','avisrating', 'comment', 'postedby', 'url_formation', 'user_recid', 'display_header_img', 'img_path', 'date_publication', 'meta_description', 'img_file_name', 'File', 'internal_url', 'video_path', 'video_path_name', 'a_retenir'] 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]) + " - Creation formation : Le champ '" + val + "' n'est pas autorisé, Creation formation annulée") return False, " Verifier votre API" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['title_formation', 'token', 'avis', 'user_ip', 'qualite', 'note'] 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 la liste des arguments ") return False, " Verifier les parametres de votre API" ''' Verification si le token et l'email sont valident ''' # recuperation des paramettre mydata = {} """ 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({'token': str(token)}) if (local_status is not True): return local_status, my_partner my_user_recid = my_partner['recid'] mydata['user_recid'] = my_user_recid mydata['token'] = token if ("title_formation" in diction.keys()): if diction['title_formation']: mydata['title_formation'] = diction['title_formation'] ''' Gestion de l'url interne. Pr un meilleur SEO, le champ 'internal_url' = titre avec des tirets, si une valeur 'internal_url' n'est pas fournie. si une valeur diction['internal_url'], on utilise celle-ci si elle est disponible en base, si non on refuse l'enregistrement et on demande a l'utilisateur de choisir un aurtre 'internal_url' ''' if( "internal_url" in diction.keys() and diction['internal_url']): my_internal_url = str(diction['internal_url']) is_my_internal_url_dispo_count = MYSY_GV.dbname['articles_help'].count_documents( {"internal_url": str(my_internal_url)}) if (is_my_internal_url_dispo_count > 0): mycommon.myprint( str(inspect.stack()[0][ 3]) + " - L'url souhaitée est déjà utilisée. Merci de choisir une autre.") return False, " L'url souhaitée est déjà utilisée. Merci de choisir une autre." else: my_internal_url = str(diction['title_formation']).replace("-","").replace(",", "").replace(" ", "").strip() local_status, my_internal_url = ES.Ela_Remove_Ponct_Special_Caractere(my_internal_url) my_internal_url = unidecode(my_internal_url.lower()) if (my_internal_url.startswith('-')): my_internal_url = my_internal_url[1:] if (my_internal_url.endswith('-')): my_internal_url = my_internal_url[:-1] my_internal_url = my_internal_url.replace(" ","-") my_internal_url = my_internal_url.replace("----", "-").replace("---", "-").replace("--", "-").replace("--", "-") while( my_internal_url.endswith("-")): my_internal_url = my_internal_url[:-1] print(" #### my_internal_url = ", my_internal_url) """ Important : 30/12/2024 : Avant d'ajouter un suffixe numerique, verifier si l'url est disponible ou pas. Si elle n'est pas disponible, on ajoute un suffixe jusqu'a trouver une url disponible """ url_dispo = False cpt = 0 while ( url_dispo is False ): is_my_internal_url_dispo_count = MYSY_GV.dbname['articles_help'].count_documents({"internal_url":str(my_internal_url)}) if( is_my_internal_url_dispo_count > 0 ): my_internal_url = str(my_internal_url) + "-" + str(cpt) cpt = cpt + 1 else: url_dispo = True mydata['internal_url'] = my_internal_url if(len(str(my_internal_url)) > 200 ): mycommon.myprint( str(inspect.stack()[0][3]) + " - L'url ne doit pas dépasser 200 caractères. Le votre fait "+str(len(str(my_internal_url)))+" caractères") return False, " L'url ne doit pas dépasser 200 caractères. Le votre fait "+str(len(str(my_internal_url)))+" caractères " if ("display_header_img" in diction.keys()): if diction['display_header_img']: mydata['display_header_img'] = diction['display_header_img'] if ("date_publication" in diction.keys()): if diction['date_publication']: date_publication = str(diction['date_publication']).strip() local_status = mycommon.CheckisDate(date_publication) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " La date de publication n'est pas au format 'jj/mm/aaaa' ") return False, " La date de publication n'est pas au format 'jj/mm/aaaa' " date_str = str(date_publication)[0:10] date_format = '%d/%m/%Y' date_obj = datetime.strptime(date_str, date_format) formated_date_publication = str(date_obj.strftime("%Y-%m-%d %H:%M:%S")) mydata['date_publication'] = formated_date_publication if ("user_ip" in diction.keys()): if diction['user_ip']: mydata['user_ip'] = diction['user_ip'] if ("user_country_code" in diction.keys()): if diction['user_country_code']: mydata['user_country_code'] = diction['user_country_code'] if ("user_country_name" in diction.keys()): if diction['user_country_name']: mydata['user_country_name'] = diction['user_country_name'] if ("user_city" in diction.keys()): if diction['user_city']: mydata['user_city'] = diction['user_city'] if ("user_postal" in diction.keys()): if diction['user_postal']: mydata['user_postal'] = diction['user_postal'] if ("user_latitude" in diction.keys()): if diction['user_latitude']: mydata['user_latitude'] = diction['user_latitude'] if ("user_longitude" in diction.keys()): if diction['user_longitude']: mydata['user_longitude'] = diction['user_longitude'] if ("user_state" in diction.keys()): if diction['user_state']: mydata['user_state'] = diction['user_state'] if ("type_formation" in diction.keys()): if diction['type_formation']: mydata['type_formation'] = diction['type_formation'] if ("qualite" in diction.keys()): if diction['qualite']: mydata['qualite'] = diction['qualite'] if ("avis" in diction.keys()): if diction['avis']: mydata['avis'] = diction['avis'] if ("note" in diction.keys()): if diction['note']: mydata['note'] = diction['note'] if ("comment" in diction.keys()): if diction['comment']: mydata['comment'] = diction['comment'] if ("postedby" in diction.keys()): if diction['postedby']: mydata['postedby'] = diction['postedby'] if ("url_formation" in diction.keys()): if diction['url_formation']: mydata['url_formation'] = diction['url_formation'] if ("avisrating" in diction.keys()): if diction['avisrating']: mydata['avisrating'] = diction['avisrating'] if ("img_file_name" in diction.keys()): if diction['img_file_name']: mydata['img_file_name'] = diction['img_file_name'] mydata['img_path'] = MYSY_GV.BLOG_IMG_URL+str(diction['img_file_name']) mydata['support_type'] = "img" if ("video_path" in diction.keys()): mydata['video_path'] = str(diction['video_path']) mydata['support_type'] = "video" meta_description = "" if ("meta_description" in diction.keys()): if diction['meta_description']: meta_description = str(diction['meta_description']).strip().lower()[0:200] else: meta_description = str(diction['title_formation']).strip().lower()[0:200] mydata['meta_description'] = meta_description a_retenir = "" if ("a_retenir" in diction.keys()): if diction['a_retenir']: a_retenir = str(diction['a_retenir']).strip().lower()[0:1000] mydata['a_retenir'] = a_retenir mydata['valide'] = '1' mydata['locked'] = '0' mydata['indexed'] = '0' mydata['indexed'] = '0' mydata['update_date'] = str(datetime.now()) mydata['date_avis'] = str(datetime.now()) mydata['isalaune'] = "0" #print(mydata) coll_name = MYSY_GV.dbname['articles_help'] ret_val = coll_name.insert_one(mydata) """ On lance l'indexation de l'article """ retval = ela_index_bdd_classes.ela_index_articles_help(str(ret_val.inserted_id), my_partner['recid']) return True, " L'article support a été bien ajouté" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de créer l'article/avis" """ Cette fonction met à jour un article avis Important : : la mise à jour du titre ne modifie pas l'url interne car elle est potentiellement deja utilisée Important : : 30/12/2024 : on va reconstituer le champs : 'img_path' en faisant img_path = MYSY_GV.BLOG_IMG_URL+img_file_name """ def update_help_article(diction): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token', 'title_formation', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'type_formation', 'qualite', 'avis', 'note', 'qualityrating', 'avisrating', 'comment', 'postedby', 'url_formation', 'user_recid', '_id','display_header_img', 'img_path', 'date_publication', 'meta_description','img_file_name', 'File', 'video_path', 'video_path_name', 'a_retenir', 'internal_url' ] 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]) + " - Creation formation : Le champ '" + val + "' n'est pas autorisé, Creation formation annulée") return False, " Verifier votre API" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['title_formation', 'token', 'avis', 'user_ip', 'qualite', 'note', '_id'] 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 la liste des arguments ") return False, " Verifier les parametres de votre API" ''' Verification si le token et l'email sont valident ''' # recuperation des paramettre mydata = {} my_user_recid = 'None' my_token = 'None' """ 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({'token':str(token)}) if (local_status is not True): return local_status, my_partner """ Verifier la validité de l'article à mettre à jour """ is_valide_article_avis_count = MYSY_GV.dbname['articles_help'].count_documents({'_id':ObjectId(str(diction['_id'])), 'valide':'1'}) if (is_valide_article_avis_count != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'article n'est pas valide ") return False, " L'identifiant de l'article n'est pas valide " mydata['user_recid'] = my_partner['recid'] mydata['token'] = my_token if ("title_formation" in diction.keys()): if diction['title_formation']: mydata['title_formation'] = diction['title_formation'] if ("display_header_img" in diction.keys()): if diction['display_header_img']: mydata['display_header_img'] = diction['display_header_img'] if ("img_path" in diction.keys()): if diction['img_path']: mydata['img_path'] = diction['img_path'] if ("img_file_name" in diction.keys()): if diction['img_file_name']: mydata['img_file_name'] = diction['img_file_name'] mydata['img_path'] = MYSY_GV.BLOG_IMG_URL + str(diction['img_file_name']) mydata['support_type'] = "img" if ("video_path" in diction.keys()): mydata['video_path'] = str(diction['video_path']) mydata['support_type'] = "video" if ("date_publication" in diction.keys()): if diction['date_publication']: date_publication = str(diction['date_publication']).strip() local_status = mycommon.CheckisDate(date_publication) if (local_status is False): mycommon.myprint( str(inspect.stack()[0][ 3]) + " La date de publication n'est pas au format 'jj/mm/aaaa' ") return False, " La date de publication n'est pas au format 'jj/mm/aaaa' " date_str = str(date_publication)[0:10] date_format = '%d/%m/%Y' date_obj = datetime.strptime(date_str, date_format) formated_date_publication = str(date_obj.strftime("%Y-%m-%d %H:%M:%S")) mydata['date_publication'] = formated_date_publication if ("user_ip" in diction.keys()): if diction['user_ip']: mydata['user_ip'] = diction['user_ip'] if ("user_country_code" in diction.keys()): if diction['user_country_code']: mydata['user_country_code'] = diction['user_country_code'] if ("user_country_name" in diction.keys()): if diction['user_country_name']: mydata['user_country_name'] = diction['user_country_name'] if ("user_city" in diction.keys()): if diction['user_city']: mydata['user_city'] = diction['user_city'] if ("user_postal" in diction.keys()): if diction['user_postal']: mydata['user_postal'] = diction['user_postal'] if ("user_latitude" in diction.keys()): if diction['user_latitude']: mydata['user_latitude'] = diction['user_latitude'] if ("user_longitude" in diction.keys()): if diction['user_longitude']: mydata['user_longitude'] = diction['user_longitude'] if ("user_state" in diction.keys()): if diction['user_state']: mydata['user_state'] = diction['user_state'] if ("type_formation" in diction.keys()): if diction['type_formation']: mydata['type_formation'] = diction['type_formation'] if ("qualite" in diction.keys()): if diction['qualite']: mydata['qualite'] = diction['qualite'] if ("avis" in diction.keys()): if diction['avis']: mydata['avis'] = diction['avis'] if ("note" in diction.keys()): if diction['note']: mydata['note'] = diction['note'] if ("comment" in diction.keys()): if diction['comment']: mydata['comment'] = diction['comment'] if ("postedby" in diction.keys()): if diction['postedby']: mydata['postedby'] = diction['postedby'] if ("url_formation" in diction.keys()): if diction['url_formation']: mydata['url_formation'] = diction['url_formation'] if ("avisrating" in diction.keys()): if diction['avisrating']: mydata['avisrating'] = diction['avisrating'] if ("qualityrating" in diction.keys()): if diction['qualityrating']: mydata['qualityrating'] = diction['qualityrating'] meta_description = "" if ("meta_description" in diction.keys()): if diction['meta_description']: meta_description = str(diction['meta_description']).strip().lower()[0:200] else: meta_description = str(diction['title_formation']).strip().lower()[0:200] mydata['meta_description'] = meta_description a_retenir = "" if ("a_retenir" in diction.keys()): if diction['a_retenir']: a_retenir = str(diction['a_retenir']).strip().lower()[0:1000] mydata['a_retenir'] = a_retenir mydata['valide'] = '0' mydata['locked'] = '1' mydata['indexed'] = '0' mydata['indexed'] = '0' mydata['update_date'] = str(datetime.now()) mydata['isalaune'] = "0" mydata['valide'] = "1" mydata['locked'] = "0" mydata['update_date'] = str(datetime.now()) mydata['update_by'] = str(my_partner['_id']) mydata['partner_owner_recid'] = str(my_partner['recid']) print(" mydata = ", mydata) ret_val = MYSY_GV.dbname['articles_help'].find_one_and_update( {'_id': ObjectId(str(diction['_id'])), }, {"$set": mydata}, return_document=ReturnDocument.AFTER, upsert=False, ) """ On lance l'indexation de l'article """ if( ret_val): retval = ela_index_bdd_classes.ela_index_articles_help(str(ret_val['_id']), my_partner['recid']) return True, "L'article de support a été correctement mis à jour " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de mettre à jour l'article/avis" ''' Cette fonction recherche help_article ''' def recherche_help_article(diction): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['_id', 'token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'search_text'] 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, recherche annuléee") return False, " Le champ '" + val + "' n'existe pas, recherche annulée" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['search_text', 'token'] 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 la liste des arguments ") return False, " : La valeur '" + val + "' n'est pas presente dans la liste des arguments " # recuperation des paramettre search_text = "" user_recid = "" token = "" if ("search_text" in diction.keys()): if diction['search_text']: search_text = diction['search_text'] if ("token" in diction.keys()): if diction['token']: token = diction['token'] my_internal_id = '' if ("_id" in diction.keys()): if diction['_id']: my_internal_id = diction['_id'] # Verification de la validité du token/mail dans le cas des user en mode connecté if (len(str(token)) > 0): retval = mycommon.check_token_validity("", token) if retval is False: mycommon.myprint(str(inspect.stack()[0][3]) + " - La session de connexion n'est pas valide") return "Err_Connexion", "La session de connexion n'est pas valide" # Recuperation du recid de l'utilisateur user_recid = mycommon.get_user_recid_from_token(token) if user_recid is False: mycommon.myprint( str(inspect.stack()[0][3]) + " - Impossible de récupérer le token de l'utilisateur") return False, " Impossible de desactiver l'objectif utilisateur" #mycommon.myprint(" On recehrche la phrase +'" + search_text + "' + user_recid = " + user_recid) # Enregistrerment de la recherche retval, message, store_recherche_Id = store_recherche_help_article(diction, user_recid) if (retval is False): return retval, message tab_training = [] tab_training = ela_index.ela_recherche_article_help_tokens(search_text) ''' Pour analyser la recherche, decommenter les 2 lignes ci-dessous ''' mycommon.myprint(" pour phrase : #" + search_text + "#, voici la liste des articles") mycommon.myprint(tab_training) ''' la valeur nb_result permettra de savoir si la requete a donnée un resultat. si ce n'est pas le cas, il faudra à la fin enregistrer la requete avec un result a vide''' nb_result = 0 insertObject = [] for x in MYSY_GV.dbname['articles_help'].find({"external_code": {"$in": tab_training}, 'valide':'1', 'locked':'0'}, {"indexed": 0, "valide": 0, "locked": 0, }): mycommon.myprint(x) user = x insertObject.append(JSONEncoder().encode(user)) for tmp in tab_training: for x in MYSY_GV.dbname['articles_help'].find({"_id": ObjectId(tmp), 'valide':'1', 'locked':'0'}, {"indexed": 0, "valide": 0, "locked": 0, }): nb_result = nb_result + 1 #mycommon.myprint(x) user = x insertObject.append(mycommon.JSONEncoder().encode(user)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de recherche un article/avis" ''' Stockage des recherches sur les avis ''' def store_recherche_help_article(diction, user_recid=""): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' if ( len(str(user_recid)) <= 0 ): user_recid = 'None' # by default field_list = ['token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state', 'search_text', 'certif', 'support', 'type', 'lang', 'price', 'distance', 'duration'] 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, recherche annulée") return False, " Le champ '" + val + "' n'existe pas, recherche annulée", None ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['search_text', 'token'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint(" : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " : La valeur '" + val + "' n'est pas presente dans la liste des arguments ", None # recuperation des paramettre mydata = {} mydata_id = "" my_token = "" if ("search_text" in diction.keys()): if diction['search_text']: mydata['search_text'] = diction['search_text'] if ("user_mail" in diction.keys()): if diction['user_recid']: mydata['user_recid'] = diction['user_recid'] if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] '''# Recuperation du recid de l'utilisateur user_recid = mycommon.get_user_recid_from_token(my_token) if user_recid is False: mycommon.myprint(" Impossible d'enregistrer la recherche de utilisateur") return False, " Impossible d'enregistrer la recherche de utilisateur" ''' mydata['user_recid'] = user_recid mydata['document_recherche'] = "articles_help" if ("source" in diction.keys()): if diction['source']: mydata['source'] = diction['source'] if ("name" in diction.keys()): if diction['name']: mydata['name'] = diction['name'] if ("id" in diction.keys()): if diction['id']: mydata_id = diction['id'] mydata['update_date'] = str(datetime.now()) mydata['valide'] = "1" # By default if ("valide" in diction.keys()): if diction['valide']: mydata['valide'] = diction['valide'] coll_name = MYSY_GV.dbname['user_recherche'] # Si le champ "id" est renseigné, il s'agit d'une mis jour #mycommon.myprint(str(inspect.stack()[0][3]) + " on va stocker "+str(mydata)+" id = "+str(mydata_id)) if( len(str(mydata_id)) > 0 ): ret_val = coll_name.find_one_and_update({'_id': ObjectId(str(mydata_id)), 'valide': '1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) if (ret_val and ret_val['_id']): nb_doc = str(ret_val['_id']) mycommon.myprint(str(inspect.stack()[0][3]) + " La recherche a bien été mise à jour . Le Doc Id = "+str(nb_doc)) return True, "La recherche a bien été mise à jour", ret_val['_id'] else: return False, "Impossible de mettre à jour a la recherche", None else: ret_val = coll_name.insert_one(mydata) if ret_val and ret_val.inserted_id: nb_doc = ret_val.inserted_id #mycommon.myprint(str(inspect.stack()[0][3]) + " La recherche a été bien enregistrée. Id = '" + str(nb_doc)+"' ") return True, "La recherche a bien été mise à jour", nb_doc else: mycommon.myprint( str(inspect.stack()[0][3]) + "Impossible d'enregistrer la recherche") return False, "Impossible d'enregistrer la recherche ", None except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, "Impossible d'enregistrer la recherche", None '''' Cette fonction retourne la liste des help_article à la une''' def get_help_article_alaune(diction): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['_id', 'token', 'user_ip', 'user_country_code', 'user_country_name', 'user_city', 'user_postal', 'user_latitude', 'user_longitude', 'user_state'] 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, recherche annulée") return False, " Le champ '" + val + "' n'existe pas, recherche annulée" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['token'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint(" : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " : La valeur '" + val + "' n'est pas presente dans la liste des arguments ", coll_name = MYSY_GV.dbname['articles_help'] insertObject = [] for x in coll_name.find({'valide': '1', 'locked': '0', 'isalaune': '1'}, {"indexed": 0, "indexed_desc": 0, "indexed_obj": 0, "isalaune": 0, "valide": 0, "locked": 0, }).sort( [ ("date_avis", pymongo.DESCENDING), ("note", pymongo.DESCENDING)]): val = x['qualite'] if (len(x['qualite']) > MYSY_GV.MAX_CARACT_DEDUIT): x['qualite'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] + " ..." else: x['qualite'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] val = x['avis'] if (len(x['avis']) > MYSY_GV.MAX_CARACT_DEDUIT): x['avis'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] + " ..." else: x['avis'] = val[:MYSY_GV.MAX_CARACT_DEDUIT] if ("support_type" not in x.keys()): x['support_type'] = "" user = x insertObject.append(mycommon.JSONEncoder().encode(user)) #print(" getObject = ", str(insertObject)) return True, insertObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, "Impossible de récupérer les articles à la Une" """ Pour alleger le serveur, le système permettra de telecharger une image qui sera deposé sur le serveur ftp des images du blog MYSY_GV.BLOG_FTP_IMG_REPOSITORY = "/blog_img/" """ def Blog_Img_Store_Article_Image(file=None, Folder=None, diction=None): try: ''' # Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés # Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API # Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste # field_list. ''' field_list = ['token'] 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, mise à jour logo lms annulée") return False, " Les informations fournies sont incorrecte", ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['token',] 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 la liste des arguments ") return False, " Les informations fournies sont incorrecte", my_token = "" if ("token" in diction.keys()): if diction['token']: my_token = diction['token'] partner_recid = mycommon.get_parnter_recid_from_token(my_token) if (partner_recid is False): mycommon.myprint( str(inspect.stack()[0][3]) + " - partner_recid KO : Impossible d'importer la liste des participants ") return False, " Les information de connexion sont incorrectes.", # Recuperation des données du partenaire local_status, my_partner = mycommon.get_partner_data_from_recid(partner_recid) if (local_status is False): mycommon.myprint(str(inspect.stack()[0][3]) + " - impossible de récupérer les données du partenaire") return False, str(inspect.stack()[0][ 3]) + " - impossible de récupérer les données du partenaire", status, saved_file = mycommon.Upload_Save_IMG_File_No_Change_Origial_Name(file, Folder) if (status == False): return False, saved_file cnopts = pysftp.CnOpts() cnopts.hostkeys = None with pysftp.Connection(host=MYSY_GV.MYSY_FTP_HOST, username=MYSY_GV.MYSY_FTP_LOGIN, password=MYSY_GV.MYSY_FTP_PWD, cnopts=cnopts) as session: print("Connection successfully established ... ") localFilePath = saved_file remoteFilePath = str(MYSY_GV.BLOG_FTP_IMG_REPOSITORY) print(" ### DEPLACEMENT DE " + str(localFilePath) + " VERS " + str(remoteFilePath) + " AVANTTT TRAITEMENT") # Use put method to upload a file try: with session.cd(remoteFilePath): print("OK cd worked") session.put(localFilePath) except: print("NOT OK cd failed") e = sys.exc_info() print("Exception: {0}".format(e)) return True, " Le fichier a été correctement mis à jour" 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 mettre à jour le fichier "