import ast import ssl from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import pymongo import zeep from pymongo import MongoClient import json from bson import ObjectId import re from datetime import datetime import prj_common as mycommon import secrets import inspect import sys, os import csv import pandas as pd from pymongo import ReturnDocument from math import isnan import GlobalVariable as MYSY_GV import ela_index_bdd_classes as eibdd import email_mgt as email import urllib.request from validate_email import validate_email import jinja2 """ Cette fonction monitore un siteweb. si Ko, alors il envoie un email à cherif. c'est une version degradée """ def check_if_site_is_up(diction): try: print(' ### diction = ', diction) URL = diction['url'] site_status = False try: response = requests.head(URL) except Exception as e: print(f"NOT OK: {str(e)}") site_status = False else: if response.status_code == 200: print("OK") site_status = True else: print(f"NOT OK: HTTP response code {response.status_code}") site_status = False if( site_status is True): return site_status, " Site OKKK" else: """ Envoyer un email """ html = '''

Votre Site mysy-training.com est KO

Bonjour,
Merci de verifier l'état de votre site MySy-Traning.com
''' html_mime = MIMEText(html, 'html') msg = MIMEMultipart("alternative") smtpserver = smtplib.SMTP(MYSY_GV.O365_SMTP_COUNT_smtpsrv, MYSY_GV.O365_SMTP_COUNT_port) msg.attach(html_mime) msg['From'] = MYSY_GV.O365_SMTP_COUNT_From_User msg['Bcc'] = 'contact@mysy-training.com' msg['Subject'] = " IMPORTANT : Votre site mysy-training.com est KO" msg['To'] = "cherif.balde@yahoo.fr, cbalde@mysy-training.com" smtpserver.ehlo() smtpserver.starttls() smtpserver.login(MYSY_GV.O365_SMTP_COUNT_user, MYSY_GV.O365_SMTP_COUNT_password) val = smtpserver.send_message(msg) smtpserver.close() print(" Email envoyé " + str(val)) html_mime = MIMEText(html, 'html') return site_status, " Site KOOO" 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 récupérer la formation" def email_validation(): try: coll_emails = MYSY_GV.dbname['bdd_email'] cpt = 0 nb_doc = coll_emails.count_documents({'valide':'1', "checked": {"$ne": "1"}}) print(" ### nb_doc = ", nb_doc) for val in coll_emails.find({'valide':'1', "checked": {"$ne": "1"}}): cpt = cpt + 1 if ("email_address" in val.keys()): is_valid = validate_email( email_address=val['email_address'], check_format=True, check_blacklist=True, check_dns=True, dns_timeout=10, check_smtp=True, smtp_timeout=10, smtp_helo_host='smtp.office365.com', smtp_from_address='support@mysy-training.com', smtp_skip_tls=False, smtp_tls_context=None, smtp_debug=False) retval = "" checked_status = "" if( is_valid ): #print(" ### l'adresse ",val['email_address']," est valide ") retval = True checked_status = "1" else: #print(" ### l'adresse ", val['email_address'], " N'EST PAS valide ") retval = False checked_status = "0" mydata = {} mytoday = datetime.today().strftime("%d/%m/%Y") mydata['last_check_date'] = str(mytoday)[0:10] mydata['checked'] = str("1") mydata['checked_status'] = str(checked_status) ret_val = coll_emails.find_one_and_update( {'email_address': str(val['email_address']), 'valide': '1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) print(" ### "+str(cpt)+" email traité sur "+str(nb_doc)) return retval, str(cpt)+" adresse email ont été mise à 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 récupérer la formation" def email_validation_mairies(): try: coll_emails = MYSY_GV.dbname['collect_list_mairies'] cpt = 0 nb_doc = coll_emails.count_documents({'valide': '1', "email_checked": {"$ne": "1"}}) print(" ### nb_doc = ", nb_doc) for val in coll_emails.find({'valide': '1', "checked": {"$ne": "1"}}): cpt = cpt + 1 if ("email" in val.keys()): is_valid = validate_email( email_address=str(val['email']).strip(), check_format=True, check_blacklist=True, check_dns=True, dns_timeout=10, check_smtp=True, smtp_timeout=10, smtp_helo_host='smtp.office365.com', smtp_from_address='support@mysy-training.com', smtp_skip_tls=False, smtp_tls_context=None, smtp_debug=False) retval = "" checked_status = "" if (is_valid): # print(" ### l'adresse ",val['email_address']," est valide ") retval = True checked_status = "1" else: # print(" ### l'adresse ", val['email_address'], " N'EST PAS valide ") retval = False checked_status = "0" mydata = {} mytoday = datetime.today().strftime("%d/%m/%Y") mydata['last_check_date'] = str(mytoday)[0:10] mydata['email_checked'] = str("1") mydata['email_checked_status'] = str(checked_status) ret_val = coll_emails.find_one_and_update( {'email': str(val['email']), 'valide': '1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) print(" ### " + str(cpt) + " / " + str(nb_doc)) return retval, str(cpt) + " adresse email ont été mise à 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 récupérer la formation" def email_validation_hopitaux_ehpad(): try: coll_emails = MYSY_GV.dbname['collect_list_hopitaux_ehpad'] cpt = 0 nb_doc = coll_emails.count_documents({'valide': '1', "email_checked": {"$ne": "1"}}) print(" ### nb_doc = ", nb_doc) for val in coll_emails.find({'valide': '1', "checked": {"$ne": "1"}}): cpt = cpt + 1 if ("email" in val.keys()): is_valid = validate_email( email_address=str(val['email']).strip(), check_format=True, check_blacklist=True, check_dns=True, dns_timeout=10, check_smtp=True, smtp_timeout=10, smtp_helo_host='smtp.office365.com', smtp_from_address='support@mysy-training.com', smtp_skip_tls=False, smtp_tls_context=None, smtp_debug=False) retval = "" checked_status = "" if (is_valid): # print(" ### l'adresse ",val['email_address']," est valide ") retval = True checked_status = "1" else: # print(" ### l'adresse ", val['email_address'], " N'EST PAS valide ") retval = False checked_status = "0" mydata = {} mytoday = datetime.today().strftime("%d/%m/%Y") mydata['last_check_date'] = str(mytoday)[0:10] mydata['email_checked'] = str("1") mydata['email_checked_status'] = str(checked_status) ret_val = coll_emails.find_one_and_update( {'email': str(val['email']), 'valide': '1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) print(" ### " + str(cpt) + " / " + str(nb_doc)) return retval, str(cpt) + " adresse email ont été mise à 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 récupérer la formation" def email_validation_list_pos_bio(): try: coll_emails = MYSY_GV.dbname['collect_list_pos_bio'] cpt = 0 nb_doc = coll_emails.count_documents({'valide': '1', "email_checked": {"$ne": "1"}}) print(" ### nb_doc = ", nb_doc) for val in coll_emails.find({'valide': '1', "checked": {"$ne": "1"}}): cpt = cpt + 1 if ("email" in val.keys()): is_valid = validate_email( email_address=str(val['email']).strip(), check_format=True, check_blacklist=True, check_dns=True, dns_timeout=10, check_smtp=True, smtp_timeout=10, smtp_helo_host='smtp.office365.com', smtp_from_address='support@mysy-training.com', smtp_skip_tls=False, smtp_tls_context=None, smtp_debug=False) retval = "" checked_status = "" if (is_valid): # print(" ### l'adresse ",val['email_address']," est valide ") retval = True checked_status = "1" else: # print(" ### l'adresse ", val['email_address'], " N'EST PAS valide ") retval = False checked_status = "0" mydata = {} mytoday = datetime.today().strftime("%d/%m/%Y") mydata['last_check_date'] = str(mytoday)[0:10] mydata['email_checked'] = str("1") mydata['email_checked_status'] = str(checked_status) ret_val = coll_emails.find_one_and_update( {'email': str(val['email']), 'valide': '1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER ) print(" ### " + str(cpt) + " / " + str(nb_doc)) return retval, str(cpt) + " adresse email ont été mise à 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 récupérer la formation" import socket import smtplib import re import dns.resolver def test_tab_mail(tab_mail): try: email_address = 'emildqsdssdqdsdie.dacossqqazta@3ds.com' # Step 1: Check email # Check using Regex that an email meets minimum requirements, throw an error if not addressToVerify = email_address match = re.match('^[_a-z0-9-]+(\.[_a-z0-9-]+)*@[a-z0-9-]+(\.[a-z0-9-]+)*(\.[a-z]{2,4})$', addressToVerify) if match == None: print('Bad Syntax in ' + addressToVerify) raise ValueError('Bad Syntax') # Step 2: Getting MX record # Pull domain name from email address domain_name = email_address.split('@')[1] # get the MX record for the domain my_resolver = dns.resolver.Resolver() records = my_resolver.resolve(domain_name, 'MX') mxRecord = records[0].exchange mxRecord = str(mxRecord) print(mxRecord) # Step 3: ping email server # check if the email address exists # Get local server hostname host = socket.gethostname() # SMTP lib setup (use debug level for full output) server = smtplib.SMTP() server.set_debuglevel(0) # SMTP Conversation server.connect(mxRecord) server.helo(host) server.mail('cbalde@mysy-training.com') code, message = server.rcpt(str(addressToVerify)) server.quit() # Assume 250 as Success if code == 250: print('Yes') return True, "OK" else: print('N') return False, "KO" return list_mail = "" if ("list_mail" in tab_mail.keys()): list_mail = str(tab_mail['list_mail']) cpt = 0 val = list_mail print(" list_mail = ", list_mail) is_valid = validate_email( email_address=str(val), check_format=True, check_blacklist=True, check_dns=True, dns_timeout=100, check_smtp=True, smtp_timeout=100, smtp_helo_host='smtp.office365.com', smtp_from_address='support@mysy-training.com', smtp_skip_tls=False, smtp_tls_context=None, smtp_debug=False) retval = "" checked_status = "" if (is_valid): print(" ### l'adresse ",val," est valide ") retval = True else: print(" ### l'adresse ", val, " N'EST PAS valide ") retval = False return retval, str(cpt) + " adresse email ont été mise à 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 tester le mail" ''' dirty function ''' def dirty(): try: tab = ['a.maisonneuve@ch-montlucon.fr', 'a.notteghem@ch-stquentin.fr', 'adamasambou@live.fr', 'as.aubert@ch-cannes.fr', 'bduquai@gmail.com', 'berdaguerd@gmail.com', 'bnoetinger@yahoo.fr', 'boucherie.gervais@gmail.com', 'boucherie.laiguillon.lh@gmail.com', 'boucherie.piquenot@gmail.com', 'boucheriebeauvais@gmail.com', 'boucheriefouache@gmail.com', 'boucheriesaintvivien@gmail.com', 'breantbaba@gmail.com', 'c.hugues@ch-stquentin.fr', 'canals.cedric@orange.fr', 'caterinallende@yahoo.es', 'catherine.hardy@ch-ariege-couserans.fr', 'cathysalva@live.fr', 'celine.viteau@ch-soissons.fr', 'charcuterieleger@orange.fr', 'charles@closdesvinsdamour.fr', 'chateaunadalhainaut@gmail.com', 'clotdelorigine@gmail.com', 'contact@bioteafull.fr', 'contact@domainepouderoux.fr', 'contact@herboristerie-moderne.fr', 'contact@lesjardinsdemeraude.fr', 'contact@masbecha.com', 'contact@masdenfelix.com ', 'contact@vins-face-b.fr', 'cyril.marais@ch-soissons.fr', 'cyrilfhal@gmail.com', 'daniel.leclerc@gmail.com', 'direction.ehpad.aixeo@hotmail.fr', 'direction.ehpad.rpc@gmail.com', 'direction@ch-hirson.fr', 'direction@chsi-ainay.fr', 'direction@hi-bsav.fr', 'directiongenerale@ch-stquentin.fr', 'dom.ponsgralet@wanadoo.fr', 'domaine.laguerre@orange.fr', 'domainespiaggia@yahoo.com', 'domitienne@outlook.com', 'drh@cas-forcalquier.fr', 'ducrocq.quentin29@gmail.com', 'ehpad.grdbosquet.villerscotterets@wanadoo.fr', 'ehpad@beauregard-residence.fr', 'enidnamag@hotmail.com', 'eric.heyrman@ch-soissons.fr', 'eric.robart@orange.fr', 'etoiledelabergere486@gmail.com', 'f.mezrouh@ch-stquentin.fr', 'flo.moussellous@gmail.com', 'foyer.occ@wanadoo.fr', 'francoise.moreau@hopcobour.net', 'gdesnoix@chsi-ainay.fr', 'helie.boucherie@orange.fr', 'herve.bessiere@gmail.com', 'hopital.saintmaur@gmail.com', 'info.domainegardies@gmail.com', 'info@masllossanes.fr', 'inge.meierhofer@yahoo.fr', 'j.louisy@ch-stquentin.fr', 'jeanmarie.colin@ch-valvert.fr', 'jfdeu@hotmail.com', 'jp.fardeau@cas-forcalquier.fr', 'labergeriedesabeilles@gmail.com', 'lafermedubiosillon@gmail.com', 'lamerblanche66@gmail.com', 'laurent.lemoux@ch-ghsa.fr', 'laurie@lejardinbiodelaurie.fr ', 'lechantdelaterre66@gmail.com', 'lemasaintantoine@gmail.com', 'lepetitsoleilbio66@outlook.fr', 'les.salicaires.vigneron@gmail.com', 'letempsbio@gmail.com', 'lherberie@protonmail.com', 'ljdt66@gmail.com', 'm.saf@ch-guise.fr', 'marc.veuillet@cgd13.fr', 'masdumoutonnoir@gmail.com', 'maugerdominique@hotmail.com', 'mylene.verdu@gmail.com ', 'n.estin@ch-stquentin.fr', 'naliane@chsi-ainay.fr', 'o.ponties@ch-rodez.fr', 'pierre.espejo2@gmail.com', 'pouletbio@live.fr', 'responsable.rh-log@ehpad-mgasquet.fr', 'romain.cochonbio@yahoo.com', 's.bousmaha@ch-stquentin.fr', 's.junker@ch-montlucon.fr', 'sarlavice76@gmail.com', 'sarlmaisoncuvier@gmail.com', 'scealaroqueta@gmail.com', 'sophie.barbier@ch-ghsa.fr', 'thierry.levionnois@orange.fr', 'veroniquedumont42@gmail.com', 'yanisleroux@orange.fr', 'andre.minyemeck@ehpad-conches.fr', 'anne.quinville@armorsante.bzh', 'audrey.licandro@chu-dijon.fr', 'c.bistue@chuzes.fr', 'ca.doussot@cht-ranceemeraude.fr', 'catherine.lahille@gh-portesdeprovence.fr', 'cecile.chalet@chu-nimes.fr', 'cgreslon@ch-bassindethau.fr', 'christian.soubie@ch-libourne.fr', 'contact-stp@hopitaloleron.net', 'contact@ch-edouard-toulouse.fr', 'contact@hopital-vicfezensac.fr', 'contact@nh-navarre.fr', 'd.faivre@chi-hc.fr', 'david.trouchaud@ght-cdn.fr', 'dg.secretariat@ch-perigueux.fr', 'diandra.tijjini@nh-navarre.fr', 'directeur@ch-saint-renan.fr', 'directeur@hopital-murat.fr', 'direction.generale@ch-perrens.fr', 'direction@ch-claudel.fr', 'direction@ch-condom.com', 'direction@gh-portesdeprovence.fr', 'direction@mr-blamont.fr', 'directionfdcmontpon@orange.fr', 'direhpadsudcher@gmail.com', 'e-barde@chu-montpellier.fr', 'ehpad.gracay@orange.fr', 'ehpad.lebrestalou@wanadoo.fr', 'ehpad.lescedres@outlook.fr', 'florie.bideplan@ch-libourne.fr', 'helene.normand@nh-navarre.fr', 'ifsivierzon@ch-vierzon.fr', 'j-lepage@chu-montpellier.fr', 'jean-baptiste.fleury@armorsante.bzh', 'l.brule@ch-hdn.fr', 'lalliot-bironneau@ch-perrens.fr', 'logistique@hlv.fr', 'logistiqueaudincourt@gmail.com', 'magali.luc@chu-nimes.fr', 'maison-retraite@mr-blamont.fr', 'maisonderetraite.montpon@orange.fr', 'marc.jaffuer@ch-libourne.fr', 'mr.la.guerche@wanadoo.fr', 'mrlaignes@wanadoo.fr', 'p.perrot@ch-stmalo.fr', 'personnel@epms-orbec.fr', 'philippe.charre@gh-portesdeprovence.fr', 'ronan.sanquer@chu-brest.fr', 'sdeduit@ch-vierzon.fr', 'sylvie.briend@ch-dinan.fr', 't-veleine@chu-montpellier.fr', 'tyandudcoz@orange.fr'] coll_bio = MYSY_GV.dbname['collect_list_pos_bio'] coll_ehpad = MYSY_GV.dbname['collect_list_hopitaux_ehpad'] coll_boucherie = MYSY_GV.dbname['collect_list_boucherie'] for val in tab : for ret1 in coll_bio.find({'email':val}): if ("email" in ret1.keys()): print( "'Point de vente BIO', '",ret1['email'], "' => '",ret1['raison_sociale'], "' => '", ret1['adress'], "'=> '", ret1['tel']+"'") for ret2 in coll_ehpad.find({'email':val}): if ("email" in ret2.keys()): print( "'EHPAD & HOPITAUX', '",ret2['email'], "'=> '",ret2['raison_sociale'], "' => '", ret2['adresse'], "'=>' ", ret2['tel']+"'") for ret3 in coll_boucherie.find({'email':val}): if ("email" in ret3.keys()): print("'BOUCHERIE', '", ret3['email'], "'=> '",ret3['raison_sociale'], "' => '---' => '", ret3['tel']+"'") return True, "ok" 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 tester le mail" # Root de chamilo url_root = 'https://online.educetera.org/' # Conexión from zeep import Client #server = SOAPProxy(url_root+'/main/webservices/soap.php' ) import requests import urllib.request import hashlib """ url = "https://lms.mysy-training.com/main/webservices/soap.php" querystring = {"wsdl":""} payload = "\n\n \n \n ---\n ---\n \n \n \n \n NY\n \n \n" headers = { 'content-type': "text/xml; charset=utf-8", 'soapaction': 'https://lms.mysy-training.com/main/webservices/soap.php/WSUser.DisableUser' } response = requests.request("POST", url, data=payload, headers=headers, params=querystring) print(response.text) """ def mysysoap_old(): try: url = "https://lms.mysy-training.com/main/webservices/soap.php" querystring = {"wsdl": ""} url_root = "http://lms.mysy-training.com/" ewurl = "https://lms.mysy-training.com/main/webservices/soap.php" full_url = "https://lms.mysy-training.com/main/webservices/testip.php" url = "https://lms.mysy-training.com/main/webservices/soap.php/WS.test" security_key = 'da6e9548ebb7137b97913ae7589c85c5' # Se encuentra en el archivo main/inc/conf/configuration.php línea 115 my_ip = requests.get(full_url).text # print(" my is = ", my_ip) # "'#readlines()[0][:-1] secret_key = hashlib.sha1(str(my_ip + security_key).encode('utf-8')).hexdigest() print(" secret_key = ", secret_key) # "'#readlines()[0][:-1] url = 'http://www.dneonline.com/calculator.asmx' xml = """ 50 20 """ headers = {'content-type': 'application/soap+xml; charset=utf-8'} #r1 = requests.post(url, data=xml, headers=headers) url = 'https://lms.mysy-training.com/main/webservices/soap.php' xml = """ """ headers = {'content-type': 'application/soap+xml; charset=utf-8'} r1 = requests.post(url, data=xml, headers=headers) print(r1.text) return True, "ok" 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 tester le mail" from zeep import Client, Settings def mysysoap(): try: """wsdl = "http://www.dneonline.com/calculator.asmx?WSDL" client = Client(wsdl) val = client.service.Add(3, 5) print(val) """ security_key = 'da6e9548ebb7137b97913ae7589c85c5' # Se encuentra en el archivo main/inc/conf/configuration.php línea 115 full_url = "https://lms.mysy-training.com/main/webservices/testip.php" my_ip = requests.get(full_url).text # print(" my is = ", my_ip) # "'#readlines()[0][:-1] secret_key = hashlib.sha1(str(my_ip + security_key).encode('utf-8')).hexdigest() print(" secret_key = ", secret_key) # "'#readlines()[0][:-1] url = "http://wsf.cdyne.com/WeatherWS/Weather.asmx?WSDL" url="https://lms.mysy-training.com/main/webservices/soap.php?wsdl" headers = {'content-type': 'text/xml'} body = """ 135b6ca3b7ed29fa787ff011859fd76ec5e1b51c ['1504'] """ response = requests.post(url, data=body, headers=headers) print(response.content) val = "grr" return True, "ok val = "+str(response.content) 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 tester le mail" # Connexion mariadb sur serveur import mariadb import hashlib import bcrypt def mysylmsdb(): try: # Rcuperation de la sequence de l'objet "lms_user_id" dans la collection : "mysy_sequence" retval = MYSY_GV.dbname['mysy_sequence'].find_one({'related_mysy_object':'lms_user_id', 'valide':'1'}) print(" ### retval = ",retval) mypasswd = 'sekou'.encode('UTF-8') salt = bcrypt.gensalt() hashed_pwd = bcrypt.hashpw(mypasswd, salt) salt = salt.encode('UTF-8') hashed_pwd = hashed_pwd.encode('UTF-8') print(" ## HASHED hashed = ", str(hashed_pwd), " SALT = , ", str(salt)) if (retval is None or "current_val" not in retval.keys()): mycommon.myprint(" Impossible de récupérer la sequence 'mysy_sequence' ") return False, "Impossible de récupérer la sequence 'mysy_sequence'" new_lms_user_id = str(retval['prefixe'])+str(retval['current_val']) local_status, local_val = mycommon.IsInt(new_lms_user_id) if( local_status is False): mycommon.myprint(" Impossible de generer l'ID du user LMS. La valeur actuelle est ", new_lms_user_id) return False, " Impossible de generer l'ID du user LMS" current_seq_value = str(retval['current_val']) conn = mariadb.connect( user=MYSY_GV.MYSY_MARIADB_USER, password=MYSY_GV.MYSY_MARIADB_USER_PASS, host=MYSY_GV.MYSY_MARIADB_HOST, port=MYSY_GV.MYSY_MARIADB_PORT, database=MYSY_GV.MYSY_LMS_BDD ) cur = conn.cursor() # retrieving information cur.execute("SELECT username, email_canonical FROM user WHERE id=?", (2,)) for username, email_canonical in cur: print(f"username: {username}, email_canonical : {email_canonical}") # insert information my_query = "INSERT INTO user SET id='"+str(new_lms_user_id)+"' , " \ "user_id='"+str(new_lms_user_id)+"' , " \ "lastname = 'carter', " \ "firstname = 'vince_022', " \ "username = 'username_1022', " \ "salt = '"+str(salt)+"'," \ " registration_date = '2022-04-17 15:34:43'," \ " credentials_expired='0', " \ "enabled='1', " \ "expired='0', " \ " status = '1', " \ "password = '"+str(hashed_pwd)+"', " \ "locked = '0', " \ "username_canonical = 'username_1022', " \ " email_canonical = 'username@mano2.fr', " \ "email = 'username@mano2.fr'," \ " official_code = 'AAA', " \ "creator_id = '1', " \ "auth_source = 'platform'," \ "roles = 'a:0:{}', " \ "language = 'french'," \ "active = '1';" try: #cur.execute("INSERT INTO employees (first_name,last_name) VALUES (?, ?)", ("Maria", "DB")) cur.execute(my_query) except mariadb.Error as e: print(f"Error: {e}") mycommon.myprint(" Impossible de créer l'utilisateur LMS "+str({e})) return False, "Impossible de créer l'utilisateur LMS "+str({e}) conn.commit() print(f"Last Inserted ID: {cur.lastrowid}") conn.close() # Crementation de la sequence de l'objet "lms_user_id" dans la collection : "mysy_sequence" new_sequence_value = int(current_seq_value)+1 mydata = {'current_val':new_sequence_value} ret_val2 = MYSY_GV.dbname['mysy_sequence'].find_one_and_update({'related_mysy_object':'lms_user_id', 'valide':'1'}, {"$set": mydata}, return_document=ReturnDocument.AFTER, upsert= False, ) return True, "Connexion mariadb OKK" 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, " Connexion mariadb KOOO" from datetime import timedelta def test_web_service(diction): try: diction = mycommon.strip_dictionary(diction) C1_pipe_qry = ([ {'$match': {'$and': [ {'valide':'1'}, {'automatic_traitement.actif':'1'}, ]} }, ]) for retval in MYSY_GV.dbname['session_formation'].aggregate(C1_pipe_qry) : print( " ### retval == ", retval) return True, "Test WebService OK Diction " 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, "Test WebService KOOO" """ Cette fonction prend un dictionnaire qui rattache chaque evenement (agenda : type planning) au contrat en cours (+les données de l'employé) et retourne une dictionnaire reformaté de manière à avoir une ligne par evenement + toutes les data du contrat en face """ def format_emplee_contrat_agenda_events(tab_diction): try: return_tab = [] for val in tab_diction: print(" #### Debut traitement de la ligne "+str(val) ) for event_data in val['Tab_agenda'] : new_node = {} new_node['rh_id'] = val['_id'] new_node['rh_partner_recid'] = val['partner_recid'] new_node['rh_civilite'] = val['civilite'] new_node['rh_nom'] = val['nom'] new_node['rh_prenom'] = val['prenom'] new_node['rh_email'] = val['email'] new_node['rh_telephone_mobile'] = val['telephone_mobile'] new_node['rh_ismanager'] = val['ismanager'] new_node['rh_fonction'] = val['fonction'] new_node['rh_telephone'] = val['telephone'] new_node['rh_purchase_price_group_id'] = val['purchase_price_group_id'] new_node['rh_date_naissance'] = val['date_naissance'] new_node['rh_adr_adresse'] = val['adr_adresse'] new_node['rh_adr_code_postal'] = val['adr_code_postal'] new_node['rh_adr_ville'] = val['adr_ville'] new_node['rh_adr_pays'] = val['adr_pays'] if ("collection_ressource_humaine_contrat" in val.keys()): if( "date_debut" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_date_debut'] = val['collection_ressource_humaine_contrat']['date_debut'] else: new_node['rh_contrat_date_debut'] = "" if ("date_fin" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_date_fin'] = val['collection_ressource_humaine_contrat']['date_fin'] else: new_node['rh_contrat_date_fin'] = "" if ("type_contrat" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_type_contrat'] = val['collection_ressource_humaine_contrat']['type_contrat'] else: new_node['rh_contrat_type_contrat'] = "" if( "type_employe" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_type_employe'] = val['collection_ressource_humaine_contrat']['type_employe'] else: new_node['rh_contrat_type_employe'] = "" if( "cout" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_cout'] = val['collection_ressource_humaine_contrat']['cout'] else: new_node['rh_contrat_cout'] = "" if( "periodicite" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_periodicite'] = val['collection_ressource_humaine_contrat']['periodicite'] else: new_node['rh_contrat_periodicite'] = "" if( "quantite" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_quantite'] = val['collection_ressource_humaine_contrat']['quantite'] else: new_node['rh_contrat_quantite'] = "" if( "groupe_prix_achat_id" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_groupe_prix_achat_id'] = val['collection_ressource_humaine_contrat']['groupe_prix_achat_id'] else: new_node['rh_contrat_groupe_prix_achat_id'] = "" if( "comment" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_comment'] = val['collection_ressource_humaine_contrat']['comment'] else: new_node['rh_contrat_comment'] = "" if( "groupe_prix_achat_cout" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_groupe_prix_achat_cout'] = val['collection_ressource_humaine_contrat']['groupe_prix_achat_cout'] else: new_node['rh_contrat_groupe_prix_achat_cout'] = "" if( "groupe_prix_achat_periodicite" in val['collection_ressource_humaine_contrat'].keys()): new_node['rh_contrat_groupe_prix_achat_periodicite'] = val['collection_ressource_humaine_contrat']['groupe_prix_achat_periodicite'] else: new_node['rh_contrat_groupe_prix_achat_periodicite'] = "" # Traitement des données de l'evement dans l'agenda new_node['rh_event_planning_event_type'] = event_data['event_type'] new_node['rh_event_planning_event_title'] = event_data['event_title'] new_node['rh_event_planning_event_start'] = event_data['event_start'] new_node['rh_event_planning_event_end'] = event_data['event_end'] new_node['rh_event_planning_agenda_date_jour'] = event_data['agenda_date_jour'] new_node['rh_event_planning_event_duration_hour'] = event_data['event_duration_hour'] new_node['rh_event_planning_event_duration_second'] = event_data['event_duration_second'] if( "comment" in event_data.keys()): new_node['rh_event_planning_even_comment'] = event_data['comment'] else: new_node['rh_event_planning_even_comment'] = "" """ # Calcul du cout total d'une tache /!\ Le prix du contrat prend toujours le pas sur le prix rentré à la main. C'est a dire que si un pour quelque raison que ce soit, sur la même période un employé a un groupe de prix d'achat et une prix saisie à la main, alors les totaux seront calculés en se basant sur le prix qui est dans le contrat """ if( "rh_contrat_groupe_prix_achat_id" in new_node.keys() and new_node['rh_contrat_groupe_prix_achat_id'] and "rh_contrat_groupe_prix_achat_periodicite" in new_node.keys() and new_node['rh_contrat_groupe_prix_achat_periodicite'] and "rh_event_planning_event_duration_hour" in new_node.keys() and new_node['rh_event_planning_event_duration_hour'] and "rh_contrat_groupe_prix_achat_cout" in new_node.keys() and new_node['rh_contrat_groupe_prix_achat_cout'] ): if( str(new_node['rh_contrat_groupe_prix_achat_periodicite']) == "heure" ): new_node['rh_event_planning_event_cost'] = str(mycommon.tryFloat(new_node['rh_contrat_groupe_prix_achat_cout'] ) * mycommon.tryFloat(new_node['rh_event_planning_event_duration_hour'] ) ) elif ( str(new_node['rh_contrat_groupe_prix_achat_periodicite']) == "fixe" ): new_node['rh_event_planning_event_cost'] = str((new_node['rh_contrat_groupe_prix_achat_cout']) ) elif ( "rh_contrat_groupe_prix_achat_id" in new_node.keys() and str(new_node['rh_contrat_groupe_prix_achat_id']) == "" and "rh_contrat_periodicite" in new_node.keys() and new_node['rh_contrat_periodicite'] and "rh_event_planning_event_duration_hour" in new_node.keys() and new_node['rh_event_planning_event_duration_hour'] and "rh_contrat_cout" in new_node.keys() and new_node['rh_contrat_cout']): if (str(new_node['rh_contrat_periodicite']) == "heure"): new_node['rh_event_planning_event_cost'] = str( mycommon.tryFloat(new_node['rh_contrat_cout']) * mycommon.tryFloat(new_node['rh_event_planning_event_duration_hour'])) elif (str(new_node['rh_contrat_groupe_prix_achat_periodicite']) == "fixe"): new_node['rh_event_planning_event_cost'] = str((new_node['rh_contrat_cout'])) else: new_node['rh_event_planning_event_cost'] = "Impossible a calculer" return_tab.append(new_node) return True, return_tab 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, "Test WebService KOOO" def flatten_data(y): out = {} def flatten(x, name=''): if type(x) is dict: for a in x: flatten(x[a], name + a + '_') elif type(x) is list: i = 0 for a in x: flatten(a, name + str(i) + '_') i += 1 else: out[name[:-1]] = x flatten(y) return out def test_web_service2(diction): try: print(" #### ENVIRONNEMENT = "+str(MYSY_GV.MYSY_ENV)+" ") if( diction ): print(" ### diction dans la fonction = ", diction) else: print(" ### diction dans la fonction est vide") return True, "ENV = "+str(MYSY_GV.MYSY_ENV)+" Test WebService OK Diction = "+str(diction) 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, "Test WebService KOOO" """ Cette fonction permet de tester l'utilisation de jinja2 avec des informations stockées dans une base de données """ def test_jnja2_database(diction): try: # Recuperation des information du template Document_Template_Data = MYSY_GV.dbname['courrier_template'].find_one({'ref_interne':'test01'}) if( Document_Template_Data is None): mycommon.myprint(str(inspect.stack()[0][3]) + " - Aucun template avec 'ref_interne':'test01' ") #return False, " - Aucun template avec 'ref_interne':'test01' " elif( "type_doc" in Document_Template_Data.keys()): print(" ### type_doc = ", str(Document_Template_Data['type_doc'])) tm = jinja2.Template("My name is {{ per.name }} and I am {{ per.age }}") person = {'name': 'Cherif BALDE', 'age': 41} msg = tm.render(per=person) print(" ### MSG APRES JINJA2 = ",msg ) return True, "test_jnja2_database => OK : MESSAGE = "+str(msg) 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, "test_jnja2_database KOOO" """ Cette fonction retourne la liste des champs (+ collection) utilisable dans la personnalisation d'un document """ def Get_Personnalisable_Collection_Fields(): try: RetObject = [] mycollection_field1 = {"collection_technical_name":'inscription', "collection_fonctional_name":'Inscription', "field_technical_name":'session_id', "field_fonctional_name":'Session Code'} mycollection_field2 = {"collection_technical_name":'inscription', "collection_fonctional_name":'Inscription', "field_technical_name": 'class_internal_url', "field_fonctional_name": 'Title Formation'} mycollection_field3 = {"collection_technical_name":'inscription', "collection_fonctional_name":'Inscription', "field_technical_name": 'date_du', "field_fonctional_name": 'Date debut'} mycollection_field4 = {"collection_technical_name":'inscription', "collection_fonctional_name":'Inscription', "field_technical_name": 'date_au', "field_fonctional_name": 'Date Fin'} RetObject.append(mycommon.JSONEncoder().encode(mycollection_field1)) RetObject.append(mycommon.JSONEncoder().encode(mycollection_field2)) RetObject.append(mycommon.JSONEncoder().encode(mycollection_field3)) RetObject.append(mycommon.JSONEncoder().encode(mycollection_field4)) mycollection2_field1 = {"collection_technical_name": 'myclass', "collection_fonctional_name": 'Formation', "field_technical_name": 'title', "field_fonctional_name": 'Titre'} mycollection2_field2 = {"collection_technical_name": 'myclass', "collection_fonctional_name": 'Formation',"field_technical_name": 'plus_produit', "field_fonctional_name": 'Avantage'} mycollection2_field3 = {"collection_technical_name": 'myclass', "collection_fonctional_name": 'Formation',"field_technical_name": 'objectif', "field_fonctional_name": 'Objectif'} RetObject.append(mycommon.JSONEncoder().encode(mycollection2_field1)) RetObject.append(mycommon.JSONEncoder().encode(mycollection2_field2)) RetObject.append(mycommon.JSONEncoder().encode(mycollection2_field3)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste collections personnalisables" """ Fonction retourne la liste des collection eligibles à la personnalisation des document """ def Get_Personnalisable_Collection(): try: RetObject = [] mycollection_field1 = {"collection_technical_name": 'inscription', "collection_fonctional_name": 'Inscription',} mycollection_field2 = {"collection_technical_name": 'myclass', "collection_fonctional_name": 'Formation',} RetObject.append(mycommon.JSONEncoder().encode(mycollection_field1)) RetObject.append(mycommon.JSONEncoder().encode(mycollection_field2)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de récupérer la liste collections personnalisables" """ Chargement d'une collection d'une base de données vers une autre base de données """ def Delete_And_Load_DB_Collection_to_DB_Collection_For_Demo_Env(): try: #REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] """ Ici les collections avec les filtres pr chaque collection """ collection_with_filter = [ {'collection': 'contrat_type', 'filter': {'partner_owner_recid': 'default'}}, {'collection':'base_config_dashbord', 'filter':{'partner_owner_recid':'default'}}, {'collection': 'base_config_modele_journee', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'base_partner_session_step', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'base_partner_setup', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'class_niveau_formation', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'courrier_template', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'courrier_template_champ_config', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'mysy_sequence', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'application_modules', 'filter': {}}, {'collection': 'mysystopwords', 'filter': {}}, {'collection': 'pack', 'filter': {}}, {'collection': 'products', 'filter': {}}, {'collection': 'produit_service', 'filter': {}}, {'collection': 'ressource_humaine_profil', 'filter': {}}, {'collection': 'search_suggestion_words', 'filter': {}}, {'collection': 'ville_commune', 'filter': {}}, {'collection': 'word_not_stem', 'filter': {}}, {'collection': 'list_mots_fr', 'filter': {}}, {'collection': 'list_mots_not_fr', 'filter': {}}, {'collection': 'liste_domaine_metier', 'filter': {}}, {'collection': 'liste_opco', 'filter': {}}, {'collection': 'lms_theme', 'filter': {}}, ] ## On vide les table d'abord dans l'environnement cible for tmp in collection_with_filter: collection = tmp['collection'] DEM_dbname[str(collection)].delete_many({}) for tmp in collection_with_filter: collection = tmp['collection'] filter = tmp['filter'] for val in REC_dbname[collection].find(filter): print(" ## val = ", val) del val['_id'] print(" ### "+str(collection)+" : "+str(val)) DEM_dbname[str(collection)].insert_one(val) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_And_Load_DB_Collection_to_DB_Collection_For_Demo_Env" def Delete_And_Load_DB_Collection_to_DB_Collection_For_PROD_Env(): try: #REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] PROD_dbname = client['cherifdb'] """ Ici les collections avec les filtres pr chaque collection """ collection_with_filter = [ {'collection': 'contrat_type', 'filter': {'partner_owner_recid': 'default'}}, {'collection':'base_config_dashbord', 'filter':{'partner_owner_recid':'default'}}, {'collection': 'base_config_modele_journee', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'base_partner_session_step', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'base_partner_setup', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'class_niveau_formation', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'courrier_template', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'courrier_template_champ_config', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'mysy_sequence', 'filter': {'partner_owner_recid': 'default'}}, {'collection': 'application_modules', 'filter': {}}, ] ## On vide les table d'abord dans l'environnement cible for tmp in collection_with_filter: collection = tmp['collection'] PROD_dbname[str(collection)].delete_many({}) for tmp in collection_with_filter: collection = tmp['collection'] filter = tmp['filter'] for val in REC_dbname[collection].find(filter): print(" ## val = ", val) del val['_id'] print(" ### "+str(collection)+" : "+str(val)) PROD_dbname[str(collection)].insert_one(val) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_And_Load_DB_Collection_to_DB_Collection_For_Demo_Env" """ Recuperation des formations de prod pour les charger dans une base temporaire (myclass_tmp) """ def Delete_And_Load_DB_Collection_MyClass_to_DB_Collection_For_Demo_Env(): try: #REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) PROD_dbname = client['cherifdb'] REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] """ Ici les collections avec les filtres pr chaque collection """ qry_POD = { 'isalaune':'1', 'valide':'1', 'published':'1', 'locked':'0', 'partner_owner_recid': {'$in':["236bff915d2afe1b50e61885e3ff7a4da5b00138a27f79f975", "5a73e4ef74e50f5ec1952832170377e8c0b573ac4eaf05538f", "462b11d82f53819d1ab00f246bb8235b317221578f992c416b", "1032b68ad0623914036ab22ae086b47787a2946e911b819779", "0e74b36de43dec6e3f2bb7df6b8c3a9ce640585d822db7bb67", "27280fe62d218fe65b0d730bea9af62077838ef8932018fdb6"] } } ## On vide les table d'abord dans l'environnement cible for tmp in PROD_dbname['myclass'].find(qry_POD): del tmp['_id'] print(" tmp = ", tmp) ret_val = DEM_dbname['myclass'].find_one_and_update( {'internal_url': str(tmp['internal_url'])}, {"$set": tmp}, return_document=ReturnDocument.AFTER, upsert=True, ) if (ret_val is None): mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible d'ajouter la formation ") return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_And_Load_DB_Collection_MyClass_to_DB_Collection_For_Demo_Env" """ Migration des compte partner """ def Delete_And_Load_DB_Collection_partnair_account_to_DB_Collection_For_Demo_Env(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) PROD_dbname = client['cherifdb'] REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] """ Ici les collections avec les filtres pr chaque collection """ qry_POD = {'recid': {'$in': ["236bff915d2afe1b50e61885e3ff7a4da5b00138a27f79f975", "5a73e4ef74e50f5ec1952832170377e8c0b573ac4eaf05538f", "462b11d82f53819d1ab00f246bb8235b317221578f992c416b", "1032b68ad0623914036ab22ae086b47787a2946e911b819779", "0e74b36de43dec6e3f2bb7df6b8c3a9ce640585d822db7bb67", "27280fe62d218fe65b0d730bea9af62077838ef8932018fdb6"]}} ## On vide les table d'abord dans l'environnement cible for tmp in PROD_dbname['partnair_account'].find(qry_POD): del tmp['_id'] print(" tmp = ", tmp) ret_val = DEM_dbname['partnair_account'].find_one_and_update( {'recid': str(tmp['recid'])}, {"$set": tmp}, return_document=ReturnDocument.AFTER, upsert=True, ) if (ret_val is None): mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible d'ajouter le compte partnenare ") return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_And_Load_DB_Collection_MyClass_to_DB_Collection_For_Demo_Env" """ Migration des collections de la base de recette vers la base de demo - courrier_template - courrier_template_type_document pour les valeur default """ def Delete_And_Load_DB_Collection_courrier_template_From_rec_to_Demo_Env(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] """ Ici les collections avec les filtres pr chaque collection """ qry_rec = {'partner_owner_recid': 'default'} ## On vide les table d'abord dans l'environnement cible for tmp in REC_dbname['courrier_template'].find(qry_rec): del tmp['_id'] print(" tmp = ", tmp) ret_val = DEM_dbname['courrier_template'].insert_one(tmp) if (ret_val is None): mycommon.myprint(str(inspect.stack()[0][3]) + " - errreur sur transfert de ") ## On vide les table d'abord dans l'environnement cible for tmp in REC_dbname['courrier_template_type_document'].find(qry_rec): del tmp['_id'] print(" tmp = ", tmp) ret_val = DEM_dbname['courrier_template_type_document'].insert_one(tmp) if (ret_val is None): mycommon.myprint(str(inspect.stack()[0][3]) + " - errreur sur transfert de ") return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_And_Load_DB_Collection_courrier_template_From_rec_to_Demo_Env" """ Nettoyage / mise en conformité html des articles du blog """ def Conformite_Article_Blog_Dev(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] DEV_dbname = client['cherifdb_dev'] """ Ici les collections avec les filtres pr chaque collection """ qry_rec = {'partner_owner_recid': 'default'} ## On vide les table d'abord dans l'environnement cible for tmp in DEV_dbname['articles_avis'].find({}): print(" tmp['avis'] = ", tmp['avis']) val = tmp['avis'] val = str(val).replace("

 

