116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
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()
|