import logging import subprocess import os from functools import wraps from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import ( Application, CommandHandler, CallbackQueryHandler, ContextTypes, ) # --- CONFIGURAZIONE --- BOT_TOKEN = os.environ.get('BOT_TOKEN') OWNER_ID = int(os.environ.get('ALLOWED_USER_ID')) SSH_USER = "daniely" NAS_USER = "daniely" MASTER_IP = "192.168.128.80" # --- LISTE DISPOSITIVI --- CORE_DEVICES = [ {"name": "πŸ“ Pi-1 (Master)", "ip": MASTER_IP, "type": "pi", "user": SSH_USER}, {"name": "πŸ“ Pi-2 (Backup)", "ip": "127.0.0.1", "type": "local", "user": ""}, {"name": "πŸ—„οΈ NAS 920+", "ip": "192.168.128.100", "type": "nas", "user": NAS_USER}, {"name": "πŸ—„οΈ NAS 214", "ip": "192.168.128.90", "type": "nas", "user": NAS_USER} ] INFRA_DEVICES = [ {"name": "πŸ“‘ Router", "ip": "192.168.128.1"}, {"name": "πŸ“Ά WiFi Sala", "ip": "192.168.128.101"}, {"name": "πŸ“Ά WiFi Luca", "ip": "192.168.128.102"}, {"name": "πŸ“Ά WiFi Taverna", "ip": "192.168.128.103"}, {"name": "πŸ“Ά WiFi Dado", "ip": "192.168.128.104"}, {"name": "πŸ”Œ Sw Sala", "ip": "192.168.128.105"}, {"name": "πŸ”Œ Sw Main", "ip": "192.168.128.106"}, {"name": "πŸ”Œ Sw Lav", "ip": "192.168.128.107"}, {"name": "πŸ”Œ Sw Tav", "ip": "192.168.128.108"} ] logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO) logger = logging.getLogger(__name__) # --- FUNZIONI --- def run_cmd(command, ip=None, user=None): try: if ip == "127.0.0.1" or ip is None: return subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, timeout=5).decode('utf-8').strip() else: safe_cmd = command.replace("'", "'\\''") full_cmd = f"ssh -o LogLevel=ERROR -o BatchMode=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=3 {user}@{ip} '{safe_cmd}'" return subprocess.check_output(full_cmd, shell=True, stderr=subprocess.STDOUT, timeout=8).decode('utf-8').strip() except Exception: return "Err" def get_ping_icon(ip): print(f"DEBUG: Pinging {ip}...") # LOG PER CAPIRE DOVE SI BLOCCA try: # Timeout aggressivo: 0.5 secondi (-W 1 Γ¨ il minimo di ping standard, ma Python taglia a 0.8) subprocess.run( ["ping", "-c", "1", "-W", "1", ip], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=0.8, # Timeout Python brutale check=True ) return "βœ…" except subprocess.TimeoutExpired: return "πŸ”΄" # Timeout Python except subprocess.CalledProcessError: return "πŸ”΄" # Risposta "Host Unreachable" except Exception as e: print(f"Errore Ping {ip}: {e}") return "❓" def get_device_stats(device): ip, user, dtype = device['ip'], device['user'], device['type'] uptime_raw = run_cmd("uptime -p 2>/dev/null || uptime", ip, user) if not uptime_raw or "Err" in uptime_raw: return "πŸ”΄ **OFFLINE**" uptime = uptime_raw.replace("up ", "").split(", load")[0].split(", ")[0] temp = "N/A" if dtype in ["pi", "local"]: t = run_cmd("cat /sys/class/thermal/thermal_zone0/temp", ip, user) if t.isdigit(): temp = f"{int(t)/1000:.1f}Β°C" elif dtype == "nas": t = run_cmd("cat /sys/class/hwmon/hwmon0/temp1_input 2>/dev/null || cat /sys/class/thermal/thermal_zone0/temp", ip, user) if t.isdigit(): v = int(t); temp = f"{v/1000:.1f}Β°C" if v > 1000 else f"{v}Β°C" if dtype == "nas": ram_cmd = "free | grep Mem | awk '{printf \"%.0f%%\", $3*100/$2}'" else: ram_cmd = "free -m | awk 'NR==2{if ($2>0) printf \"%.0f%%\", $3*100/$2; else print \"0%\"}'" disk_path = "/" if dtype != "nas" else "/volume1" disk_cmd = f"df -h {disk_path} | awk 'NR==2{{print $5}}'" return f"βœ… **ONLINE**\n⏱️ Up: {uptime}\n🌑️ Temp: {temp} | 🧠 RAM: {run_cmd(ram_cmd, ip, user)} | πŸ’Ύ Disk: {run_cmd(disk_cmd, ip, user)}" def read_log_file(filepath, lines=15): if not os.path.exists(filepath): return "⚠️ File non trovato." try: return subprocess.check_output(['tail', '-n', str(lines), filepath], stderr=subprocess.STDOUT).decode('utf-8') except Exception as e: return f"Errore: {str(e)}" def run_speedtest(): try: return subprocess.check_output("speedtest-cli --simple", shell=True, timeout=50).decode('utf-8') except: return "Errore Speedtest" # --- BOT HANDLERS --- def restricted(func): @wraps(func) async def wrapped(update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs): if update.effective_user.id != OWNER_ID: return return await func(update, context, *args, **kwargs) return wrapped @restricted async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: keyboard = [ [InlineKeyboardButton("πŸ–₯️ Core Server", callback_data="menu_core"), InlineKeyboardButton("πŸ” Scan LAN", callback_data="menu_lan")], [InlineKeyboardButton("πŸ›‘οΈ Pi-hole", callback_data="menu_pihole"), InlineKeyboardButton("🌐 Rete", callback_data="menu_net")], [InlineKeyboardButton("πŸ“œ Logs", callback_data="menu_logs")] ] text = "πŸŽ› **Loogle Control Center v6.2**\nSeleziona un pannello:" if update.message: await update.message.reply_text(text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") else: await update.callback_query.edit_message_text(text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") @restricted async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() data = query.data if data == "main_menu": await start(update, context) elif data == "menu_core": keyboard = [] for i, dev in enumerate(CORE_DEVICES): keyboard.append([InlineKeyboardButton(dev['name'], callback_data=f"stat_{i}")]) keyboard.append([InlineKeyboardButton("πŸ“Š Report Completo", callback_data="stat_all")]) keyboard.append([InlineKeyboardButton("⬅️ Indietro", callback_data="main_menu")]) await query.edit_message_text("πŸ–₯️ **Core Servers**\nDettagli CPU/RAM/Temp:", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data == "stat_all": await query.edit_message_text("⏳ **Analisi Core Servers...**", parse_mode="Markdown") report = "πŸ“Š **REPORT CORE**\n" for dev in CORE_DEVICES: report += f"\nπŸ”Ή **{dev['name']}**\n{get_device_stats(dev)}\n" await query.edit_message_text(report, reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_core")]]), parse_mode="Markdown") elif data.startswith("stat_"): dev = CORE_DEVICES[int(data.split("_")[1])] await query.edit_message_text(f"⏳ Controllo {dev['name']}...", parse_mode="Markdown") await query.edit_message_text(f"πŸ”Ή **{dev['name']}**\n\n{get_device_stats(dev)}", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_core")]]), parse_mode="Markdown") elif data == "menu_lan": await query.edit_message_text("⏳ **Scansione LAN rapida...**", parse_mode="Markdown") report = "πŸ” **DIAGNOSTICA LAN**\n\n" try: # Core Devices for dev in CORE_DEVICES: report += f"{get_ping_icon(dev['ip'])} `{dev['ip']}` - {dev['name']}\n" report += "\n" # Infra Devices for dev in INFRA_DEVICES: report += f"{get_ping_icon(dev['ip'])} `{dev['ip']}` - {dev['name']}\n" except Exception as e: report += f"\n⚠️ Errore imprevisto durante scansione: {e}" keyboard = [ [InlineKeyboardButton("⚑ Menu Riavvio", callback_data="menu_reboot")], [InlineKeyboardButton("πŸ”„ Aggiorna", callback_data="menu_lan"), InlineKeyboardButton("⬅️ Indietro", callback_data="main_menu")] ] await query.edit_message_text(report, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data == "menu_reboot": keyboard = [] for i, dev in enumerate(INFRA_DEVICES): if "Router" not in dev['name']: keyboard.append([InlineKeyboardButton(f"⚑ {dev['name']}", callback_data=f"reboot_{i}")]) keyboard.append([InlineKeyboardButton("⬅️ Indietro", callback_data="menu_lan")]) await query.edit_message_text("⚠️ **RIAVVIO REMOTO**\nFunziona solo se il dispositivo supporta SSH e hai le chiavi.", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data.startswith("reboot_"): dev = INFRA_DEVICES[int(data.split("_")[1])] res = run_cmd("reboot", dev['ip'], "admin") await query.edit_message_text(f"⚑ Comando inviato a {dev['name']}...\n\nRisposta:\n`{res}`", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_reboot")]]), parse_mode="Markdown") # --- ALTRI MENU STANDARD --- elif data == "menu_pihole": status_raw = run_cmd("sudo pihole status", MASTER_IP, SSH_USER) icon = "βœ…" if "Enabled" in status_raw or "enabled" in status_raw else "πŸ”΄" text = f"πŸ›‘οΈ **Pi-hole Master**\nStato: {icon}\n\n`{status_raw}`" keyboard = [[InlineKeyboardButton("⏸️ 5m", callback_data="ph_disable_300"), InlineKeyboardButton("⏸️ 30m", callback_data="ph_disable_1800")], [InlineKeyboardButton("▢️ Attiva", callback_data="ph_enable"), InlineKeyboardButton("πŸ”„ Restart", callback_data="ph_restart")], [InlineKeyboardButton("⬅️ Indietro", callback_data="main_menu")]] await query.edit_message_text(text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data.startswith("ph_"): if "disable" in data: run_cmd(f"sudo pihole disable {data.split('_')[2]}s", MASTER_IP, SSH_USER) elif "enable" in data: run_cmd("sudo pihole enable", MASTER_IP, SSH_USER) elif "restart" in data: run_cmd("sudo systemctl restart pihole-FTL", MASTER_IP, SSH_USER) await query.edit_message_text("βœ… Comando inviato.", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_pihole")]]), parse_mode="Markdown") elif data == "menu_net": ip = run_cmd("curl -s ifconfig.me") keyboard = [[InlineKeyboardButton("πŸš€ Speedtest", callback_data="net_speedtest")], [InlineKeyboardButton("⬅️ Indietro", callback_data="main_menu")]] await query.edit_message_text(f"🌐 **Rete**\n🌍 IP: `{ip}`", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data == "net_speedtest": await query.edit_message_text("πŸš€ **Speedtest in corso...**", parse_mode="Markdown") res = run_speedtest() await query.edit_message_text(f"πŸš€ **Risultato:**\n\n`{res}`", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_net")]]), parse_mode="Markdown") elif data == "menu_logs": keyboard = [[InlineKeyboardButton("🐢 Watchdog", callback_data="log_wd"), InlineKeyboardButton("πŸ’Ύ Backup", callback_data="log_bk")], [InlineKeyboardButton("πŸ”„ NPM Sync", callback_data="log_npm"), InlineKeyboardButton("⬅️ Indietro", callback_data="main_menu")]] await query.edit_message_text("πŸ“œ **Logs**", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") elif data.startswith("log_"): paths = {"log_wd": "/logs/dhcp-watchdog.log", "log_bk": "/logs/raspiBackup.log", "log_npm": "/logs/sync-npm.log"} log_c = read_log_file(paths[data]) await query.edit_message_text(f"πŸ“œ **Log:**\n\n`{log_c}`", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅️ Indietro", callback_data="menu_logs")]]), parse_mode="Markdown") def main(): application = Application.builder().token(BOT_TOKEN).build() application.add_handler(CommandHandler("start", start)) application.add_handler(CallbackQueryHandler(button_handler)) application.run_polling() if __name__ == "__main__": main()