120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
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()
|