100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
import mysql.connector
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
def lire_alertes_sondes():
|
|
conn = None
|
|
cursor = None
|
|
try:
|
|
conn = get_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
cursor.execute("SELECT * FROM Alertes_Chaufferie ORDER BY Debut_defaut DESC")
|
|
return cursor.fetchall()
|
|
except Exception as e:
|
|
import streamlit as st
|
|
st.warning(f"Erreur lecture alertes : {e}")
|
|
return []
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if conn and conn.is_connected():
|
|
conn.close()
|
|
|
|
def acquitter_alerte(id_alerte):
|
|
conn = None
|
|
cursor = None
|
|
try:
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("UPDATE Alertes_Chaufferie SET Etat = 'Acquitté' WHERE Id = %s", (id_alerte,))
|
|
conn.commit()
|
|
except Exception as e:
|
|
import streamlit as st
|
|
st.warning(f"Erreur lors de l'acquittement : {e}")
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if conn and conn.is_connected():
|
|
conn.close()
|
|
|
|
|
|
def verifier_utilisateur_commun(login, password):
|
|
conn = get_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
query = """
|
|
SELECT NomUtilisateur FROM Commun.AccesUtilisateurs
|
|
WHERE NomUtilisateur = %s
|
|
AND MotDePasse = %s
|
|
AND (DateExpiration IS NULL OR DateExpiration >= CURDATE())
|
|
"""
|
|
cursor.execute(query, (login, password))
|
|
user = cursor.fetchone()
|
|
cursor.close()
|
|
conn.close()
|
|
return user # None si non trouvée
|
|
|
|
def get_connection():
|
|
return mysql.connector.connect(
|
|
host=os.getenv("DB_HOST"),
|
|
user=os.getenv("DB_USER"),
|
|
password=os.getenv("DB_PASSWORD"),
|
|
database=os.getenv("DB_NAME")
|
|
)
|
|
|
|
def get_latest_chaufferie():
|
|
conn = get_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
query = """
|
|
SELECT c1.Sonde, c1.Temperature, c1.Date, c1.Topic
|
|
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.Topic, c1.Sonde
|
|
"""
|
|
cursor.execute(query)
|
|
data = cursor.fetchall()
|
|
cursor.close()
|
|
conn.close()
|
|
return data
|
|
|
|
def get_history_by_sonde(sonde_name):
|
|
conn = get_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
query = """
|
|
SELECT Date, Temperature
|
|
FROM Chaufferie
|
|
WHERE Sonde = %s AND Date >= NOW() - INTERVAL 1 DAY
|
|
ORDER BY Date ASC
|
|
"""
|
|
cursor.execute(query, (sonde_name,))
|
|
data = cursor.fetchall()
|
|
cursor.close()
|
|
conn.close()
|
|
return data
|