Mise en place d'un journal de connexions

app/Monitor_connexions.py

@@ -0,0 +1,183 @@
+import os
+import time
+import logging
+from typing import Optional
+import json
+import pymysql
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(message)s"
+)
+log = logging.getLogger("journal_connexions")
+
+
+def env_str(name: str, default: Optional[str] = None) -> Optional[str]:
+    value = os.getenv(name, default)
+    if value is None:
+        return None
+    value = value.strip()
+    return value if value else default
+
+
+DB_CONFIG = {
+    "host": env_str("DB_HOST"),
+    "port": int(env_str("DB_PORT", "3306")),
+    "user": env_str("DB_USER2"),
+    "password": env_str("DB_PASS2"),
+    "database": env_str("DB_NAME2", "Acces"),
+    "charset": "utf8mb4",
+    "cursorclass": pymysql.cursors.DictCursor,
+    "autocommit": True,
+}
+
+SYNO_CHAT_WEBHOOK = env_str("SYNO_CHAT_WEBHOOK_CONNEXIONS")
+SYNO_CHAT_BOTNAME = env_str("SYNO_CHAT_BOTNAME_CONNEXIONS", "Journal Connexions")
+POLL_INTERVAL = int(env_str("POLL_INTERVAL", "10"))
+
+
+def get_connection():
+    return pymysql.connect(**DB_CONFIG)
+
+
+def format_message(row: dict) -> str:
+    return (
+        f"[Connexion MySQL]\n"
+        f"Utilisateur : {row.get('NomUtilisateur', '')}\n"
+        f"Poste : {row.get('PosteClient', '')}\n"
+        f"Tableur : {row.get('TableurSource', '')}\n"
+        f"Windows : {row.get('UtilisateurWindows', '')}\n"
+        f"Site : {row.get('SiteDemande', '')}\n"
+        f"Service : {row.get('ServiceDemande', '')}\n"
+        f"DSN : {row.get('DSN', '')}\n"
+        f"BDD : {row.get('BDD', '')}\n"
+        f"Statut : {row.get('Statut', '')}\n"
+        f"Motif : {row.get('Motif', '')}\n"
+        f"Heure : {row.get('DateHeure', '')}\n"
+        f"Session : {row.get('SessionID', '')}"
+    )
+
+
+def send_synology_chat(message: str) -> None:
+    if not SYNO_CHAT_WEBHOOK:
+        raise RuntimeError("SYNO_CHAT_WEBHOOK_CONNEXIONS non configuré")
+
+    syno_payload = {
+        "text": message
+    }
+
+    response = requests.post(
+        SYNO_CHAT_WEBHOOK,
+        data={"payload": json.dumps(syno_payload, ensure_ascii=False)},
+        timeout=10
+    )
+
+    log.info("Synology Chat HTTP=%s body=%s", response.status_code, response.text)
+
+    response.raise_for_status()
+
+    body = response.json()
+    if not body.get("success", False):
+        raise RuntimeError(f"Synology Chat erreur: {body}")
+
+
+def fetch_pending_rows(conn) -> list[dict]:
+    sql = """
+        SELECT
+            Id_Journal,
+            DateHeure,
+            NomUtilisateur,
+            PosteClient,
+            TableurSource,
+            UtilisateurWindows,
+            SiteDemande,
+            ServiceDemande,
+            DSN,
+            BDD,
+            Statut,
+            Motif,
+            SessionID
+        FROM `Acces`.`JournalConnexions`
+        WHERE NotificationEnvoyee = 0
+        ORDER BY DateHeure ASC, Id_Journal ASC
+        LIMIT 50
+    """
+    print("SQL fetch_pending_rows =")
+    print(sql)
+
+    with conn.cursor() as cur:
+        cur.execute(sql)
+        return cur.fetchall()
+
+
+def mark_sent(conn, row_id: int) -> None:
+    sql = """
+        UPDATE Acces.JournalConnexions
+        SET NotificationEnvoyee = 1,
+            DateNotification = NOW(),
+            ErreurNotification = NULL
+        WHERE Id_Journal = %s
+    """
+    with conn.cursor() as cur:
+        cur.execute(sql, (row_id,))
+
+
+def mark_error(conn, row_id: int, error_msg: str) -> None:
+    sql = """
+        UPDATE `Acces`.`JournalConnexions`
+        SET ErreurNotification = %s
+        WHERE Id_Journal = %s
+    """
+    with conn.cursor() as cur:
+        cur.execute(sql, (error_msg[:255], row_id))
+
+
+def process_once() -> None:
+    with get_connection() as conn:
+        rows = fetch_pending_rows(conn)
+
+        if not rows:
+            log.info("Aucune nouvelle connexion à notifier.")
+            return
+
+        log.info("%s connexion(s) à notifier.", len(rows))
+
+        for row in rows:
+            row_id = row["Id_Journal"]
+            try:
+                message = format_message(row)
+                send_synology_chat(message)
+                mark_sent(conn, row_id)
+                log.info("Notification envoyée pour Id_Journal=%s", row_id)
+
+            except Exception as exc:
+                log.exception("Erreur d'envoi pour Id_Journal=%s", row_id)
+                try:
+                    mark_error(conn, row_id, str(exc))
+                except Exception:
+                    log.exception("Impossible d'écrire ErreurNotification pour Id_Journal=%s", row_id)
+
+
+def main():
+    missing = [k for k, v in DB_CONFIG.items() if v is None and k != "port"]
+    if missing:
+        raise RuntimeError(f"Variables d'environnement manquantes : {', '.join(missing)}")
+
+    if not SYNO_CHAT_WEBHOOK:
+        raise RuntimeError("SYNO_CHAT_WEBHOOK_CONNEXIONS manquant")
+
+    log.info("Surveillance de JournalConnexions démarrée.")
+    while True:
+        try:
+            process_once()
+        except Exception:
+            log.exception("Erreur générale dans la boucle de surveillance")
+        time.sleep(POLL_INTERVAL)
+
+
+if __name__ == "__main__":
+    main()