daniele/loogle-scripts
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backup automatico script del 2026-07-19 07:00

Browse Source
This commit is contained in:
daniele
2026-07-19 07:00:03 +02:00
parent 927f0d4184
commit 4802b021fe
24 changed files with 686 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/telegram-bot/net_quality.py
+54 -85
View File
@@ -4,14 +4,7 @@ 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"]
import sys
# BERSAGLIO (Cloudflare è solitamente il più stabile per i ping)
TARGET_HOST = "1.1.1.1"
@@ -33,72 +26,48 @@ def log_line(message: str) -> None:
except Exception:
pass
def send_telegram(msg, chat_ids: Optional[List[str]] = None):
"""Invia alert su Telegram e/o WebApp via alert_dispatcher."""
try:
import sys
sys.path.insert(0, "/home/daniely/docker/shared")
from loogle_core.alert_dispatcher import dispatch_alert
plain = msg.replace("*", "").replace("_", "").replace("`", "")
dispatch_alert(
"Degrado qualità linea",
plain,
category="net_quality",
severity="warning",
telegram_text=msg,
)
return
except Exception:
pass
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 send_alert(title: str, body: str, severity: str = "warning") -> None:
"""Invia alert alla WebApp Loogle Casa."""
sys.path.insert(0, "/home/daniely/docker/shared")
from loogle_core.alert_dispatcher import dispatch_alert
dispatch_alert(title, body, category="net_quality", severity=severity)
def load_state():
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE, 'r') as f: return json.load(f)
except: pass
with open(STATE_FILE, "r") as f:
return json.load(f)
except Exception:
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
with open(STATE_FILE, "w") as f:
json.dump({"alert_active": active}, f)
except Exception:
pass
def measure_quality(chat_ids: Optional[List[str]] = None):
def measure_quality(dry_run: bool = False):
print("--- Avvio Test Qualità Linea ---")
log_line("INFO 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')
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 ""
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'([0-9]+(?:[\\.,][0-9]+)?)% packet loss', output)
loss_match = re.search(r"([0-9]+(?:[\\.,][0-9]+)?)% packet loss", output)
if loss_match:
loss_raw = loss_match.group(1).replace(",", ".")
try:
@@ -107,18 +76,16 @@ def measure_quality(chat_ids: Optional[List[str]] = None):
loss = 100.0
else:
loss = 100.0
# Clamp to avoid parsing artifacts (e.g., "0.96078%" -> 0.96078, not 96078).
if loss < 0:
loss = 0.0
if loss > 100:
loss = 100.0
# 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)
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))
@@ -129,49 +96,51 @@ def measure_quality(chat_ids: Optional[List[str]] = None):
print(result_line)
log_line(f"INFO {result_line}")
# --- LOGICA ALLARME ---
# Ping senza risposte (es. iptables DROP icmp locale) → misura non affidabile.
if loss >= 99.0 and avg_ping == 0.0 and jitter == 0.0:
print("Misura non affidabile (ping senza risposte, possibile blocco ICMP locale).")
log_line("WARN Misura non affidabile: ping senza risposte")
return
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"
body_parts = ["Degrado qualità linea rilevato."]
if loss >= LIMIT_LOSS:
msg += f"🔴 **Packet Loss:** `{loss:.2f}%` (Soglia {LIMIT_LOSS}%)\n"
body_parts.append(f"Packet loss: {loss:.2f}% (soglia {LIMIT_LOSS}%)")
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)
body_parts.append(f"Jitter: {jitter}ms (soglia {LIMIT_JITTER}ms)")
body_parts.append(f"Ping medio: {avg_ping}ms")
body = "\n".join(body_parts)
if not dry_run:
send_alert("Degrado qualità linea", body, severity="warning")
save_state(True)
print("Allarme inviato.")
print("Allarme inviato." if not dry_run else "Allarme (dry-run).")
log_line("WARN Allarme inviato")
else:
print("Qualità ancora scarsa (già notificato).")
log_line("WARN 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:.2f}%`"
send_telegram(msg, chat_ids=chat_ids)
body = (
f"Qualità linea ripristinata.\n"
f"Ping: {avg_ping}ms | Jitter: {jitter}ms | Loss: {loss:.2f}%"
)
if not dry_run:
send_alert("Qualità linea ripristinata", body, severity="info")
save_state(False)
print("Recovery inviata.")
print("Recovery inviata." if not dry_run else "Recovery (dry-run).")
log_line("INFO Recovery inviata")
else:
print("Linea OK.")
log_line("INFO 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])
parser.add_argument("--dry-run", action="store_true", help="Esegui senza inviare notifiche")
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)
measure_quality(dry_run=args.dry_run)