83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
# utils/db.py
|
||
|
||
from datetime import datetime
|
||
from app.utils_db import connect_to_mysql # ✅ Import centralisé
|
||
|
||
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 Sondes.Chaufferie c1
|
||
INNER JOIN (
|
||
SELECT Sonde, MAX(Date) AS MaxDate
|
||
FROM Sondes.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 Sondes.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 Sondes.MotsDePasse
|
||
WHERE utilisateur = %s AND mot_de_passe = %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 Sondes.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 Sondes.Alertes_Chaufferie SET Etat = 'Acquitté' WHERE Id = %s"
|
||
cursor.execute(query, (id_alerte,))
|
||
db.commit()
|
||
cursor.close()
|
||
db.close()
|