136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
import argparse
|
|
import subprocess
|
|
import re
|
|
import os
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
from typing import List, Optional
|
|
|
|
# --- CONFIGURAZIONE ---
|
|
BOT_TOKEN="8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
|
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
|
|
|
# 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, chat_ids: Optional[List[str]] = None):
|
|
"""
|
|
Args:
|
|
msg: Messaggio da inviare
|
|
chat_ids: Lista di chat IDs (default: TELEGRAM_CHAT_IDS)
|
|
"""
|
|
if not BOT_TOKEN or "INSERISCI" in BOT_TOKEN: return
|
|
if chat_ids is None:
|
|
chat_ids = TELEGRAM_CHAT_IDS
|
|
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
|
for chat_id in 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(chat_ids: Optional[List[str]] = None):
|
|
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, chat_ids=chat_ids)
|
|
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, chat_ids=chat_ids)
|
|
save_state(False)
|
|
print("Recovery inviata.")
|
|
else:
|
|
print("Linea OK.")
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Network quality monitor")
|
|
parser.add_argument("--debug", action="store_true", help="Invia messaggi solo all'admin (chat ID: %s)" % TELEGRAM_CHAT_IDS[0])
|
|
args = parser.parse_args()
|
|
|
|
# In modalità debug, invia solo al primo chat ID (admin)
|
|
chat_ids = [TELEGRAM_CHAT_IDS[0]] if args.debug else None
|
|
|
|
measure_quality(chat_ids=chat_ids)
|