import os # Импортируем модуль os для выполнения системных команд в Linux import sys # Импортируем sys для работы с системными переменными import gc # Импортируем gc (Garbage Collector) для принудительной очистки ОЗУ на 4GB системе import time # Импортируем time для вычисления аптайма import psutil # Импортируем psutil для сбора метрик процессора, памяти и диска import asyncio # Импортируем asyncio для поддержки асинхронной работы скрипта import platform # Импортируем platform для определения версии ядра Linux import datetime # Импортируем datetime для форматирования времени import concurrent.futures # Импортируем пул потоков для вынесения блокирующих системных вызовов из основного цикла import subprocess # Импортируем subprocess для запуска фоновых процессов системы from aiogram import Bot, Dispatcher, types, F # Импортируем ключевые инструменты библиотеки aiogram 3.x from aiogram.filters import Command # Импортируем фильтр команд для обработки /start и /open_server from aiogram.fsm.context import FSMContext # Импортируем контекст машины состояний (FSM) from aiogram.fsm.state import State, StatesGroup # Импортируем классы для создания шагов диалога from aiogram.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardBuilder # Импортируем конструкторы для инлайн и реплай кнопок # Было: # from tplink_router_api import TPLinkRouterApi # Импортируем официальный класс для работы с API роутеров TP-Link # Стало: from tplinkrouterc6u import TplinkRouter # Импортируем класс для работы с API роутеров TP-Link # --- НАСТРОЙКИ БЕЗОПАСНОСТИ --- BOT_TOKEN = "8807352870:AAHoKDDWNGyP4urfSSp2IlGkFDGAIu7kYzg" # Твой полный рабочий токен от BotFather ADMIN_ID = 5050706507 # Твой личный Telegram ID (только ты сможешь управлять сервером!) LOCAL_SUBNET = "192.168.1" # Три первых октета твоей домашней сети для сканирования (кто дома) UPLOAD_DIR = "./uploads" # Папка на сервере, куда бот будет сохранять присланные тобой файлы # Инициализируем объекты бота и главного диспетчера событий bot = Bot(token=BOT_TOKEN) # Создаем объект бота с токеном доступа dp = Dispatcher() # Создаем диспетчер для обработки всех входящих событий START_TIME = time.time() # Фиксируем время запуска бота для статистики # --- МАШИНА СОСТОЯНИЙ (FSM) --- class ServerStates(StatesGroup): # Группа состояний для машины переходов waiting_for_command = State() # Ожидание ввода команды ZSH waiting_for_filepath = State() # Ожидание ввода пути для скачивания файла с сервера waiting_for_upload = State() # Ожидание отправки файла для заливки на сервер def check_auth(user_id: int) -> bool: # Функция авторизации return user_id == ADMIN_ID # Сравниваем ID отправителя с ID администратора def get_uptime() -> str: # Функция получения аптайма ОС uptime_seconds = time.time() - psutil.boot_time() # Получаем разницу во времени старта и текущим моментом return str(datetime.timedelta(seconds=int(uptime_seconds))) # Возвращаем форматированную строку времени def get_cpu_temp() -> str: # Получение текущей температуры процессора try: # Начало блока обработки ошибок temps = psutil.sensors_temperatures() # Опрашиваем доступные температурные датчики устройства if 'coretemp' in temps: # Если есть датчик Intel Core return f"{temps['coretemp'][0].current}°C" # Возвращаем температуру первого ядра elif 'acpitz' in temps: # Если есть датчик материнской платы ACPI return f"{temps['acpitz'][0].current}°C" # Возвращаем температуру датчика платы with open("/sys/class/thermal/thermal_zone0/temp", "r") as f: # Пробуем получить данные напрямую из Linux temp = float(f.read().strip()) / 1000.0 # Конвертируем значение в градусы Цельсия return f"{temp:.1f}°C" # Округляем до десятых долей градуса except Exception: # Если произошла непредвиденная ошибка получения температуры return "N/A" # Возвращаем статус недоступности датчика def get_updates_count() -> str: # Функция проверки пакетов обновлений ОС try: # Начало отлова системных ошибок res = subprocess.run("apt-get -s upgrade | grep -P '^\\d+ (улучшено|upgraded)'", shell=True, capture_output=True, text=True) # Имитируем обновление пакетов APT out = res.stdout.strip() # Вычищаем лишние переносы и пробелы из потока вывода if out: # Если обновления действительно найдены в выводе команды return out # Возвращаем форматированное количество пакетов return "Система обновлена" # Если обновлений не требуется except Exception: # Если менеджер пакетов заблокирован или недоступен return "Не удалось проверить" # Возвращаем текстовую ошибку # --- ГЕНЕРАТОРЫ КЛАВИАТУР --- def create_menu_keyboard(): # Генератор главного инлайн-меню бота builder = InlineKeyboardBuilder() # Создаем новый инлайн-конструктор кнопок builder.button(text="📊 Метрики и Статус", callback_data="status") # Добавляем кнопку вывода характеристик builder.button(text="💻 Терминал (ZSH)", callback_data="terminal") # Добавляем кнопку запуска терминала builder.button(text="📥 Скачать файл", callback_data="download_file") # Добавляем кнопку скачивания файлов builder.button(text="📤 Закинуть файл", callback_data="upload_file") # Добавляем кнопку заливки файлов на сервер builder.button(text="🏠 Кто у меня дома?", callback_data="ping_network") # Добавляем кнопку сканирования роутера builder.button(text="🔄 Reboot сервера", callback_data="reboot_confirm") # Добавляем кнопку перезагрузки системы builder.adjust(2, 2, 1, 1) # Задаем сетку размещения кнопок в чате return builder.as_markup() # Конвертируем конструктор в разметку кнопок def create_cancel_reply_keyboard(): # Генератор реплай-кнопки отмены для FSM переходов builder = ReplyKeyboardBuilder() # Создаем конструктор реплай-кнопок builder.button(text="❌ Отменить действие") # Кнопка отмены любого диалога return builder.as_markup(resize_keyboard=True, one_time_keyboard=True) # Возвращаем клавиатуру, которая сжимается под экран def create_ping_reply_keyboard(): # Генератор реплай-кнопок для интерфейса сканера сети builder = ReplyKeyboardBuilder() # Инициализируем реплай-конструктор builder.button(text="🔄 Пересканировать") # Кнопка повторного сканирования пингом builder.button(text="↩️ Вернуться в главное меню") # Кнопка быстрого возврата на главный экран builder.adjust(1, 1) # Размещаем кнопки друг под другом return builder.as_markup(resize_keyboard=True) # Возвращаем подстроенную по размеру клавиатуру # --- ОБРАБОТЧИКИ КОМАНД --- @dp.message(Command("start")) # Вешаем хэндлер на команду /start @dp.message(Command("open_server")) # Дублируем на команду /open_server async def cmd_start_or_open(message: types.Message, state: FSMContext): # Главная функция входа в бота if not check_auth(message.from_user.id): # Если ID пользователя не прошел проверку доступа await message.answer("❌ Доступ заблокирован. Этот сервер принадлежит 𝖒ими_𝖒𝖎𝖓𝖓𝖆𝖗>_<.") # Отказываем во входе return # Прерываем дальнейшее выполнение await state.clear() # Принудительно чистим любые висящие состояния FSM welcome_text = ( # Создаем красивое стартовое приветствие "🟢 **Сервер Asus E410K онлайн и готов к работе!**\n\n" # Текст статуса сервера "Выбери необходимое действие на интерактивной панели ниже:" # Подсказка по меню ) # Завершаем инициализацию строки await message.answer(welcome_text, parse_mode="Markdown", reply_markup=create_menu_keyboard()) # Отправляем меню в ответ @dp.message(F.text == "❌ Отменить действие") # Ловим текстовую кнопку отмены на любом этапе @dp.message(F.text == "↩️ Вернуться в главное меню") # Ловим текстовую кнопку возврата на главный экран async def process_cancel_to_menu(message: types.Message, state: FSMContext): # Функция отмены и возврата await state.clear() # Полностью сбрасываем состояние диалога бота await message.answer( # Пишем об успешном сбросе состояния и перенаправлении text="↩️ **Возвращаю тебя в главное меню:**", # Пишем сообщение parse_mode="Markdown", # Подключаем разметку текста Markdown reply_markup=types.ReplyKeyboardRemove() # Принудительно скрываем текстовую реплай-панель клавиатуры ) # Завершаем отправку await message.answer( # Отправляем свежее инлайн-меню действий text="🟢 Выбери необходимое действие на панели:", # Текст меню reply_markup=create_menu_keyboard() # Прикрепляем инлайн-клавиатуру ) # Завершаем отправку # --- ОБРАБОТЧИК КНОПКИ МОНИТОРИНГА --- @dp.callback_query(F.data == "status") # Обрабатываем клик по инлайн-кнопке метрик async def process_status(callback_query: types.CallbackQuery): # Функция обновления метрик if not check_auth(callback_query.from_user.id): # Проверяем авторизацию админа await callback_query.answer("Доступ запрещен!", show_alert=True) # Выводим всплывающее окно запрета return # Выходим cpu_usage = psutil.cpu_percent(interval=0.3) # Снимаем нагрузку CPU за короткий промежуток ram = psutil.virtual_memory() # Запрашиваем состояние ОЗУ сервера disk = psutil.disk_usage('/') # Запрашиваем утилизацию системного диска / temp = get_cpu_temp() # Вызываем функцию опроса тепловых датчиков процессора uptime = get_uptime() # Вызываем расчет непрерывной работы ОС kernel_version = platform.release() # Получаем версию активного ядра Linux updates = get_updates_count() # Инициализируем опрос обновлений системы через пакетный менеджер status_text = ( # Формируем красивую сводную таблицу статистики сервера f"🖥️ **МОНИТОРИНГ СЕРВЕРА (Asus E410K):**\n" # Название устройства f"━━━━━━━━━━━━━━━━━━━━\n" # Разделитель строки f"🐧 **Ядро Linux:** `{kernel_version}`\n" # Строка ядра ОС f"📦 **Обновления:** `{updates}`\n" # Наличие обновлений f"🌡️ **Температура CPU:** {temp}\n" # Значение датчика тепла f"📈 **Загрузка CPU:** {cpu_usage}%\n" # Нагрузка в процентах f"💾 **ОЗУ:** {ram.percent}% ({ram.used // (1024**2)}MB / {ram.total // (1024**2)}MB)\n" # Использование памяти f"💿 **Диск /:** {disk.percent}% ({disk.free // (1024**3)}GB свободно)\n" # Память диска f"⏱️ **Uptime системы:** {uptime}\n" # Время непрерывной работы f"━━━━━━━━━━━━━━━━━━━━\n" # Разделитель f"🕒 _Обновлено в реальном времени: {datetime.datetime.now().strftime('%H:%M:%S')}_" # Точное время апдейта сообщения ) # Завершаем конкатенацию строки try: # Защита на случай, если текст сообщения не изменился в процессе обновления await callback_query.message.edit_text( # Пытаемся заменить текст инлайна text=status_text, # Передаем обновленный текст метрик parse_mode="Markdown", # Подключаем разметку reply_markup=create_menu_keyboard() # Прикрепляем главное меню обратно ) # Завершаем редактирование except Exception: # Перехватываем исключение отсутствия изменений pass # Просто гасим ошибку без вывода в консоль await callback_query.answer() # Убираем колесо бесконечной анимации загрузки на кнопке # --- РАБОТА С ТЕРМИНАЛОМ (ZSH) --- @dp.callback_query(F.data == "terminal") # Обрабатываем кнопку перехода в режим консоли ZSH async def process_terminal_click(callback_query: types.CallbackQuery, state: FSMContext): # Функция перехода в терминал if not check_auth(callback_query.from_user.id): # Валидируем права пользователя await callback_query.answer("Доступ запрещен!", show_alert=True) # Пишем отказ return # Выходим await callback_query.message.answer( # Информируем о включении режима терминала text="⌨️ **Введи команду для выполнения в ZSH:**\n_(Например: ls -la / или pm2 status)_", # Текст-подсказка parse_mode="Markdown", # Подключаем парсер markdown reply_markup=create_cancel_reply_keyboard() # Выводим кнопку отмены действия ) # Завершаем отправку сообщения await state.set_state(ServerStates.waiting_for_command) # Назначаем состояние ожидания ввода консольной команды await callback_query.answer() # Гасим анимацию кнопки @dp.message(ServerStates.waiting_for_command) # Обрабатываем ввод команды во время активного состояния FSM async def execute_zsh_command(message: types.Message, state: FSMContext): # Функция отправки команды в ZSH if not check_auth(message.from_user.id): # Валидируем права на сервер return # Выходим command = message.text # Запоминаем текст присланной пользователем команды status_msg = await message.answer(f"⏳ **Выполняю через ZSH:** `{command}`...") # Выводим сообщение о старте процесса try: # Открываем защищенный блок выполнения системных задач process = await asyncio.create_subprocess_shell( # Асинхронно запускаем консоль ZSH command, # Передаем саму команду stdout=asyncio.subprocess.PIPE, # Захватываем поток вывода данных stderr=asyncio.subprocess.PIPE, # Захватываем поток вывода ошибок executable='/bin/zsh' # Указываем выполнять строго в ZSH ) # Конец запуска # Исправлено: ждем выполнение команды 15 секунд через асинхронный wait_for stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=15) # Ограничиваем выполнение по времени output = stdout.decode('utf-8', errors='ignore') # Раскодируем стандартный системный вывод в UTF-8 error_output = stderr.decode('utf-8', errors='ignore') # Раскодируем вывод ошибок в UTF-8 response_text = "✅ **Результат выполнения (ZSH):**\n\n" # Пишем шапку ответа if output: # Если получили данные в стандартном выводе response_text += f"```text\n{output[:3500]}\n```" # Ограничиваем длину текста, чтобы не выйти за лимит Telegram if error_output: # Если получили текст системной ошибки response_text += f"\n⚠️ **Ошибки:**\n```text\n{error_output[:1000]}\n```" # Прикрепляем логи ошибок if not output and not error_output: # Если команда выполнилась молча response_text += "_Команда выполнена успешно, вывод пуст._" # Оповещаем пользователя except asyncio.TimeoutError: # Обрабатываем прерывание по таймауту выполнения response_text = "❌ **Ошибка:** Превышено время ожидания команды (Таймаут 15 сек)." # Текст ошибки таймаута except Exception as e: # Обрабатываем другие системные поломки response_text = f"❌ **Системная ошибка:** {str(e)}" # Записываем системный трейсбек в строку await message.answer( # Выводим результат и прячем клавиатуру отмены text=response_text, # Отправляем лог работы терминала parse_mode="Markdown", # Парсим markdown reply_markup=types.ReplyKeyboardRemove() # Скрываем кнопку отмены действия ) # Завершаем отправку результата await message.answer("🟢 Действие завершено. Что делаем дальше?", reply_markup=create_menu_keyboard()) # Возвращаем главное инлайн-меню await state.clear() # Полностью очищаем состояние FSM переходов gc.collect() # Вызываем чистку неиспользуемой оперативной памяти # --- СКАЧИВАНИЕ ФАЙЛОВ С СЕРВЕРА НА СМАРТФОН --- @dp.callback_query(F.data == "download_file") # Обрабатываем инлайн-кнопку скачивания async def process_download_click(callback_query: types.CallbackQuery, state: FSMContext): # Функция запроса файла if not check_auth(callback_query.from_user.id): # Проверка доступа к файловой системе await callback_query.answer("Доступ запрещен!", show_alert=True) # Запрещаем return # Выход await callback_query.message.answer( # Спрашиваем путь к файлу text="📂 **Введи абсолютный путь к файлу на сервере, который нужно скачать:**\n_(Пример: /home/minnar/main.py)_", # Текст-подсказка parse_mode="Markdown", # Разметка текста reply_markup=create_cancel_reply_keyboard() # Показываем кнопку отмены действия ) # Конец сообщения await state.set_state(ServerStates.waiting_for_filepath) # Переводим бота в режим ожидания строки пути await callback_query.answer() # Гасим клик инлайн-кнопки @dp.message(ServerStates.waiting_for_filepath) # Обрабатываем введенную строку пути async def send_file_to_user(message: types.Message, state: FSMContext): # Функция отправки файла в чат if not check_auth(message.from_user.id): # Валидация прав return # Выходим filepath = message.text.strip() # Вычищаем случайные лишние пробелы по краям строки if os.path.exists(filepath) and os.path.isfile(filepath): # Проверяем, существует ли файл и не папка ли это await message.answer("📤 **Отправляю файл...**", reply_markup=types.ReplyKeyboardRemove()) # Убираем клавиатуру отмены и информируем о старте try: # Открываем блок работы с отправкой файлов file_to_send = types.FSInputFile(filepath) # Подготавливаем файл для передачи асинхронным потоком await message.answer_document(file_to_send, reply_markup=create_menu_keyboard()) # Отправляем документ и возвращаем главное меню except Exception as e: # Обрабатываем возможный сбой при отправке await message.answer(f"❌ Ошибка отправки: {str(e)}", reply_markup=create_menu_keyboard()) # Оповещаем и возвращаем главное меню else: # Если файла нет на диске или указан неверный путь await message.answer("❌ Файл не найден по указанному пути! Попробуй еще раз или отмени команду.", reply_markup=create_cancel_reply_keyboard()) # Сообщаем об ошибке пути return # Оставляем состояние FSM активным, чтобы дать возможность ввести корректный путь еще раз await state.clear() # Сбрасываем состояние FSM только при успешной отправке файла gc.collect() # Проводим сборщик мусора в Python для высвобождения памяти # --- ЗАГРУЗКА ФАЙЛОВ ИЗ ТЕЛЕГРАМА НА СЕРВЕР (ФИКС: ОТДЕЛЬНАЯ КНОПКА) --- @dp.callback_query(F.data == "upload_file") # Обработчик новой кнопки загрузки файлов на сервер async def process_upload_click(callback_query: types.CallbackQuery, state: FSMContext): # Функция запроса файла от пользователя if not check_auth(callback_query.from_user.id): # Валидируем права администратора await callback_query.answer("Доступ запрещен!", show_alert=True) # Отклоняем клик return # Выход await callback_query.message.answer( # Отправляем сообщение-запрос документа в чат text="📥 **Пришли мне любой файл (документ) для загрузки на сервер:**\n_(Он сохранится в папку: " + UPLOAD_DIR + ")_", # Текст-подсказка parse_mode="Markdown", # Разметка текста reply_markup=create_cancel_reply_keyboard() # Выводим кнопку отмены действия в реплай панели ) # Конец вывода await state.set_state(ServerStates.waiting_for_upload) # Включаем специальное состояние ожидания загрузки файла await callback_query.answer() # Гасим анимацию инлайн-кнопки @dp.message(ServerStates.waiting_for_upload, F.document) # Обрабатываем присланный документ во время активного ожидания async def handle_incoming_document(message: types.Message, state: FSMContext): # Функция сохранения документа на сервер if not check_auth(message.from_user.id): # Проверка прав администратора return # Выход document = message.document # Считываем объект документа из сообщения file_name = document.file_name # Достаем оригинальное имя файла if not os.path.exists(UPLOAD_DIR): # Если указанной директории для загрузок еще нет на сервере os.makedirs(UPLOAD_DIR) # Создаем директорию рекурсивно destination_path = os.path.join(UPLOAD_DIR, file_name) # Формируем итоговый путь записи файла status_msg = await message.answer(f"📥 **Получаю файл:** `{file_name}`...", reply_markup=types.ReplyKeyboardRemove()) # Оповещаем пользователя и гасим кнопки try: # Открываем блок защищенной потоковой записи файла на диск file_info = await bot.get_file(document.file_id) # Запрашиваем путь к системному кэшу файла у серверов Telegram await bot.download_file( # Потоково скачиваем файл без утилизации ОЗУ целиком (оптимально под 4GB) file_path=file_info.file_path, # Задаем путь в облаке destination=destination_path, # Указываем конечную точку сохранения на сервере chunk_size=65536 # Задаем жесткий размер чанка в 64 КБ ) # Конец скачивания success_text = ( # Текст при успешном сохранении файла f"✅ **Файл успешно сохранен на сервере!**\n\n" # Статус успеха f"📁 **Имя:** `{file_name}`\n" # Выводим имя f"📍 **Путь:** `{os.path.abspath(destination_path)}`\n" # Абсолютный путь расположения на диске f"📦 **Размер:** `{round(document.file_size / 1024, 2)} KB`" # Размер в килобайтах ) # Конец строки успеха await status_msg.answer(success_text, parse_mode="Markdown", reply_markup=create_menu_keyboard()) # Отправляем сообщение и возвращаем инлайн-меню except Exception as e: # Обрабатываем ошибки записи/скачивания await status_msg.answer(f"❌ **Ошибка при сохранении файла:** `{str(e)}`", parse_mode="Markdown", reply_markup=create_menu_keyboard()) # Пишем об ошибке finally: # Всегда выполняем завершающий блок await state.clear() # Очищаем состояние машины переходов gc.collect() # Проводим принудительный сбор неиспользуемой оперативной памяти @dp.message(ServerStates.waiting_for_upload) # Защита от отправки текста вместо документа в режиме заливки файла async def handle_invalid_upload_format(message: types.Message): # Обрабатываем ошибочный формат данных if not check_auth(message.from_user.id): # Проверяем авторизацию администратора return # Выходим await message.answer( # Требуем отправить именно файл text="⚠️ **Пожалуйста, отправь именно файл (документ), а не обычное текстовое сообщение!**", # Пишем ошибку формата reply_markup=create_cancel_reply_keyboard() # Оставляем кнопку отмены на экране ) # Завершаем сообщение # --- СКАНЕР СЕТИ WI-FI («КТО ДОМА?») --- @dp.callback_query(F.data == "ping_network") # Обработчик инлайн-кнопки сканирования локальной сети def ping_single_host(ip: str) -> str or None: # Создаем чистую синхронную функцию для проверки одного конкретного IP-адреса try: # Открываем блок отлова системных ошибок для изоляции процесса res = subprocess.run( # Синхронно запускаем бинарник ping в изолированном подпроцессе операционной системы ["ping", "-c", "1", "-W", "1", ip], # Передаем аргументы: 1 пакет, жесткий таймаут ожидания ответа 1 секунда stdout=subprocess.DEVNULL, # Полностью глушим стандартный вывод утилиты в черную дыру stderr=subprocess.DEVNULL # Полностью подавляем вывод любых системных ошибок в консоль ) # Завершаем конфигурацию и выполнение команды ping if res.returncode == 0: # Если код возврата системы равен нулю, значит хост ответил return ip # Возвращаем живой IP-адрес для последующего добавления в финальный отчет except Exception: # Если во время вызова подпроцесса произошел непредвиденный сбой pass # Молча игнорируем ошибку выполнения утилиты ping return None # Возвращаем None, если устройство не пингуется или произошло исключение async def scan_local_network_smart(subnet: str): # Главная асинхронная функция сканирования домашней локальной сети active_ips = [] # Инициализируем пустой массив для накопления ответивших IP-адресов устройств host_list = [f"{subnet}.{i}" for i in range(1, 101)] # Генерируем список из 100 целевых IP-адресов внутри твоей подсети # Запускаем параллельное выполнение пингов в пуле потоков, чтобы Asus E410K не зависал на асинхронных пайпах with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: # Выделяем ровно 20 независимых потоков под сетевые задачи loop = asyncio.get_event_loop() # Получаем текущий запущенный асинхронный цикл событий aiogram promises = [ # Начинаем формировать пул асинхронных обещаний для каждого потока loop.run_in_executor(executor, ping_single_host, ip) # Планируем синхронный пинг хоста в отдельном потоке for ip in host_list # Итерируемся по всему сгенерированному списку из 100 IP-адресов ] # Завершаем сборку массива потоковых задач results = await asyncio.gather(*promises) # Асинхронно дожидаемся завершения работы всех потоков без блокировки бота active_ips = [ip for ip in results if ip is not None] # Фильтруем массив, оставляя только успешно пинганувшиеся IP result_text = "🏠 **АКТИВНЫЕ УСТРОЙСТВА В СЕТИ:**\n━━━━━━━━━━━━━━━━━━━━\n" # Формируем стартовую шапку отчета if active_ips: # Если в результате потокового сканирования удалось найти хоть один живой хост for ip in sorted(active_ips, key=lambda x: int(x.split('.')[-1])): # Сортируем адреса по последнему октету для красоты note = " 🌐 (Роутер)" if ip.endswith(".1") else "" # Добавляем маркер шлюза для самого первого адреса подсети result_text += f"🟢 Device IP: `{ip}`{note}\n" # Конкатенируем отформатированную строку устройства в общий текст else: # Если абсолютно все 100 хостов проигнорировали запросы или подсеть указана неверно result_text += f"❌ Активных устройств в сети `{subnet}.X` не обнаружено.\n_Проверь LOCAL_SUBNET в настройках кода!_" # Добавляем ошибку с указанием текущей подсети result_text += "\n━━━━━━━━━━━━━━━━━━━━" # Добавляем нижнюю декоративную черту-разделитель в конец сообщения return result_text # Возвращаем полностью готовый текст отчета в вызывающий хэндлер @dp.callback_query(F.data == "ping_network") # Ловим нажатие на инлайн-кнопку сканирования сети на интерактивной панели async def process_ping_network(callback_query: types.CallbackQuery): # Объявляем функцию обработки клика по инлайн-кнопке if not check_auth(callback_query.from_user.id): # Проверяем ID пользователя на соответствие твои правам администратора await callback_query.answer("Доступ запрещен!", show_alert=True) # Показываем системный алерт в случае несовпадения ID return # Немедленно выходим из хэндлера, блокируя выполнение кода await callback_query.answer() # Сразу гасим анимацию загрузки Loading на кнопке, чтобы она не зависала в приложении status_msg = await callback_query.message.answer( # Отправляем новое сообщение, информирующее о начале сетевого анализа text="🔍 **Сканирую локальную сеть в многопоточном режиме...**", # Информационный текст для вывода на экран reply_markup=create_ping_reply_keyboard() # Прикрепляем нижние реплай-кнопки управления сканером ) # Завершаем отправку промежуточного статуса report = await scan_local_network_smart(LOCAL_SUBNET) # Передаем LOCAL_SUBNET в потоковый сканер и ждем результат await status_msg.edit_text(report, parse_mode="Markdown") # Меняем текст промежуточного статуса на готовый финальный отчет gc.collect() # Вызываем сборщик мусора Python для утилизации отработавших потоков и очистки оперативной памяти @dp.message(F.text == "🔄 Пересканировать") # Ловим текстовую команду со встроенной реплай-кнопки повторного анализа async def process_re_ping_network(message: types.Message): # Функция быстрого перезапуска сканера прямо из диалога чата if not check_auth(message.from_user.id): # Проверяем уровень доступа текущего отправителя сообщения к серверу return # Моментально прерываем выполнение функции, если пишет посторонний юзер status_msg = await message.answer("🔍 **Повторно сканирую локальную сеть в многопоточном режиме...**") # Выводим статус перезапуска задачи report = await scan_local_network_smart(LOCAL_SUBNET) # Повторно вызываем потоковый сканер подсети await status_msg.edit_text(report, parse_mode="Markdown") # Редактируем сообщение, выводя свежий список устройств в чат gc.collect() # Очищаем О # --- МЕНЮ ПЕРЕЗАГРУЗКИ ASUS --- @dp.callback_query(F.data == "reboot_confirm") # Ловим инлайн-кнопку подтверждения перезагрузки Asus async def process_reboot_confirm(callback_query: types.CallbackQuery): # Функция меню безопасности перезагрузки if not check_auth(callback_query.from_user.id): # Проверяем права администратора await callback_query.answer("Доступ запрещен!", show_alert=True) # Блокируем доступ return # Выход reboot_builder = InlineKeyboardBuilder() # Инициализируем новый инлайн конструктор для клавиатур reboot_builder.button(text="⚠️ ДА, перезагрузить!", callback_data="reboot_execute") # Кнопка опасной перезагрузки reboot_builder.button(text="❌ Отмена", callback_data="status") # Кнопка безопасного возврата к мониторингу reboot_builder.adjust(1) # Настраиваем кнопки строго друг под другом await callback_query.message.edit_text( # Переписываем инлайн-сообщение в чате на меню подтверждения text="❓ **Вы действительно хотите полностью перезагрузить сервер Asus E410K?**", # Текст предупреждения parse_mode="Markdown", # Подключаем разметку текста reply_markup=reboot_builder.as_markup() # Прикрепляем новую клавиатуру перезагрузки ) # Конец редактирования await callback_query.answer() # Гасим анимацию кнопки @dp.callback_query(F.data == "reboot_execute") # Ловим подтверждение перезагрузки сервера async def process_reboot_execute(callback_query: types.CallbackQuery): # Функция системного ребута if not check_auth(callback_query.from_user.id): # Дополнительная валидация прав админа await callback_query.answer("Доступ запрещен!", show_alert=True) # Запрещаем ребут return # Выходим await callback_query.message.edit_text( # Редактируем сообщение перед отключением бота text="🔄 **Выполняю перезагрузку сервера...**\nБот вернется в онлайн примерно через минуту.", # Текст ребута сервера parse_mode="Markdown" # Включаем парсинг ) # Завершаем редактирование await bot.session.close() # Безопасно завершаем текущую асинхронную сессию бота перед падением ОС os.system("sudo reboot") # Выполняем жесткую команду перезагрузки операционной системы Linux # --- СТАРТ БОТА --- async def main(): # Главная функция асинхронного запуска бота await bot.delete_webhook(drop_pending_updates=True) # Принудительно чистим кэш пропущенных обновлений перед пуском await dp.start_polling(bot) # Запускаем бесконечный асинхронный цикл поллинга (опроса серверов) if __name__ == "__main__": # Проверка точки входа в скрипт try: # Открываем блок отлова прерываний asyncio.run(main()) # Запускаем асинхронный движок except KeyboardInterrupt: # Ловим ручную остановку скрипта пользователем (Ctrl + C) print("Бот успешно остановлен пользователем.") # Пишем отладочное сообщение в лог терминала