Sync Loogle Bot 2025-12-26 15:02

services/telegram-bot/net_quality.py

@@ -0,0 +1,119 @@
+import subprocess
+import re
+import os
+import json
+import time
+import urllib.request
+import urllib.parse
+
+# --- CONFIGURAZIONE ---
+BOT_TOKEN="8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
+CHAT_ID="64463169"
+
+# BERSAGLIO (Cloudflare è solitamente il più stabile per i ping)
+TARGET_HOST = "1.1.1.1"
+
+# SOGLIE DI ALLARME
+LIMIT_LOSS = 5.0   # % di pacchetti persi (sopra il 5% è grave)
+LIMIT_JITTER = 30.0 # ms di deviazione (sopra 30ms lagga la voce/gioco)
+
+# File di stato
+STATE_FILE = "/home/daniely/docker/telegram-bot/quality_state.json"
+
+def send_telegram(msg):
+    if "INSERISCI" in TELEGRAM_BOT_TOKEN: return
+    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
+    for chat_id in TELEGRAM_CHAT_IDS:
+        try:
+            payload = {"chat_id": chat_id, "text": msg, "parse_mode": "Markdown"}
+            data = urllib.parse.urlencode(payload).encode('utf-8')
+            req = urllib.request.Request(url, data=data)
+            with urllib.request.urlopen(req) as r: pass
+            time.sleep(0.2)
+        except: pass
+
+def load_state():
+    if os.path.exists(STATE_FILE):
+        try:
+            with open(STATE_FILE, 'r') as f: return json.load(f)
+        except: pass
+    return {"alert_active": False}
+
+def save_state(active):
+    try:
+        with open(STATE_FILE, 'w') as f: json.dump({"alert_active": active}, f)
+    except: pass
+
+def measure_quality():
+    print("--- Avvio Test Qualità Linea ---")
+    
+    # Esegue 50 ping rapidi (0.2s intervallo)
+    # -q: quiet (solo riepilogo finale)
+    # -c 50: conta 50 pacchetti
+    # -i 0.2: intervallo rapido
+    # -w 15: timeout massimo 15 secondi
+    cmd = f"ping -c 50 -i 0.2 -q -w 15 {TARGET_HOST}"
+    
+    try:
+        output = subprocess.check_output(cmd, shell=True).decode('utf-8')
+    except subprocess.CalledProcessError as e:
+        # Se ping fallisce completamente (es. internet down), catturiamo l'output comunque se c'è
+        output = e.output.decode('utf-8') if e.output else ""
+        if not output:
+            print("Errore critico: Nessuna connessione.")
+            return
+
+    # Parsing Packet Loss
+    # Cerca pattern: "X% packet loss"
+    loss_match = re.search(r'(\d+)% packet loss', output)
+    loss = int(loss_match.group(1)) if loss_match else 100
+
+    # Parsing Jitter (mdev)
+    # Output tipico: rtt min/avg/max/mdev = 10.1/12.5/40.2/5.1 ms
+    # mdev è la 4a cifra
+    jitter = 0.0
+    rtt_match = re.search(r'rtt min/avg/max/mdev = ([\d\.]+)/([\d\.]+)/([\d\.]+)/([\d\.]+)', output)
+    
+    if rtt_match:
+        avg_ping = float(rtt_match.group(2))
+        jitter = float(rtt_match.group(4))
+    else:
+        avg_ping = 0.0
+
+    print(f"Risultati: Loss={loss}% | Jitter={jitter}ms | AvgPing={avg_ping}ms")
+
+    # --- LOGICA ALLARME ---
+    state = load_state()
+    was_active = state.get("alert_active", False)
+    
+    is_bad = (loss >= LIMIT_LOSS) or (jitter >= LIMIT_JITTER)
+    
+    if is_bad:
+        if not was_active:
+            # NUOVO ALLARME
+            msg = f"📉 **DEGRADO QUALITÀ LINEA**\n\n"
+            if loss >= LIMIT_LOSS:
+                msg += f"🔴 **Packet Loss:** `{loss}%` (Soglia {LIMIT_LOSS}%)\n"
+            if jitter >= LIMIT_JITTER:
+                msg += f"⚠️ **Jitter (Instabilità):** `{jitter}ms` (Soglia {LIMIT_JITTER}ms)\n"
+            
+            msg += f"\n_Ping Medio: {avg_ping}ms_"
+            send_telegram(msg)
+            save_state(True)
+            print("Allarme inviato.")
+        else:
+            print("Qualità ancora scarsa (già notificato).")
+
+    elif was_active and not is_bad:
+        # RECOVERY
+        msg = f"✅ **QUALITÀ LINEA RIPRISTINATA**\n\n"
+        msg += f"I parametri sono rientrati nella norma.\n"
+        msg += f"Ping: `{avg_ping}ms` | Jitter: `{jitter}ms` | Loss: `{loss}%`"
+        send_telegram(msg)
+        save_state(False)
+        print("Recovery inviata.")
+    else:
+        print("Linea OK.")
+
+if __name__ == "__main__":
+    measure_quality()