Remise en état des fichiers

app/db.py

@@ -0,0 +1,82 @@
+# 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()