102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
# utils/db.py
|
||
|
||
import mysql.connector
|
||
import os
|
||
import sys
|
||
import logging
|
||
from datetime import datetime
|
||
|
||
|
||
def connect_to_mysql(db_name: str = None):
|
||
"""Connexion à MySQL via variables d'environnement."""
|
||
try:
|
||
mydb = mysql.connector.connect(
|
||
host=os.getenv("DB_HOST"),
|
||
user=os.getenv("DB_USER"),
|
||
password=os.getenv("DB_PASSWORD"),
|
||
database=db_name if db_name else os.getenv("DB_NAME")
|
||
)
|
||
return mydb
|
||
except mysql.connector.Error as err:
|
||
logging.error(f"Erreur de connexion MySQL : {err}")
|
||
sys.exit(1)
|
||
|
||
|
||
def get_latest_chaufferie():
|
||
"""Renvoie les dernières valeurs par sonde dans la table 'Chaufferie'."""
|
||
db = connect_to_mysql()
|
||
cursor = db.cursor(dictionary=True)
|
||
query = """
|
||
SELECT c1.*
|
||
FROM Chaufferie c1
|
||
INNER JOIN (
|
||
SELECT Sonde, MAX(Date) AS MaxDate
|
||
FROM Chaufferie
|
||
GROUP BY Sonde
|
||
) c2 ON c1.Sonde = c2.Sonde AND c1.Date = c2.MaxDate
|
||
ORDER BY c1.Sonde;
|
||
"""
|
||
cursor.execute(query)
|
||
result = cursor.fetchall()
|
||
cursor.close()
|
||
db.close()
|
||
return result
|
||
|
||
|
||
def get_history_by_sonde(sonde: str, start: datetime, end: datetime):
|
||
"""Retourne l’historique des températures d’une sonde entre deux dates."""
|
||
db = connect_to_mysql()
|
||
cursor = db.cursor(dictionary=True)
|
||
query = """
|
||
SELECT * FROM Chaufferie
|
||
WHERE Sonde = %s AND Date BETWEEN %s AND %s
|
||
ORDER BY Date;
|
||
"""
|
||
cursor.execute(query, (sonde, start, end))
|
||
result = cursor.fetchall()
|
||
cursor.close()
|
||
db.close()
|
||
return result
|
||
|
||
|
||
def verifier_utilisateur_commun(utilisateur: str, motdepasse: str):
|
||
"""Vérifie si un utilisateur (non superviseur) existe dans la table MotsDePasse."""
|
||
db = connect_to_mysql()
|
||
cursor = db.cursor(dictionary=True)
|
||
query = """
|
||
SELECT * FROM MotsDePasse
|
||
WHERE utilisateur = %s AND motdepasse = %s AND role = 'utilisateur'
|
||
"""
|
||
cursor.execute(query, (utilisateur, motdepasse))
|
||
result = cursor.fetchone()
|
||
cursor.close()
|
||
db.close()
|
||
return result
|
||
|
||
|
||
def lire_alertes_sondes():
|
||
"""Renvoie la liste des alertes non acquittées dans la table Alertes_Chaufferie."""
|
||
db = connect_to_mysql()
|
||
cursor = db.cursor(dictionary=True)
|
||
query = """
|
||
SELECT * FROM Alertes_Chaufferie
|
||
WHERE Etat != 'Acquitté'
|
||
ORDER BY Debut_defaut DESC
|
||
"""
|
||
cursor.execute(query)
|
||
result = cursor.fetchall()
|
||
cursor.close()
|
||
db.close()
|
||
return result
|
||
|
||
|
||
def acquitter_alerte(id_alerte: int):
|
||
"""Met à jour une alerte comme acquittée dans la base."""
|
||
db = connect_to_mysql()
|
||
cursor = db.cursor()
|
||
query = "UPDATE Alertes_Chaufferie SET Etat = 'Acquitté' WHERE Id = %s"
|
||
cursor.execute(query, (id_alerte,))
|
||
db.commit()
|
||
cursor.close()
|
||
db.close()
|