Ela_Back/lms_chamilo/mysy_lms.py

534 lines
23 KiB
Python

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 socket
import smtplib
import re
import requests
# Connexion mariadb sur serveur
import mariadb
import hashlib
import bcrypt
"""
Cette fonction créer un utilisateur lms mysy, donc dans chamilo
"""
def Create_MySy_LMS_User(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = ['token', 'lastname', 'firstname', 'email', 'password', 'locked', 'official_code', 'username']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Le champ '" + val + "' n'est pas autorisé", False
'''
Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente,
On controle que les champs obligatoires sont presents dans la liste
'''
field_list_obligatoire = ['token', 'lastname', 'firstname', 'email', 'password', 'locked', 'official_code']
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, " Toutes les informations obligatoire n'ont pas été fournies", False
# recuperation des paramettre
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
partner_recid = ""
# Verification de la validité du token/mail dans le cas des user en mode connecté
if (len(str(my_token)) > 0):
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token n'est pas valide")
return False, " Impossible de créer le compte LMS", False
# Recuperation du recid de l'utilisateur
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if partner_recid is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de recuperer le token du partenaire")
return False, " Impossible de recuperer le token du partenaire", False
if (len(str(my_token)) <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token est vide")
return False, " Impossible de créer le compte LMS", False
lastname = ""
if ("lastname" in diction.keys()):
if diction['lastname']:
lastname = diction['lastname']
firstname = ""
if ("firstname" in diction.keys()):
if diction['firstname']:
firstname = diction['firstname']
password = ""
if ("password" in diction.keys()):
if diction['password']:
password = diction['password']
email = ""
if ("email" in diction.keys()):
if diction['email']:
email = diction['email']
locked = ""
if ("locked" in diction.keys()):
if diction['email']:
locked = diction['locked']
official_code = ""
if ("official_code" in diction.keys()):
if diction['official_code']:
official_code = diction['official_code']
username = ""
if ("username" in diction.keys()):
if diction['username']:
username = diction['username']
registration_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 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'})
mypasswd = password.encode('UTF-8')
salt = bcrypt.gensalt()
hashed_pwd = bcrypt.hashpw(mypasswd, salt)
if (retval is None or "current_val" not in retval.keys()):
mycommon.myprint(" Impossible de recuperer la sequence 'mysy_sequence' ")
return False, "Impossible de recuperer la sequence 'mysy_sequence'",False
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", False
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 = '"+lastname+"', " \
"firstname = '"+firstname+"', " \
"username = '"+username+"', " \
"salt = '"+str(salt)+"'," \
" registration_date = '"+registration_date+"'," \
" credentials_expired='0', " \
"enabled='1', " \
"expired='0', " \
"created_at='" + str( registration_date) + "'," \
" updated_at='" + str( registration_date) + "', " \
" status = '1', " \
"password = '"+str(hashed_pwd)+"', " \
"locked = '0', " \
"username_canonical = '"+username+"', " \
" email_canonical = '"+email+"', " \
"email = '"+email+"'," \
" official_code = '"+official_code+"', " \
"creator_id = '1', " \
"auth_source = 'platform'," \
"roles = 'a:0:{}', " \
"language = 'french'," \
"active = '1';"
try:
#print(" ### my_query = ", my_query)
#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}), False
conn.commit()
print(f"Last Inserted ID: {cur.lastrowid}")
lastrowid = 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, " Creation du compte LMS OK. l'id = "+str(lastrowid), str(lastrowid)
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 créer le compte LMS", False
"""
Cette fonction permet de mofifier le mot de passe de l'utilisateur
"""
def Update_Passwd_MySy_LMS_User(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = ['token', 'new_password', 'lms_user_id']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Le champ '" + val + "' n'est pas autorisé", False
'''
Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente,
On controle que les champs obligatoires sont presents dans la liste
'''
field_list_obligatoire = ['token', 'new_password', 'lms_user_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, " Toutes les informations obligatoire n'ont pas été fournies", False
# recuperation des paramettre
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
partner_recid = ""
# Verification de la validité du token/mail dans le cas des user en mode connecté
if (len(str(my_token)) > 0):
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token n'est pas valide")
return False, " Impossible de mettre à jour le compte LMS", False
# Recuperation du recid de l'utilisateur
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if partner_recid is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de recuperer le token du partenaire")
return False, " Impossible de recuperer le token du partenaire", False
if (len(str(my_token)) <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token est vide")
return False, " Impossible de mettre à jour le compte LMS", False
newpassword = ""
if ("new_password" in diction.keys()):
if diction['new_password']:
newpassword = diction['new_password']
lms_user_id = ""
if ("lms_user_id" in diction.keys()):
if diction['lms_user_id']:
lms_user_id = diction['lms_user_id']
updated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
mypasswd = newpassword.encode('UTF-8')
salt = bcrypt.gensalt()
hashed_pwd = bcrypt.hashpw(mypasswd, salt)
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 = "UPDATE user SET salt = '" + str(salt) + "', credentials_expired='0', " \
"updated_at='" + str( updated_at) + " ', " \
"password = '" + str( hashed_pwd) + "' " \
" WHERE " \
"id='" + str(lms_user_id) + "' " \
"AND user_id='" + str( lms_user_id) + "' AND enabled='1' AND status = '1' AND active = '1' ;"
try:
print(" ### my_query update = ", my_query)
# 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}), False
conn.commit()
#print(f"Last Inserted ID: {cur.lastrowid}")
lastrowid = cur.lastrowid
conn.close()
return True, " Creation du compte LMS OK. l'id modifié est = " + str(lastrowid), str(lastrowid)
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 créer le compte LMS", False
"""
Cette fonction permet de mettre à jour
- le pack abonnement et
- le nombre max de formations
"""
def Update_Pack_MySy_LMS_User(diction):
try:
'''
# Verification que les champs reçus dans l'API sont bien dans la liste des champs autorisés
# Cela evite le cas ou une entité tierce ajouter les valeurs inconnu dans l'API
# Ici on doit mettre tous les champs possible (obligatoire ou non) de la BDD dans la liste
# field_list.
'''
field_list = ['token', 'lms_user_id', 'mysy_pack', 'mysy_nb_formation']
incom_keys = diction.keys()
for val in incom_keys:
if val not in field_list:
mycommon.myprint(str(inspect.stack()[0][
3]) + " Le champ '" + val + "' n'est pas autorisé")
return False, " Le champ '" + val + "' n'est pas autorisé", False
'''
Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente,
On controle que les champs obligatoires sont presents dans la liste
'''
field_list_obligatoire = ['token', 'lms_user_id', 'mysy_pack', 'mysy_nb_formation']
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, " Toutes les informations obligatoire n'ont pas été fournies", False
# recuperation des paramettre
my_token = ""
if ("token" in diction.keys()):
if diction['token']:
my_token = diction['token']
partner_recid = ""
# Verification de la validité du token/mail dans le cas des user en mode connecté
if (len(str(my_token)) > 0):
retval = mycommon.check_partner_token_validity("", my_token)
if retval is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token n'est pas valide")
return False, " Impossible de mettre à jour le compte LMS", False
# Recuperation du recid de l'utilisateur
partner_recid = mycommon.get_parnter_recid_from_token(my_token)
if partner_recid is False:
mycommon.myprint(str(inspect.stack()[0][3]) + " - Impossible de recuperer le token du partenaire")
return False, " Impossible de recuperer le token du partenaire", False
if (len(str(my_token)) <= 0):
mycommon.myprint(str(inspect.stack()[0][3]) + " - Le token est vide")
return False, " Impossible de mettre à jour le compte LMS", False
mysy_pack = ""
if ("mysy_pack" in diction.keys()):
if diction['mysy_pack']:
mysy_pack = diction['mysy_pack']
mysy_nb_formation = ""
if ("mysy_nb_formation" in diction.keys()):
if diction['mysy_nb_formation']:
mysy_nb_formation = diction['mysy_nb_formation']
lms_user_id = ""
if ("lms_user_id" in diction.keys()):
if diction['lms_user_id']:
lms_user_id = diction['lms_user_id']
updated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
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()
""" MISE à JOUR DU mysy_pack """
query_get_mysy_pack_id = "select id from extra_field where variable = 'mysy_pack' "
print(" ### query_get_mysy_pack_id = ", query_get_mysy_pack_id)
mysy_pack_id = cur.execute(query_get_mysy_pack_id)
mysy_pack_id_val = ""
for mysy_pack_id in cur:
print(f"mysy_pack_id: ", mysy_pack_id[0] )
mysy_pack_id_val = mysy_pack_id[0]
query_2 = "select id from extra_field_values where field_id = '"+str(mysy_pack_id_val)+"' and item_id = '"+str(lms_user_id)+"' "
print(" ### query_2 = ", query_2)
mysy_pack_id2 = cur.execute(query_2)
mysy_pack_id_val2 = ""
for mysy_pack_id2 in cur:
print(f"mysy_pack_id2: ", mysy_pack_id2[0])
mysy_pack_id_val2 = mysy_pack_id2[0]
query_get_mysy_nb_formation_id = "select id from extra_field where variable = '" + str(mysy_nb_formation) + "' "
localquery = "update extra_field_values set extra_field_values.value ='"+str(mysy_pack)+"' where id = '"+str(mysy_pack_id_val2)+"' "
print(" ### localquery = ", localquery)
try:
print(" ### my_query update = ", localquery)
# cur.execute("INSERT INTO employees (first_name,last_name) VALUES (?, ?)", ("Maria", "DB"))
cur.execute(localquery)
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}), False
conn.commit()
""" MISE à JOUR DU mysy_nb_formation """
query_get_mysy_pack_id = "select id from extra_field where variable = 'mysy_nb_formation' "
print(" ### query_get_mysy_pack_id = ", query_get_mysy_pack_id)
mysy_pack_id = cur.execute(query_get_mysy_pack_id)
mysy_pack_id_val = ""
for mysy_pack_id in cur:
print(f"mysy_pack_id: ", mysy_pack_id[0])
mysy_pack_id_val = mysy_pack_id[0]
query_2 = "select id from extra_field_values where field_id = '" + str(
mysy_pack_id_val) + "' and item_id = '" + str(lms_user_id) + "' "
print(" ### query_2 = ", query_2)
mysy_pack_id2 = cur.execute(query_2)
mysy_pack_id_val2 = ""
for mysy_pack_id2 in cur:
print(f"mysy_pack_id2: ", mysy_pack_id2[0])
mysy_pack_id_val2 = mysy_pack_id2[0]
query_get_mysy_nb_formation_id = "select id from extra_field where variable = '" + str(
mysy_nb_formation) + "' "
localquery = "update extra_field_values set extra_field_values.value ='" + str(
mysy_nb_formation) + "' where id = '" + str(mysy_pack_id_val2) + "' "
print(" ### localquery = ", localquery)
try:
print(" ### my_query update = ", localquery)
# cur.execute("INSERT INTO employees (first_name,last_name) VALUES (?, ?)", ("Maria", "DB"))
cur.execute(localquery)
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}), False
conn.commit()
return True, " Creation du compte LMS OK. l'id modifié est = ", 1
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 créer le compte LMS", False