Tri alphabétique

sms_alert.py

@@ -0,0 +1,115 @@
+import os
+import mysql.connector
+import ovh
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# 📍 Liste des numéros de téléphone par site
+PHONE_NUMBERS_BY_SITE = {
+    "Roissy": ["+336XXXXXXXX"],
+    "Meudon": ["+336XXXXXXXX"],
+    "Saclay": ["+336XXXXXXXX"]
+}
+
+STATE_FILE = "last_ids.txt"
+
+# ✅ Lecture des derniers IDs connus
+def load_last_ids():
+    if not os.path.exists(STATE_FILE):
+        return {}
+    with open(STATE_FILE, "r") as f:
+        return dict(line.strip().split(":") for line in f)
+
+# ✅ Sauvegarde des derniers IDs
+def save_last_ids(ids):
+    with open(STATE_FILE, "w") as f:
+        for site, id_val in ids.items():
+            f.write(f"{site}:{id_val}\n")
+
+# ✅ Fonction SMS via OVH
+def send_sms(message: str, site: str) -> None:
+    phone_numbers = PHONE_NUMBERS_BY_SITE.get(site)
+    if not phone_numbers or phone_numbers == ['']:
+        print(f"[!] Aucun numéro défini pour le site {site}. SMS non envoyé.")
+        return
+
+    try:
+        client = ovh.Client(
+            endpoint=os.getenv("OVH_ENDPOINT"),
+            application_key=os.getenv("OVH_APP_KEY"),
+            application_secret=os.getenv("OVH_APP_SECRET"),
+            consumer_key=os.getenv("OVH_CONSUMER_KEY"),
+        )
+
+        sender = os.getenv("OVH_SMS_SENDER")
+        account = os.getenv("OVH_SMS_ACCOUNT")
+
+        result = client.post(f"/sms/{account}/jobs",
+                             sender=sender,
+                             message=message,
+                             receivers=phone_numbers,
+                             priority='high',
+                             noStopClause=True,
+                             charset='UTF-8',
+                             class_='phoneDisplay',
+                             coding='7bit',
+                             senderForResponse=False,
+                             validityPeriod=2880)
+
+        print(f"[✓] SMS envoyé à {phone_numbers} pour {site}")
+    except Exception as e:
+        print(f"[!] Erreur envoi SMS OVH : {e}")
+
+# ✅ Connexion à la base Commun pour lire les identifiants des sites
+def get_db_connections():
+    conn = mysql.connector.connect(
+        host=os.getenv("DB_HOST"),
+        user=os.getenv("DB_USER"),
+        password=os.getenv("DB_PASSWORD"),
+        database=os.getenv("DB_NAME")
+    )
+    cursor = conn.cursor(dictionary=True)
+    cursor.execute("SELECT BDD, UID, PWD FROM Connexions")
+    connexions = cursor.fetchall()
+    cursor.close()
+    conn.close()
+    return connexions
+
+# ✅ Vérifie chaque site pour nouvelle ligne
+def check_new_logs():
+    last_ids = load_last_ids()
+    connexions = get_db_connections()
+
+    for conn_info in connexions:
+        site = conn_info['BDD']
+        try:
+            conn = mysql.connector.connect(
+                host=os.getenv("DB_HOST"),
+                user=conn_info['UID'],
+                password=conn_info['PWD'],
+                database=site
+            )
+            cursor = conn.cursor()
+            cursor.execute("SELECT MAX(Id) FROM Log_Inventaires")
+            result = cursor.fetchone()
+            latest_id = result[0] or 0
+            old_id = int(last_ids.get(site, 0))
+
+            print(f"{site} → dernier ID enregistré : {old_id}, ID actuel : {latest_id}")
+
+            if latest_id > old_id:
+                message = f"🔔 Nouveau log dans {site} (Id: {latest_id})"
+                send_sms(message, site)
+                last_ids[site] = str(latest_id)
+
+            cursor.close()
+            conn.close()
+        except Exception as e:
+            print(f"[{site}] Erreur : {e}")
+
+    save_last_ids(last_ids)
+
+# ✅ Point d'entrée principal
+if __name__ == "__main__":
+    check_new_logs()