", "") val = str(val).replace('

', '
') val = str(val).replace("

", "") val = str(val).replace("

", '

') val = str(val).replace("

", "
") val = str(val).replace('

 ', "") val = str(val).replace('
    ', '
      ') val = str(val).replace('
        ', '
          ') print(" ===================================================") print(" val = ", val) DEV_dbname['articles_avis'].find_one_and_update({'_id': ObjectId(str(tmp['_id'])),}, {"$set": {'avis':val}}, return_document=ReturnDocument.AFTER, upsert=True, ) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Conformite_Article_Blog" def Conformite_Article_Blog_Recette(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] DEV_dbname = client['cherifdb_dev'] """ Ici les collections avec les filtres pr chaque collection """ qry_rec = {'partner_owner_recid': 'default'} ## On vide les table d'abord dans l'environnement cible for tmp in REC_dbname['articles_avis'].find({}): print(" tmp['avis'] = ", tmp['avis']) val = tmp['avis'] val = str(val).replace("

           

          ", "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "") val = str(val).replace("

          ", '

          ') val = str(val).replace("

          ", "
          ") val = str(val).replace('

           ', "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "") print(" ===================================================") print(" val = ", val) REC_dbname['articles_avis'].find_one_and_update({'_id': ObjectId(str(tmp['_id'])),}, {"$set": {'avis':val}}, return_document=ReturnDocument.AFTER, upsert=True, ) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Conformite_Article_Blog REC_dbname" def Conformite_Article_Blog_Demo(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) REC_dbname = client['cherifdb_rec'] DEM_dbname = client['cherifdb_demo'] DEV_dbname = client['cherifdb_dev'] """ Ici les collections avec les filtres pr chaque collection """ qry_rec = {'partner_owner_recid': 'default'} ## On vide les table d'abord dans l'environnement cible for tmp in DEM_dbname['articles_avis'].find({}): print(" tmp['avis'] = ", tmp['avis']) val = tmp['avis'] val = str(val).replace("

           

          ", "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "") val = str(val).replace("

          ", '

          ') val = str(val).replace("

          ", "
          ") val = str(val).replace('

           ', "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "") val = str(val).replace('


          ', '
          ') print(" ===================================================") print(" val = ", val) DEM_dbname['articles_avis'].find_one_and_update({'_id': ObjectId(str(tmp['_id'])),}, {"$set": {'avis':val}}, return_document=ReturnDocument.AFTER, upsert=True, ) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Conformite_Article_Blog DEM_dbname" def Conformite_Article_Blog_Prod(): try: # REC_base_config_dashbord CONNECTION_STRING = "mongodb://localhost:27017" client = MongoClient(CONNECTION_STRING) PROD_dbname = client['cherifdb'] """ Ici les collections avec les filtres pr chaque collection """ qry_rec = {'partner_owner_recid': 'default'} ## On vide les table d'abord dans l'environnement cible for tmp in PROD_dbname['articles_avis'].find({}): print(" tmp['avis'] = ", tmp['avis']) val = tmp['avis'] val = str(val).replace("

           

          ", "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "
          ") val = str(val).replace("

          ", '

          ') val = str(val).replace("

          ", "
          ") val = str(val).replace('

           ', "") val = str(val).replace('

          ', '
          ') val = str(val).replace("

          ", "") print(" ===================================================") print(" val = ", val) PROD_dbname['articles_avis'].find_one_and_update({'_id': ObjectId(str(tmp['_id'])),}, {"$set": {'avis':val}}, return_document=ReturnDocument.AFTER, upsert=True, ) return True, "OK" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Conformite_Article_Blog PROD_dbname" """ Cette fonction supprimer les compte utilisateur pour nicole """ delete_nicole_account_key = "ZzyCHGZESqu9bALbw0TqlFflPATX1nfFXsXYrauCO5wJcsXMnMfYpVRQBiczbE2z" def Delete_Nicole_Account(diction): try: diction = mycommon.strip_dictionary(diction) print(" inside diction = ", diction) field_list_obligatoire = ['token', 'user_email'] 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 les informations" if( str(diction['token']) != str(delete_nicole_account_key) ): mycommon.myprint( str(inspect.stack()[0][3]) + "La clé de sécurité est invalide ") return False, " La clé de sécurité est invalide " if ("nicole" not in str(diction['user_email']) and "beauches" not in str(diction['user_email'])): mycommon.myprint( str(inspect.stack()[0][3]) + " L'adresse email ne contient pas les mots clés necessaires ") return False, " L'adresse email ne contient pas les mots clés necessaires " # 0 - Recuperer le recid du partner qry = {'email':str(diction['user_email'])} partnair_account_data = MYSY_GV.dbname['partnair_account'].find_one(qry) if( partnair_account_data is None ): mycommon.myprint( str(inspect.stack()[0][3]) + " Ce compte n'existe pas ") return False, " Ce compte n'existe pas " partner_recid = partnair_account_data['recid'] print(' ## partnair_account_data = ', partnair_account_data) is_partner_admin_account = "" if( "is_partner_admin_account" in partnair_account_data.keys() and partnair_account_data['is_partner_admin_account']): is_partner_admin_account = partnair_account_data['is_partner_admin_account'] if( is_partner_admin_account != "1"): mycommon.myprint( str(inspect.stack()[0][3]) + " Vous devez fournir le compte principale pour supprimer l'abonnement ") return False, " Vous devez fournir le compte principale pour supprimer l'abonnement " """ /!\ : Verifier le compte contient les chaines (email) : "nicole" ou "beauchesne" """ deleted_message = "" # a - Suppression des inscrits delete_row = MYSY_GV.dbname['inscription'].delete_many({'partner_owner_recid':str(partner_recid)}) deleted_message = deleted_message+" -- "+str(delete_row.deleted_count) +" Inscrits Supprimé(s)" # b - Suppression des apprenants delete_row = MYSY_GV.dbname['apprenant'].delete_many({'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str(delete_row.deleted_count) + " Apprenant(s) Supprimé(s)" # c - Suppression des agendas delete_row = MYSY_GV.dbname['agenda'].delete_many({'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str(delete_row.deleted_count) + " Agenda(s) Supprimé(s)" # d - Suppression des ressources humaines delete_row = MYSY_GV.dbname['ressource_humaine'].delete_many({'partner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str(delete_row.deleted_count) + " Ressource_humaine(s) Supprimé(s)" # d bis - Suppression des ressources humaines affectation delete_row = MYSY_GV.dbname['ressource_humaine_affectation'].delete_many({'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Ressource_humaine_affectation(s) Supprimé(s)" # e - Suppression des ressources materielles delete_row = MYSY_GV.dbname['ressource_materielle'].delete_many({'partner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Ressource_materielle(s) Supprimé(s)" # e bis - Suppression des ressources materielles affectation delete_row = MYSY_GV.dbname['ressource_materielle_affectation'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Ressource_materielle_affectation(s) Supprimé(s)" # f - Suppression des clients delete_row = MYSY_GV.dbname['partner_client'].delete_many( {'partner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Ressource_materielle_affectation(s) Supprimé(s)" # f - Suppression des commandes et devis header delete_row = MYSY_GV.dbname['partner_order_header'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Commande et devis (entete) (s) Supprimé(s)" # g - Suppression des commandes et devis ligne delete_row = MYSY_GV.dbname['partner_order_line'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Commande et devis (ligne) (s) Supprimé(s)" # h - Suppression des factures entete delete_row = MYSY_GV.dbname['partner_invoice_header'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Factures (entete) (s) Supprimé(s)" # i - Suppression des factures ligne delete_row = MYSY_GV.dbname['partner_invoice_line'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Factures (ligne) (s) Supprimé(s)" # j - Suppression des factures ligne detail delete_row = MYSY_GV.dbname['partner_invoice_line_detail'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Factures (ligne details) (s) Supprimé(s)" # k - Suppression des session de formation delete_row = MYSY_GV.dbname['session_formation'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Session(s) de formation Supprimé(s)" # l - Suppression des formations delete_row = MYSY_GV.dbname['myclass'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Formations (s)" # m - Suppression des opportunités delete_row = MYSY_GV.dbname['crm_opportunite'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Opportunité (s) Supprimée(s)" # m - Suppression des unite d'enseignement delete_row = MYSY_GV.dbname['unite_enseignement'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " unite_enseignement (s) Supprimée(s)" # n - Suppression des base_partner_session_step delete_row = MYSY_GV.dbname['base_partner_session_step'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " base_partner_session_step (s) Supprimée(s)" # o - Suppression des courrier_template_tracking_history delete_row = MYSY_GV.dbname['courrier_template_tracking_history'].delete_many( {'partner_owner_recid': str(partner_recid)}) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " courrier_template_tracking_history (s) Supprimée(s)" # p - Suppression des partnair_account delete_row = MYSY_GV.dbname['partnair_account'].delete_many( {"$or": [{"recid": str(partner_recid)}, {"partner_owner_recid": str(partner_recid)}]} ) deleted_message = deleted_message + " -- " + str( delete_row.deleted_count) + " Compte(s) Utilisateur Supprimée(s)" return True, str(deleted_message) except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() print(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de Delete_Nicole_Account"