Skip to content
Snippets Groups Projects
Commit 0cb27a7e authored by SBAIHI Dylan's avatar SBAIHI Dylan
Browse files

Upload New File

parent c2119c1c
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
python file/serverTCP_1_.py 0 → 100644
+ 218
− 0
View file @ 0cb27a7e
import socket
import threading
import json
import mysql.connector
import hashlib
# Configuration de la base de données
db_config = {
'host': 'localhost',
'user': 'admin',
'password': 'admin',
'database': 'departement_rt'
}
# pas oublier la requete avec la photo dans add_new student
sql_queries = {
'get_user_by_username_password': "SELECT * FROM Utilisateurs WHERE username = %s AND password = %s",
'insert_user': "INSERT IGNORE INTO Utilisateurs(username, password) VALUES (%s, %s)",
'add_new_student': "INSERT INTO Etudiant (Prenom, Nom, Annee, Email, Groupe) VALUES (%s, %s, %s, %s, %s)",
'add_new_prof': "INSERT INTO Prof (Nom, Bureau, MatSpec) VALUES (%s, %s, %s)",
'delete_user' : "",
'check_user_exist' : "SELECT * FROM Utilisateurs WHERE username = %s",
'get_nom_prof' : "SELECT * FROM Prof WHERE Nom = %s",
'get_list_student_year' : "SELECT * FROM Etudiant WHERE Annee = %s"
# Ajoutez d'autres requêtes SQL avec des mots-clés appropriés ici
}
class TCPServer:
def __init__(self, host, port):
self.host = host
self.port = port
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.bind((host, port))
self.server_socket.listen(5)
# Connexion à la base de données
try:
self.conn = mysql.connector.connect(**db_config)
except mysql.connector.Error as err:
# Gestion des erreurs de la base de données
print(f"Erreur MySQL: {err}")
response = f"Server response: Database error - {err}"
#self.server_socket.send(response.encode('utf-8'))
def handle_client(self, client_socket):
with client_socket:
while True:
request = client_socket.recv(4096).decode('utf-8')
if not request:
break
print(f"Received from client: {request}")
try:
response_data={}
existing_user = None
# Convertir la requête JSON en objet Python
query_data = json.loads(request)
# Extraire les données de la requête
query_type = query_data.get('request')
query_params = query_data.get('params')
# Initialisation du curseur :
cursor = self.conn.cursor()
if query_type == 'get_user_by_username_password':
# Code pour la requête 'get_user_by_username_password'
print("Exécution de la requête 'get_user_by_username_password'")
cursor.execute(sql_queries[query_type], query_params)
results = cursor.fetchall()
if results:
response_data = {'success': True, 'data': results}
print("Request success : ", "data : ", results)
else:
response_data = {'success': False, 'error': 'Aucun résultat trouvé ou la requete à échoué'}
print("Request failed : ", response_data)
# Traitement des résultats...
elif query_type == 'get_nom_prof':
# Code pour la requête 'get_nom_prof'
print("Exécution de la requête 'get_nom_prof'")
nom = query_params
cursor.execute(sql_queries[query_type], (nom))
results = cursor.fetchall()
if results:
response_data = {'success': True, 'data': results}
print("Request success : ", "data : ", results)
else:
response_data = {'success': False, 'error': 'Aucun résultat trouvé ou la requete à échoué'}
print("Request failed : ", response_data)
# Traitement des résultats...
elif query_type == 'get_list_student_year':
# Code pour la requête 'get_list_student_year'
print("Exécution de la requête 'get_list_student_year'")
year = query_params
cursor.execute(sql_queries[query_type], (year))
results = cursor.fetchall()
all_stud=[]
for stud in results:
info = [stud[1],stud[2]]
all_stud.append(info)
if results:
response_data = {'success': True, 'data': all_stud}
print("Request success : ", "data : ", results)
else:
response_data = {'success': False, 'error': 'Aucun résultat trouvé ou la requete à échoué'}
print("Request failed : ", response_data)
# Traitement des résultats...
elif query_type == 'insert_user':
# Code pour la requête 'insert_user'
print("Exécution de la requête 'insert_user'")
username, password = query_params # Assurez-vous que query_params a les bons éléments
# Décrypter le mot de passe
decrypted_password = self.decrypt_password(password)
query_insert_user = sql_queries[query_type]
cursor.execute(query_insert_user, (username, password))
# Récupérer l'ID du dernier enregistrement inséré
last_inserted_id = cursor.lastrowid
print(f"L'utilisateur avec le nom {username} a été inséré avec succès. ID : {last_inserted_id}")
elif query_type == 'add_new_student':
# Code pour la requête 'add_new_student'
print("Exécution de la requête 'add_new_student'")
Prenom, Nom, Annee, Email, Groupe = query_params
if existing_user != None:
print(f"L'étudiant : {query_params[0]} existe déjà dans la BD.")
else:
#l'étudiant n'est pas présent dans la BD :
query_add_new_student = sql_queries[query_type]
cursor.execute(query_add_new_student, (Prenom, Nom, Annee, Email, Groupe))
last_inserted_id = cursor.lastrowid
print(f"L'étudiant : {Prenom} a été inséré avec succès dans Etudiant. ID : {last_inserted_id}")
# Traitement des résultats...
elif query_type == 'add_new_prof':
# Code pour la requête 'add_new_prof'
print("Exécution de la requête 'add_new_prof'")
Nom, Bureau, Matiere = query_params
if existing_user != None:
print(f"Le professeur : {query_params[0]} existe déjà dans la BD. {query_params}")
else:
query_add_new_prof = sql_queries[query_type]
print(query_params,"QUERY PARAMS")
cursor.execute(query_add_new_prof, (Nom, Bureau, Matiere))
last_inserted_id = cursor.lastrowid
print(f"Le professeur : {Nom} a été inséré avec succès dans Prof. ID : {last_inserted_id}")
# Traitement des résultats...
elif query_type == 'delete_user':
print("Exécution de la requête 'delete_user'")
# Code pour la requête 'delete_user'
# ...
pass
elif query_type == 'check_user_exist':
# Code pour la requête 'check_user_exist'
print("Exécution de la requête 'check_user_exist'")
cursor.execute(sql_queries[query_type], query_params)
results = cursor.fetchall()
if results:
response_data = {'success': True, 'data': results}
print("Request success : ", "data : ", results)
else:
response_data = {'success': False, 'error': 'Aucun résultat trouvé ou la requete à échoué'}
print("Request failed : ", response_data)
# Traitement des résultats...
else:
print(f"Requête non reconnue : {query_type}")
# Requête non reconnue
response_data = {'success': False, 'error': 'Requête non reconnue', 'invalid_request': True}
# Commit si nécessaire
if query_type.upper().startswith(('INSERT', 'UPDATE', 'DELETE')):
self.conn.commit()
response = json.dumps(response_data)
client_socket.send(response.encode('utf-8'))
client_socket.send("".encode('utf-8'))
except mysql.connector.Error as err:
# Gestion des erreurs
print(f"Erreur MySQL: {err}")
# ... (votre code précédent)
def start(self):
print("Server is ready, waiting for connections...")
while True:
client, addr = self.server_socket.accept()
print(f"Connection accepted from {addr[0]}:{addr[1]}")
client_handler = threading.Thread(target=self.handle_client, args=(client,))
client_handler.start()
def decrypt_password(self,encrypted_password):
# Retirer la chaîne spéciale
reversed_password = encrypted_password.replace("_special_string", "")
# Inverser à nouveau l'ordre des caractères
decrypted_password = reversed_password[::-1]
return decrypted_password
# Utilisation de la classe TCPServer
if __name__ == "__main__":
server_instance = TCPServer('localhost', 50000)
server_instance.start()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment