Ela_Back/tools_cherif/tools_cherif.py

1747 lines
75 KiB
Python

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 = '''
<!DOCTYPE html>
<html>
<body>
<h2> Votre Site mysy-training.com est KO </h2>
Bonjour, <br/>
Merci de verifier l'état de votre site MySy-Traning.com
<br/>
</body>
</html>'''
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 = "<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<soap:Envelope xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\" xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\">\n <soap:Header>\n <UserCredentials xmlns=\"http://microsoft.com/webservices/\">\n <userName>---</userName>\n <password>---</password>\n </UserCredentials>\n </soap:Header>\n <soap:Body>\n <getTrainScheduleJSON xmlns=\"http://microsoft.com/webservices/\">\n <station>NY</station>\n </getTrainScheduleJSON>\n </soap:Body>\n</soap:Envelope>"
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 = """
<soap:Envelope xmlns:soap="http://www.w3.org/2003/05/soap-envelope" xmlns:tem="http://tempuri.org/">
<soap:Header/>
<soap:Body>
<tem:Add>
<tem:intA>50</tem:intA>
<tem:intB>20</tem:intB>
</tem:Add>
</soap:Body>
</soap:Envelope>"""
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 = """
<soap:Envelope xmlns:soap="http://www.w3.org/2003/05/soap-envelope" xmlns:tem="http://tempuri.org/">
<soap:Header/>
<soap:Body>
<tem:WS.test>
</tem:WS.test>
</soap:Body>
</soap:Envelope>"""
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 = """<?xml version="1.0" encoding="UTF-8"?>
<SOAP-ENV:Envelope xmlns:ns0="https://lms.mysy-training.com/main/webservices/" xmlns:ns1="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/">
<SOAP-ENV:Header/>
<ns1:Body>
<ns0:WSCreateUsers>
<secret_key>135b6ca3b7ed29fa787ff011859fd76ec5e1b51c</secret_key>
<users>['1504']</users>
</ns0:WSCreateUsers>
</ns1:Body>
</SOAP-ENV:Envelope>"""
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"
def test_web_service(diction):
try:
diction = mycommon.strip_dictionary(diction)
"""
Verification des input acceptés
"""
field_list = [ 'token', 'session_id', ]
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'est pas autorisé")
return False, " Les informations fournies sont incorrectes"
"""
Verification des champs obligatoires
"""
field_list_obligatoire = ['token', 'session_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 liste ")
return False, " Les informations fournies sont incorrectes"
local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction)
if (local_status is not True):
return local_status, my_partner
# 1 - Recuperer la liste des participants
List_participant = []
for tmp in MYSY_GV.dbname['inscription'].find({'session_id':diction['session_id'], 'status':'1'},{'_id':1, 'email':1}):
node = {}
node['_id'] = str(tmp['_id'])
node['email'] = str(tmp['email'])
List_participant.append(node)
# 2 - Liste des documents traqués
List_Document_a_Traquer = []
for tmp in MYSY_GV.dbname['courrier_template_tracking'].find({'partner_owner_recid':str(my_partner['recid']),
'valide':'1', 'locked':'0'},
{'_id':1, 'courrier_template_type_document_id':1,
'courrier_template_type_document_ref_interne':1, 'rang':1}).sort([("rang", pymongo.ASCENDING)]):
node = {}
node['_id'] = str(tmp['_id'])
node['courrier_template_type_document_id'] = str(tmp['courrier_template_type_document_id'])
node['courrier_template_type_document_ref_interne'] = str(tmp['courrier_template_type_document_ref_interne'])
node['rang'] = str(tmp['rang'])
List_Document_a_Traquer.append(node)
print(" LISTE des PARTIPANTS")
for val in List_participant:
print("Participant : "+str(val))
print(" LISTE des DOCUMENT A TRAQUER")
for val in List_Document_a_Traquer:
print("Document a traquer "+str(val))
"""
Si pour chaque type de document a traquer, tous les partipants sont dans history, alors on mais ok
si aucun, on met ko,
si au moins 1, on met : partiel
la clé est : courrier_template_tracking.courrier_template_type_document_ref_interne
-- clé de recherhce : document a traquer (courrier_template_tracking._id) et inscription_id
"""
for doc_a_tracker in List_Document_a_Traquer:
print("TRAITEMENT DU DOCUMENT " + str(doc_a_tracker['courrier_template_type_document_ref_interne'])+" - _id = "+ str(doc_a_tracker['_id']))
local_nb_inscrit = 0
local_nb_has_history_tracking = 0
local_status = "Aucun"
for val in List_participant:
#print("Participant zzz: " + str(val))
print(" cle de recherche : courrier_template_tracking_id = ",str(doc_a_tracker['_id']), " / target_collection_recid = "+str(val['_id']) )
local_nb_inscrit = local_nb_inscrit + 1
local_qry = {'partner_owner_recid':str(my_partner['recid']),
'valide':'1', 'locked':'0', 'target_collection_recid':str(val['_id']),
'courrier_template_tracking_id':str(doc_a_tracker['_id'])}
is_tracking_history = MYSY_GV.dbname['courrier_template_tracking_history'].count_documents(local_qry)
if( is_tracking_history > 0 ):
local_nb_has_history_tracking = local_nb_has_history_tracking + 1
print("local_qry : " + str(local_qry))
print(" Pour le document a traquer : ", doc_a_tracker['courrier_template_type_document_ref_interne'])
print("local_nb_inscrit VS local_nb_has_history_tracking: " + str(local_nb_inscrit)+" VS "+str(local_nb_has_history_tracking))
if( local_nb_inscrit == local_nb_has_history_tracking and local_nb_has_history_tracking == 0 ):
local_status = "Aucun"
if( local_nb_inscrit == local_nb_has_history_tracking and local_nb_has_history_tracking != 0 ):
local_status = "Complet"
if (local_nb_inscrit > local_nb_has_history_tracking and local_nb_has_history_tracking > 0):
local_status = "Partiel"
doc_a_tracker['global_status'] = str(local_status)
print(" Statut : ", local_status)
print(" END LISTE des DOCUMENT A TRAQUER WITH STATUS ")
for val in List_Document_a_Traquer:
print("Document a traquer " + str(val['courrier_template_type_document_ref_interne'])+" STATUS = "+ str(val['global_status']) )
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("<p>&nbsp;</p>", "")
val = str(val).replace('<h3 style="font-size: 1.5rem;">',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
val = str(val).replace("<p>", ' <div style="font-family: Inter; font-style: normal; font-weight: 400; font-size: 20px; line-height: 32px; color: #4B5162; margin-bottom: 2rem;">')
val = str(val).replace("</p>", "</div>")
val = str(val).replace('<h3 class="MsoNormal">&nbsp;</div>', "")
val = str(val).replace('<ol>', '<ol style="margin-bottom: 2rem;">')
val = str(val).replace('<ul>', '<ul style="margin-bottom: 2rem;">')
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("<p>&nbsp;</p>", "")
val = str(val).replace('<h3 style="font-size: 1.5rem;">',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
val = str(val).replace("<p>", ' <div style="font-family: Inter; font-style: normal; font-weight: 400; font-size: 20px; line-height: 32px; color: #4B5162; margin-bottom: 2rem;">')
val = str(val).replace("</p>", "</div>")
val = str(val).replace('<h3 class="MsoNormal">&nbsp;</div>', "")
val = str(val).replace('<h3>',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
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("<p>&nbsp;</p>", "")
val = str(val).replace('<h3 style="font-size: 1.5rem;">',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
val = str(val).replace("<p>", ' <div style="font-family: Inter; font-style: normal; font-weight: 400; font-size: 20px; line-height: 32px; color: #4B5162; margin-bottom: 2rem;">')
val = str(val).replace("</p>", "</div>")
val = str(val).replace('<h3 class="MsoNormal">&nbsp;</div>', "")
val = str(val).replace('<h3>',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
val = str(val).replace('<h3><br>',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
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("<p>&nbsp;</p>", "")
val = str(val).replace('<h3 style="font-size: 1.5rem;">',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
val = str(val).replace("<p>", ' <div style="font-family: Inter; font-style: normal; font-weight: 400; font-size: 20px; line-height: 32px; color: #4B5162; margin-bottom: 2rem;">')
val = str(val).replace("</p>", "</div>")
val = str(val).replace('<h3 class="MsoNormal">&nbsp;</div>', "")
val = str(val).replace('<h3>',
'<div style="font-family: DM Sans; font-style: normal;font-weight: 600; font-size: 24px; line-height: 28px; color: #16274F; text-align: left; margin-bottom: 10px;"> ')
val = str(val).replace("</h3>", "</div>")
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"