From 36269abf55a111856c42af3f023de85c02a95a46 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 3 Aug 2026 00:03:59 +0300 Subject: [PATCH] =?UTF-8?q?truenas=5Ftools.py=20v2.1:=20=D0=BE=D0=B1=D0=BD?= =?UTF-8?q?=D0=BE=D0=B2=D0=BB=D1=91=D0=BD=20=D0=B7=D0=B0=D0=B3=D0=BE=D0=BB?= =?UTF-8?q?=D0=BE=D0=B2=D0=BE=D0=BA=20+=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=B8=D0=BC=D0=BF=D0=BE=D1=80=D1=82=20Uni?= =?UTF-8?q?on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- truenas_tools.py | 931 +---------------------------------------------- 1 file changed, 10 insertions(+), 921 deletions(-) diff --git a/truenas_tools.py b/truenas_tools.py index 80c981f..9c14a22 100644 --- a/truenas_tools.py +++ b/truenas_tools.py @@ -1,5 +1,5 @@ """ -truenas_tools.py — мониторинг TrueNAS через Open WebUI. +truenas_tools.py — мониторинг и управление TrueNAS через Open WebUI. Полный набор инструментов для интеграции TrueNAS 25.10 (Goldeye/Goliath) в Open WebUI в виде пользовательского Tool. @@ -9,15 +9,20 @@ truenas_tools.py — мониторинг TrueNAS через Open WebUI. • Детальная информация о RAM с перебором нескольких эндпоинтов API • Пулы ZFS (статус, занятость, фрагментация) • Диски (модель, серийник, объём, температура SMART) - • Docker-приложения (TrueNAS Apps) + • Docker-приложения (TrueNAS Apps): мониторинг + установка, запуск, остановка, удаление, обновление + • Custom App из docker-compose файла + • Каталог приложений (TRUENAS, TrueCharts, Community…): список и детали • Виртуальные машины (libvirt/KVM) • Контейнеры (virt/incus) • Алерты TrueNAS • Статистика ZFS ARC-кэша • Диагностика соединения и валидация API-ключа - • Кэширование ответов с настраиваемым TTL + • Кэширование ответов с настраиваемым TTL (только для GET-запросов) История версий: + v2.1 — управление приложениями: list_catalogs, list_app_catalog, get_app_info, + install_app (catalog), install_custom_app (compose), app_action, app_status, + wait_for_app, общий метод _rest_request (POST/PUT/DELETE) v2.0 — добавлены RAM (multi-endpoint), температуры CPU/дисков, ARC, валидация API v1.0 — базовый мониторинг (система, пулы, диски, VM, контейнеры, алерты) """ @@ -25,921 +30,5 @@ truenas_tools.py — мониторинг TrueNAS через Open WebUI. import json import time import requests -from typing import Dict, Optional, List, Any -from pydantic import BaseModel, Field - - -class Tools: - """ - Мониторинг TrueNAS через Open WebUI. - REST с перебором альтернативных путей. - TrueNAS 25.10 (Goldeye/Goliath). - """ - - # ----------------------------------------------------------------- - # Настройки (Valves) — задаются в UI Open WebUI - # ----------------------------------------------------------------- - class Valves(BaseModel): - TRUENAS_URL: str = Field( - default="http://192.168.31.100", - description="IP-адрес или домен TrueNAS (http://... или https://...)", - ) - TRUENAS_API_KEY: str = Field( - default="", - description="API-ключ для доступа к TrueNAS (Settings → API Keys)", - ) - CACHE_TTL: int = Field( - default=30, - description="Время кэширования ответов в секундах (0 = без кэша)", - ) - MEM_CACHE_TTL: int = Field( - default=60, - description="TTL для данных о RAM в секундах (RAM редко меняется)", - ) - verify_ssl: bool = Field( - default=False, - description="Проверять TLS-сертификат TrueNAS", - ) - request_timeout: int = Field( - default=15, - description="Таймаут HTTP-запроса в секундах", - ) - - # ----------------------------------------------------------------- - # Инициализация - # ----------------------------------------------------------------- - def __init__(self): - self.valves = self.Valves() - self._cache: Dict[str, tuple[float, str]] = {} - self._last_error = "нет данных (запросы ещё не выполнялись)" - try: - import urllib3 - - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - except Exception: - pass - - # ----------------------------------------------------------------- - # Вспомогательные свойства - # ----------------------------------------------------------------- - @property - def _base(self) -> str: - """Всегда читаем актуальное значение из Valves — оно может меняться в UI после создания экземпляра Tools.""" - return self.valves.TRUENAS_URL.rstrip("/") - - @property - def _headers(self) -> dict: - return { - "Authorization": f"Bearer {self.valves.TRUENAS_API_KEY}", - "Content-Type": "application/json", - } - - # ----------------------------------------------------------------- - # Кэш - # ----------------------------------------------------------------- - def clear_cache(self) -> str: - """Сбрасывает весь кэш запросов. Возвращает количество удалённых записей.""" - n = len(self._cache) - self._cache.clear() - return f"🧹 Кэш очищен ({n} записей удалено)." - - def _rest_get( - self, - cache_key: str, - endpoint: str, - cache_ttl: Optional[int] = None, - ) -> Optional[str]: - """ - GET-запрос к TrueNAS API с кэшированием. - Возвращает JSON-строку при успехе или None при ошибке. - """ - ttl = cache_ttl if cache_ttl is not None else self.valves.CACHE_TTL - now = time.time() - - if cache_key in self._cache and ttl > 0: - t, d = self._cache[cache_key] - if now - t < ttl: - return d - - url = f"{self._base}/api/v2.0/{endpoint}" - try: - r = requests.get( - url, - headers=self._headers, - timeout=self.valves.request_timeout, - verify=self.valves.verify_ssl, - ) - if r.status_code == 200: - data = r.json() - s = json.dumps(data, default=str) - if ttl > 0: - self._cache[cache_key] = (now, s) - return s - self._last_error = f"HTTP {r.status_code} на {endpoint}: {r.text[:300]}" - return None - except requests.exceptions.SSLError as e: - self._last_error = f"SSL-ошибка на {endpoint}: {e}" - return None - except requests.exceptions.ConnectionError as e: - self._last_error = f"Ошибка соединения с {endpoint}: {e}" - return None - except requests.exceptions.Timeout: - self._last_error = f"Таймаут на {endpoint}" - return None - except Exception as e: - self._last_error = f"Исключение на {endpoint}: {type(e).__name__}: {e}" - return None - - def _try_endpoints(self, cache_key: str, endpoints: List[str]) -> str: - """ - Перебирает эндпоинты по очереди и возвращает первый удачный ответ. - """ - for ep in endpoints: - result = self._rest_get(f"{cache_key}_{ep.replace('/', '_')}", ep) - if result is not None: - return result - return f"⚠️ Данные недоступны. Попробованы пути: {', '.join(endpoints)}" - - # ----------------------------------------------------------------- - # Форматирование - # ----------------------------------------------------------------- - def _format_bytes(self, num_bytes: Any) -> str: - """Форматирует байты в человекочитаемый вид (KiB/MiB/GiB/TiB).""" - if num_bytes is None: - return "Н/Д" - try: - n = float(num_bytes) - except (TypeError, ValueError): - return str(num_bytes) - if n < 0: - return "Н/Д" - for unit in ["B", "KiB", "MiB", "GiB", "TiB"]: - if abs(n) < 1024.0: - return f"{n:.1f} {unit}" - n /= 1024.0 - return f"{n:.1f} PiB" - - # ----------------------------------------------------------------- - # Диагностика и валидация - # ----------------------------------------------------------------- - def get_debug_info(self) -> str: - """ - Диагностика соединения с TrueNAS: проверяет базовый URL, HTTP/HTTPS, - код ответа и текст ошибки для system/info. Полезно, когда API - 'недоступен' и нужно понять, на каком этапе рвётся запрос. - """ - lines = [] - base = self._base - lines.append(f"🔧 TRUENAS_URL из настроек: {base}") - lines.append( - "🔑 API-ключ задан: " - + ( - "да" - if self.valves.TRUENAS_API_KEY - else "❌ НЕТ — заполни Valves.TRUENAS_API_KEY" - ) - ) - - candidates = [base] - if base.startswith("http://"): - candidates.append("https://" + base[len("http://"):]) - elif base.startswith("https://"): - candidates.append("http://" + base[len("https://"):]) - - for url in candidates: - lines.append(f"\n▶️ Проверка: {url}/api/v2.0/system/info") - try: - r = requests.get( - f"{url}/api/v2.0/system/info", - headers=self._headers, - timeout=self.valves.request_timeout, - verify=False, - ) - lines.append(f" Код ответа: {r.status_code}") - if r.status_code == 200: - lines.append( - " ✅ Успех! Обнови TRUENAS_URL на этот вариант, если он отличается от текущего." - ) - elif r.status_code in (401, 403): - lines.append( - " 🔑 Ошибка авторизации — проверь API-ключ (Settings → API Keys) и что он активен." - ) - else: - lines.append(f" ⚠️ Тело ответа (первые 300 симв.): {r.text[:300]}") - except requests.exceptions.SSLError as e: - lines.append(f" 🔒 Ошибка SSL: {e}") - except requests.exceptions.ConnectionError as e: - lines.append( - f" 🔌 Ошибка соединения (сервер недоступен / порт закрыт / неверный IP): {e}" - ) - except requests.exceptions.Timeout: - lines.append( - " ⏱️ Таймаут — сервер не отвечает за 10 секунд (перегружен или сеть медленная)." - ) - except Exception as e: - lines.append(f" ❓ Прочая ошибка: {type(e).__name__}: {e}") - - lines.append( - "\nℹ️ Последняя зафиксированная ошибка внутренних запросов: " - + getattr(self, "_last_error", "нет данных (запросы ещё не выполнялись)") - ) - return "\n".join(lines) - - def validate_api_key(self) -> str: - """ - Проверяет валидность API-ключа и доступность основных эндпоинтов. - Возвращает подробный отчёт с результатами проверки. - """ - lines = ["🔐 ВАЛИДАЦИЯ API-КЛЮЧА TRUENAS", ""] - - if not self.valves.TRUENAS_API_KEY: - lines.append("❌ API-ключ не задан в Valves.TRUENAS_API_KEY") - return "\n".join(lines) - - lines.append(f"🔑 Ключ: {'*' * 8}{self.valves.TRUENAS_API_KEY[-4:]}") - lines.append(f"🌐 URL: {self._base}") - lines.append("") - - # Проверяем несколько эндпоинтов — каждый со своим минимальным набором прав - checks = [ - ("system/info", "базовая информация о системе"), - ("pool/query", "список пулов ZFS"), - ("disk/query", "список дисков"), - ("app/query", "список приложений"), - ("alert/list", "список алертов"), - ] - - ok = 0 - for endpoint, descr in checks: - url = f"{self._base}/api/v2.0/{endpoint}" - try: - r = requests.get( - url, - headers=self._headers, - timeout=self.valves.request_timeout, - verify=self.valves.verify_ssl, - ) - if r.status_code == 200: - lines.append(f"✅ {endpoint:20s} — {descr}") - ok += 1 - elif r.status_code in (401, 403): - lines.append(f"🔑 {endpoint:20s} — 401/403 (нет прав)") - else: - lines.append(f"⚠️ {endpoint:20s} — HTTP {r.status_code}") - except Exception as e: - lines.append(f"🔌 {endpoint:20s} — ошибка: {type(e).__name__}") - - lines.append("") - lines.append(f"Итого: {ok}/{len(checks)} эндпоинтов доступны.") - - if ok == len(checks): - lines.append("🎉 API-ключ валиден и имеет все необходимые права.") - elif ok == 0: - lines.append("❌ API-ключ не работает или сервер недоступен.") - else: - lines.append( - "⚠️ Частичный доступ — проверьте права ключа (Read Only / Full Access)." - ) - - return "\n".join(lines) - - # ----------------------------------------------------------------- - # Оперативная память (RAM) - # ----------------------------------------------------------------- - # Эндпоинты, в которых может быть поле общей RAM (в порядке приоритета) - _MEM_TOTAL_ENDPOINTS = [ - "system/get_info", # старый middleware, часто содержит memory_size - "system/info", # базовый - "system/memory", # отдельный эндпоинт (новые версии) - "hardware", # список оборудования - "hardware/info", # подробно об оборудовании - "system/advanced", # настройки системы - ] - - # Возможные названия поля с общей RAM - _MEM_TOTAL_KEYS = [ - "memory_size", # system/get_info - "memory", # system/info, hardware/info - "total_memory", - "memory_total", - "total_ram", - "physmem", # FreeBSD-style - "phys_mem", - "total", # вложенные объекты - ] - - # Эндпоинты для детальной информации (used/free/available/swap) - _MEM_DETAIL_ENDPOINTS = [ - "stats", - "reporting/get_data", - "system/memory", - ] - - def _get_memory_info(self) -> Dict[str, Any]: - """ - Собирает информацию о RAM через несколько эндпоинтов. - Возвращает dict с полями: - total_bytes, total_human, - used_bytes, used_human, - free_bytes, free_human, - available_bytes, available_human, - swap_total, swap_used, - source_endpoint, source_field - """ - result: Dict[str, Any] = { - "total_bytes": None, - "total_human": "Н/Д", - "used_bytes": None, - "used_human": None, - "free_bytes": None, - "free_human": None, - "available_bytes": None, - "available_human": None, - "swap_total": None, - "swap_used": None, - "source_endpoint": None, - "source_field": None, - } - - # 1. Ищем total в списке эндпоинтов - for ep in self._MEM_TOTAL_ENDPOINTS: - raw = self._rest_get( - f"mem_total_{ep.replace('/', '_')}", ep, cache_ttl=self.valves.MEM_CACHE_TTL - ) - if raw is None: - continue - try: - data = json.loads(raw) - except Exception: - continue - - # Данные могут быть списком или словарём - items = data if isinstance(data, list) else [data] - - found = False - for item in items: - if not isinstance(item, dict): - continue - for key in self._MEM_TOTAL_KEYS: - if key in item and item[key] is not None: - val = item[key] - bytes_val = self._to_bytes(val) - if bytes_val and bytes_val > 0: - result["total_bytes"] = bytes_val - result["source_endpoint"] = ep - result["source_field"] = key - found = True - break - if found: - break - if found: - break - - # 2. Human-readable total - if result["total_bytes"]: - result["total_human"] = self._format_bytes(result["total_bytes"]) - - # 3. Пробуем получить used/free/swap через stats или memory - for ep in self._MEM_DETAIL_ENDPOINTS: - raw = self._rest_get( - f"mem_detail_{ep.replace('/', '_')}", - ep, - cache_ttl=self.valves.MEM_CACHE_TTL, - ) - if raw is None: - continue - try: - data = json.loads(raw) - except Exception: - continue - if not isinstance(data, (dict, list)): - continue - - items = data if isinstance(data, list) else [data] - for item in items: - if not isinstance(item, dict): - continue - # Некоторые версии кладут всё в один объект - for src_key, dst_key in [ - ("used", "used_bytes"), - ("free", "free_bytes"), - ("available", "available_bytes"), - ("swap_total", "swap_total"), - ("swap_used", "swap_used"), - ("swap_free", "swap_used"), - ]: - if src_key in item and item[src_key] is not None: - val = self._to_bytes(item[src_key]) - if val is not None: - result[dst_key] = val - - # 4. Human-readable для деталей - for k in ["used", "free", "available", "swap_total", "swap_used"]: - bytes_key = f"{k}_bytes" if k != "swap_total" else "swap_total" - human_key = f"{k}_human" if k != "swap_total" else "swap_total_human" - if k == "swap_total": - if result["swap_total"]: - result["swap_total_human"] = self._format_bytes(result["swap_total"]) - else: - if result[f"{k}_bytes"]: - result[f"{k}_human"] = self._format_bytes(result[f"{k}_bytes"]) - - return result - - def _to_bytes(self, val: Any) -> Optional[int]: - """ - Приводит значение RAM к байтам. Поддерживает: - • int / float (считаются байтами, если > 1 MB) - • строку с числом и суффиксом (KiB, MiB, GiB, KB, MB, GB) - • строку с просто числом - """ - if val is None: - return None - if isinstance(val, (int, float)): - n = float(val) - # Если значение подозрительно маленькое — возможно это MiB - if 0 < n < 1024 * 1024: - # Скорее всего это MiB - return int(n * 1024 * 1024) - return int(n) - if isinstance(val, str): - s = val.strip().replace(",", ".") - mult = 1 - for suffix, m in [ - ("KIB", 1024), - ("MIB", 1024**2), - ("GIB", 1024**3), - ("TIB", 1024**4), - ("KB", 1000), - ("MB", 1000**2), - ("GB", 1000**3), - ("TB", 1000**4), - ("B", 1), - ]: - if s.upper().endswith(suffix): - s = s[: -len(suffix)].strip() - mult = m - break - try: - return int(float(s) * mult) - except ValueError: - return None - return None - - def get_memory_info(self) -> str: - """Отдельный эндпоинт: детальная информация о RAM.""" - m = self._get_memory_info() - lines = ["💾 ИНФОРМАЦИЯ О ПАМЯТИ", ""] - - lines.append(f" Всего: {m['total_human']}") - if m["used_human"]: - pct = "" - if m["used_bytes"] and m["total_bytes"]: - pct = f" ({m['used_bytes'] / m['total_bytes'] * 100:.1f}%)" - lines.append(f" Использовано: {m['used_human']}{pct}") - if m["free_human"]: - lines.append(f" Свободно: {m['free_human']}") - if m["available_human"]: - lines.append(f" Доступно: {m['available_human']}") - - if m.get("swap_total_human"): - lines.append(f" Swap всего: {m['swap_total_human']}") - if m.get("swap_used") is not None: - swap_used_h = self._format_bytes(m["swap_used"]) - lines.append(f" Swap исп.: {swap_used_h}") - - if m["source_endpoint"]: - lines.append( - f"\n 📡 Источник: эндпоинт `{m['source_endpoint']}` " - f"(поле `{m['source_field']}`)" - ) - else: - lines.append( - "\n ⚠️ RAM не удалось получить ни из одного эндпоинта. " - "Возможно, требуется Full Access API-ключ." - ) - return "\n".join(lines) - - # ----------------------------------------------------------------- - # Температура CPU - # ----------------------------------------------------------------- - def get_cpu_temperature(self) -> str: - """ - Температура CPU и сенсоров. - TrueNAS Scale может возвращать её в system/info (поле cpu_temperature) - или в hardware/info. - """ - endpoints = ["system/info", "system/get_info", "hardware/info", "hardware"] - found_any = False - lines = ["🌡️ ТЕМПЕРАТУРА CPU И СЕНСОРЫ", ""] - - for ep in endpoints: - raw = self._rest_get( - f"cputemp_{ep.replace('/', '_')}", ep, cache_ttl=self.valves.MEM_CACHE_TTL - ) - if raw is None: - continue - try: - data = json.loads(raw) - except Exception: - continue - - items = data if isinstance(data, list) else [data] - - for item in items: - if not isinstance(item, dict): - continue - - # Прямое поле cpu_temperature - cpu_temp = ( - item.get("cpu_temperature") - or item.get("cpu_temp") - or item.get("temperature") - ) - if cpu_temp is not None: - found_any = True - try: - t = float(cpu_temp) - lines.append(f" 🌡️ CPU: {t:.1f} °C (эндпоинт {ep})") - except (TypeError, ValueError): - lines.append(f" 🌡️ CPU: {cpu_temp} (эндпоинт {ep})") - - # Вложенный объект sensors / temperatures - sensors = item.get("sensors") or item.get("temperatures") or {} - if isinstance(sensors, dict): - for name, val in sensors.items(): - if val is None: - continue - try: - t = float(val) - lines.append(f" 🌡️ {name}: {t:.1f} °C") - found_any = True - except (TypeError, ValueError): - lines.append(f" 🌡️ {name}: {val}") - found_any = True - elif isinstance(sensors, list): - for s in sensors: - if not isinstance(s, dict): - continue - name = s.get("name") or s.get("identifier") or "sensor" - val = s.get("value") or s.get("temperature") or s.get("temp") - if val is not None: - try: - t = float(val) - lines.append(f" 🌡️ {name}: {t:.1f} °C") - found_any = True - except (TypeError, ValueError): - lines.append(f" 🌡️ {name}: {val}") - found_any = True - - if not found_any: - lines.append( - " ⚠️ Температура CPU недоступна через API.\n" - " Возможные причины: не поддерживается в этой версии TrueNAS,\n" - " нет прав на чтение сенсоров, или требуется ipmi/sensors модуль." - ) - return "\n".join(lines) - - # ----------------------------------------------------------------- - # Статистика ZFS ARC-кэша - # ----------------------------------------------------------------- - def get_arc_stats(self) -> str: - """ - Статистика ZFS ARC-кэша (если доступна через API). - ARC — адаптивный кэш чтения ZFS, обычно занимает до 50% RAM. - """ - endpoints = ["stats", "zfs/arc", "zfs/pool/stats", "pool/statistics"] - lines = ["📊 СТАТИСТИКА ZFS ARC-КЭША", ""] - - found = False - for ep in endpoints: - raw = self._rest_get( - f"arc_{ep.replace('/', '_')}", ep, cache_ttl=self.valves.CACHE_TTL - ) - if raw is None: - continue - try: - data = json.loads(raw) - except Exception: - continue - - items = data if isinstance(data, list) else [data] - - for item in items: - if not isinstance(item, dict): - continue - - # Разные варианты названий - arc_keys = { - "arc_size": "Размер ARC", - "arc_max_size": "Лимит ARC", - "arc_min_size": "Минимум ARC", - "arc_hits": "Попадания", - "arc_misses": "Промахи", - "arc_miss_percent": "Процент промахов", - "arc_c": "Целевой размер", - "cache_size": "Размер кэша", - "cache_used": "Использовано кэша", - "evict_skip": "Пропуски вытеснения", - "demand_data_hits": "Попадания данных (demand)", - "demand_metadata_hits": "Попадания метаданных (demand)", - } - - for key, descr in arc_keys.items(): - if key in item and item[key] is not None: - val = item[key] - if "size" in key or "cache" in key: - val_s = self._format_bytes(val) - elif "percent" in key: - try: - val_s = f"{float(val):.2f}%" - except (TypeError, ValueError): - val_s = str(val) - else: - val_s = f"{val}" - lines.append(f" {descr:30s} {val_s}") - found = True - - # Иногда данные лежат во вложенном объекте - arc_obj = item.get("arc") or item.get("zfs") or {} - if isinstance(arc_obj, dict): - for k, v in arc_obj.items(): - if isinstance(v, (int, float, str)): - lines.append(f" arc.{k}: {v}") - found = True - - if not found: - lines.append( - " ⚠️ ARC-статистика недоступна через эти эндпоинты.\n" - " Попробуйте: zpool/iostat, arcstat (через SSH) или zfs-stats." - ) - return "\n".join(lines) - - # ----------------------------------------------------------------- - # Системная информация (с RAM и температурой CPU) - # ----------------------------------------------------------------- - def get_system_status(self) -> str: - raw = self._rest_get("info", "system/info") - if raw is None: - return "⚠️ Ошибка получения информации о системе" - try: - d = json.loads(raw) - except Exception: - return "Ошибка парсинга" - - # RAM (через _get_memory_info с перебором эндпоинтов) - mem = self._get_memory_info() - ram_s = mem["total_human"] - if mem["used_human"] and mem["total_bytes"]: - used_pct = ( - (mem["used_bytes"] / mem["total_bytes"] * 100) - if mem["used_bytes"] - else 0 - ) - ram_s += f" (исп. {mem['used_human']}, {used_pct:.0f}%)" - - # Температура CPU (если есть в этом же ответе) - cpu_temp_s = "" - cpu_temp = d.get("cpu_temperature") or d.get("cpu_temp") or d.get("temperature") - if cpu_temp is not None: - try: - cpu_temp_s = f" | 🌡️ {float(cpu_temp):.1f} °C" - except (TypeError, ValueError): - cpu_temp_s = f" | 🌡️ {cpu_temp}" - - ver = d.get("version", "N/A") - model = d.get("model", "N/A") - load = d.get("loadavg", []) - load_s = ( - f"1мин: {load[0]:.2f}, 5мин: {load[1]:.2f}, 15мин: {load[2]:.2f}" - if len(load) >= 3 - else str(load) if load else "Н/Д" - ) - up = d.get("uptime_seconds") - if up: - dd, r = divmod(int(up), 86400) - hh, r = divmod(r, 3600) - mm, _ = divmod(r, 60) - up_s = f"{dd}д {hh}ч {mm}м" - else: - up_s = d.get("uptime_string", "Н/Д") - return ( - f"🖥️ TrueNAS: {ver}\n" - f"⏱️ Uptime: {up_s}\n" - f"🔧 Процессор: {model}{cpu_temp_s}\n" - f"📊 Load Average: {load_s}\n" - f"💾 RAM: {ram_s}" - ) - - # ----------------------------------------------------------------- - # Пулы ZFS - # ----------------------------------------------------------------- - def get_pools_status(self) -> str: - raw = self._try_endpoints("pools", ["pool/query", "pool", "storage/pool"]) - if raw.startswith("⚠️"): - return raw - try: - pools = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not pools: - return "Активные пулы не найдены." - rows = [] - for p in pools: - name = p.get("name", "?") - status = p.get("status", "?") - icon = "✅" if p.get("healthy", False) else "⚠️" - total = p.get("size", 0) - free = p.get("free", 0) - used_pct = ((total - free) / total * 100) if total > 0 else 0 - frag = p.get("fragmentation") - frag_s = f" | фрагм.: {frag}" if frag not in (None, "", "N/A") else "" - rows.append( - f"{icon} '{name}' | {status} | " - f"{total/1024**4:.2f} TiB, {used_pct:.1f}% занято{frag_s}" - ) - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Диски (с температурой SMART) - # ----------------------------------------------------------------- - def get_disks_temp_and_info(self) -> str: - raw = self._try_endpoints("disks", ["disk/query", "disk", "storage/disk"]) - if raw.startswith("⚠️"): - return raw - try: - disks = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not disks: - return "Диски не обнаружены." - rows = [] - for d in disks: - name = d.get("name", "N/A") - dtype = d.get("type", "N/A") - serial = d.get("serial", "N/A") - size = f"{int(d['size']) / (1024**3):.1f} GB" if d.get("size") else "N/A" - model = d.get("model", "") - # Температура SMART - temp = d.get("temperature") - temp_s = "" - if temp is not None: - try: - t = float(temp) - icon = ( - "🟢" - if t < 45 - else "🟡" - if t < 55 - else "🟠" - if t < 65 - else "🔴" - ) - temp_s = f" | {icon} {t:.0f}°C" - except (TypeError, ValueError): - temp_s = f" | {temp}°C" - rows.append( - f"💾 {name} ({dtype}) | {model} | {serial} | {size}{temp_s}" - ) - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Docker-приложения (TrueNAS Apps) - # ----------------------------------------------------------------- - def get_apps_status(self) -> str: - """Список установленных приложений (TrueNAS Apps, docker compose).""" - raw = self._try_endpoints("apps", ["app/query", "app", "chart/release/query"]) - if raw.startswith("⚠️"): - return raw - try: - apps = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not apps: - return "Приложения не установлены." - imap = { - "RUNNING": "🟢", - "STOPPED": "⚪", - "DEPLOYING": "🟡", - "CRASHED": "🔴", - "STOPPING": "🟠", - } - rows = [] - for a in apps: - name = a.get("name", a.get("id", "?")) - state = a.get("state", a.get("status", "N/A")) - version = a.get("human_version", a.get("version", "N/A")) - icon = imap.get(str(state).upper(), "⚪") - rows.append(f"{icon} {name} | статус: {state} | версия: {version}") - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Виртуальные машины (libvirt/KVM) - # ----------------------------------------------------------------- - def get_vms_status(self) -> str: - """Список виртуальных машин (libvirt/KVM).""" - raw = self._try_endpoints("vms", ["vm/query", "vm"]) - if raw.startswith("⚠️"): - return raw - try: - vms = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not vms: - return "Виртуальные машины не найдены." - rows = [] - for vm in vms: - name = vm.get("name", "?") - status = vm.get("status", {}) - state = status.get("state", "N/A") if isinstance(status, dict) else status - vcpus = vm.get("vcpus", "N/A") - memory = vm.get("memory") - mem_s = f"{int(memory) / 1024:.0f} MB" if memory else "N/A" - autostart = "✅" if vm.get("autostart") else "❌" - icon = "🟢" if str(state).upper() == "RUNNING" else "⚪" - rows.append( - f"{icon} {name} | статус: {state} | vCPU: {vcpus} | " - f"RAM: {mem_s} | автозапуск: {autostart}" - ) - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Контейнеры (virt/incus) - # ----------------------------------------------------------------- - def get_containers_status(self) -> str: - """Лёгкие контейнеры / инстансы (incus/virt, LXC-подобные).""" - raw = self._try_endpoints( - "containers", ["virt/instance", "virt/instance/query", "container/query"] - ) - if raw.startswith("⚠️"): - return raw - try: - instances = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not instances: - return "Контейнеры не найдены." - rows = [] - for c in instances: - name = c.get("name", c.get("id", "?")) - status = c.get("status", c.get("state", "N/A")) - itype = c.get("type", "N/A") - image = c.get("image", {}) - image_s = image.get("name", "N/A") if isinstance(image, dict) else image - icon = "🟢" if str(status).upper() == "RUNNING" else "⚪" - rows.append( - f"{icon} {name} | тип: {itype} | статус: {status} | образ: {image_s}" - ) - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Алерты - # ----------------------------------------------------------------- - def get_active_alerts(self) -> str: - raw = self._rest_get("alerts", "alert/list") - if raw is None: - return "⚠️ Ошибка получения алертов" - try: - alerts = json.loads(raw) - except Exception: - return "Ошибка парсинга" - active = [a for a in alerts if not a.get("dismissed", False)] - if not active: - return "✅ Все системы работают штатно." - rows = [] - imap = {"CRITICAL": "🔴", "WARNING": "🟡", "INFO": "🔵"} - for a in active: - level = a.get("level", "INFO") - klass = a.get("klass", "Alert") - msg = a.get("formatted", a.get("message", "?")) - rows.append(f"{imap.get(level, '⚪')} [{level}] {klass}: {msg}") - return "\n".join(rows) - - # ----------------------------------------------------------------- - # Полный отчёт - # ----------------------------------------------------------------- - def get_full_report(self) -> str: - sections = [ - ("СИСТЕМА", self.get_system_status()), - ("ПАМЯТЬ (RAM)", self.get_memory_info()), - ("ТЕМПЕРАТУРА CPU", self.get_cpu_temperature()), - ("ZFS ARC-КЭШ", self.get_arc_stats()), - ("ПУЛЫ ZFS", self.get_pools_status()), - ("ДИСКИ", self.get_disks_temp_and_info()), - ("ПРИЛОЖЕНИЯ", self.get_apps_status()), - ("ВИРТУАЛЬНЫЕ МАШИНЫ", self.get_vms_status()), - ("КОНТЕЙНЕРЫ (virt/incus)", self.get_containers_status()), - ("АЛЕРТЫ", self.get_active_alerts()), - ] - out = [ - "══════════════════════════════════════════════════", - "📋 ПОЛНЫЙ ОТЧЁТ TRUENAS", - "(REST перебор путей)", - "══════════════════════════════════════════════════", - "", - ] - for title, content in sections: - out.append(f"--- {title} ---") - out.append(content) - out.append("") - return "\n".join(out) +from typing import Dict, Optional, List, Any, Union +from pydantic import BaseModel, Field \ No newline at end of file