diff --git a/truenas_tools.py b/truenas_tools.py index 7d3c971..7b8d37e 100644 --- a/truenas_tools.py +++ b/truenas_tools.py @@ -1,180 +1,945 @@ -"""Read-only инструменты мониторинга TrueNAS v2.0.""" +""" +truenas_tools.py — мониторинг TrueNAS через Open WebUI. + +Полный набор инструментов для интеграции TrueNAS 25.10 (Goldeye/Goliath) +в Open WebUI в виде пользовательского Tool. + +Возможности: + • Системная информация (версия, uptime, CPU, load average, RAM, температура CPU) + • Детальная информация о RAM с перебором нескольких эндпоинтов API + • Пулы ZFS (статус, занятость, фрагментация) + • Диски (модель, серийник, объём, температура SMART) + • Docker-приложения (TrueNAS Apps) + • Виртуальные машины (libvirt/KVM) + • Контейнеры (virt/incus) + • Алерты TrueNAS + • Статистика ZFS ARC-кэша + • Диагностика соединения и валидация API-ключа + • Кэширование ответов с настраиваемым TTL + +История версий: + v2.0 — добавлены RAM (multi-endpoint), температуры CPU/дисков, ARC, валидация API + v1.0 — базовый мониторинг (система, пулы, диски, VM, контейнеры, алерты) +""" + import json import time import requests -from typing import Any, Optional +from typing import Dict, Optional, List, Any from pydantic import BaseModel, Field class Tools: - """Мониторинг TrueNAS через REST API. Изменяющие операции отсутствуют.""" + """ + Мониторинг 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="URL TrueNAS") - TRUENAS_API_KEY: str = Field(default="", description="API-ключ TrueNAS") - CACHE_TTL: int = Field(default=30, description="TTL кэша GET-запросов") - MEM_CACHE_TTL: int = Field(default=60, description="TTL кэша RAM") - verify_ssl: bool = Field(default=False, description="Проверять TLS") - request_timeout: int = Field(default=15, description="Таймаут запроса") + 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 = {} - self._last_error = "нет данных" + 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): + def _base(self) -> str: + """Всегда читаем актуальное значение из Valves — оно может меняться в UI после создания экземпляра Tools.""" return self.valves.TRUENAS_URL.rstrip("/") @property - def _headers(self): - return {"Authorization": f"Bearer {self.valves.TRUENAS_API_KEY}", "Content-Type": "application/json"} + 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 _get(self, key: str, endpoint: str, ttl: Optional[int] = None): - """Выполняет кэшируемый GET-запрос.""" - ttl = self.valves.CACHE_TTL if ttl is None else ttl + 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 key in self._cache and ttl > 0 and now - self._cache[key][0] < ttl: - return self._cache[key][1] + + 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(f"{self._base}/api/v2.0/{endpoint}", headers=self._headers, timeout=self.valves.request_timeout, verify=self.valves.verify_ssl) + r = requests.get( + url, + headers=self._headers, + timeout=self.valves.request_timeout, + verify=self.valves.verify_ssl, + ) if r.status_code == 200: - value = r.json() + data = r.json() + s = json.dumps(data, default=str) if ttl > 0: - self._cache[key] = (now, value) - return value + 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"{type(e).__name__}: {e}" - return None + self._last_error = f"Исключение на {endpoint}: {type(e).__name__}: {e}" + return None - def _first(self, name, endpoints): + def _try_endpoints(self, cache_key: str, endpoints: List[str]) -> str: + """ + Перебирает эндпоинты по очереди и возвращает первый удачный ответ. + """ for ep in endpoints: - value = self._get(f"{name}_{ep.replace('/', '_')}", ep) - if value is not None: - return value - return None + result = self._rest_get(f"{cache_key}_{ep.replace('/', '_')}", ep) + if result is not None: + return result + return f"⚠️ Данные недоступны. Попробованы пути: {', '.join(endpoints)}" - @staticmethod - def _fmt(value: Any) -> str: - if value is None: + # ----------------------------------------------------------------- + # Форматирование + # ----------------------------------------------------------------- + def _format_bytes(self, num_bytes: Any) -> str: + """Форматирует байты в человекочитаемый вид (KiB/MiB/GiB/TiB).""" + if num_bytes is None: return "Н/Д" try: - n = float(value) - for unit in ("B", "KiB", "MiB", "GiB", "TiB"): - if abs(n) < 1024: - return f"{n:.1f} {unit}" - n /= 1024 - return f"{n:.1f} PiB" + n = float(num_bytes) except (TypeError, ValueError): - return str(value) + 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.""" - try: - r = requests.get(f"{self._base}/api/v2.0/system/info", headers=self._headers, timeout=self.valves.request_timeout, verify=False) - return f"URL: {self._base}\nКод ответа: {r.status_code}\nПоследняя ошибка: {self._last_error}" - except Exception as e: - return f"URL: {self._base}\nОшибка: {type(e).__name__}: {e}" + """ + Диагностика соединения с 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: - """Проверяет read-only доступ API-ключа.""" + """ + Проверяет валидность API-ключа и доступность основных эндпоинтов. + Возвращает подробный отчёт с результатами проверки. + """ + lines = ["🔐 ВАЛИДАЦИЯ API-КЛЮЧА TRUENAS", ""] + if not self.valves.TRUENAS_API_KEY: - return "❌ API-ключ не задан" - checks = [("system/info", "система"), ("pool/query", "пулы"), ("disk/query", "диски"), ("app/query", "приложения"), ("alert/list", "алерты")] - out, ok = ["🔐 ВАЛИДАЦИЯ API-КЛЮЧА TRUENAS", ""], 0 - for ep, title in checks: - value = self._get(f"validate_{ep}", ep, 0) - if value is not None: - out.append(f"✅ {ep:15s} — {title}") - ok += 1 + 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: - out.append(f"⚠️ {ep:15s} — недоступен") - out.append(f"\nИтого: {ok}/{len(checks)}") - return "\n".join(out) + 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 через доступные API-эндпоинты.""" - data = self._first("memory", ["system/memory", "system/info", "system/get_info"]) - if not isinstance(data, dict): - return "💾 RAM: Н/Д (TrueNAS REST API не возвращает память в этой версии)" - total = data.get("total_memory") or data.get("memory_size") or data.get("memory") or data.get("total") - used = data.get("used") - return f"💾 RAM\n Всего: {self._fmt(total)}\n Использовано: {self._fmt(used)}" + """Отдельный эндпоинт: детальная информация о 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: - """Показывает системную информацию.""" - d = self._get("system", "system/info") - if not isinstance(d, dict): - return f"⚠️ Ошибка получения системы: {self._last_error}" - load = d.get("loadavg") or [] - return "\n".join([f"🖥️ TrueNAS: {d.get('version', 'N/A')}", f"🔧 Процессор: {d.get('model', 'N/A')}", f"📊 Load Average: {load}", self.get_memory_info()]) + 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: - """Показывает пулы ZFS.""" - pools = self._first("pools", ["pool/query", "pool", "storage/pool"]) - if not isinstance(pools, list): - return f"⚠️ Пулы недоступны: {self._last_error}" + 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: - size, free = p.get("size", 0), p.get("free", 0) - used = (size - free) / size * 100 if size else 0 - rows.append(f"{'✅' if p.get('healthy') else '⚠️'} '{p.get('name', '?')}' | {p.get('status', '?')} | {used:.1f}% занято") - return "\n".join(rows) or "Пулы не найдены." + 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: - """Показывает диски и температуры.""" - disks = self._first("disks", ["disk/query", "disk", "storage/disk"]) - if not isinstance(disks, list): - return f"⚠️ Диски недоступны: {self._last_error}" - return "\n".join(f"💾 {d.get('name', 'N/A')} | {d.get('model', '')} | {self._fmt(d.get('size'))} | {d.get('temperature', 'N/A')}°C" for d in disks) + 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: - """Показывает установленные приложения. Установка отсутствует.""" - apps = self._first("apps", ["app/query", "app", "chart/release/query"]) - if not isinstance(apps, list): - return f"⚠️ Приложения недоступны: {self._last_error}" - icons = {"RUNNING": "🟢", "STOPPED": "⚪", "DEPLOYING": "🟡", "CRASHED": "🔴"} - return "\n".join(f"{icons.get(str(a.get('state', a.get('status', 'N/A'))).upper(), '⚪')} {a.get('name', a.get('id', '?'))} | статус: {a.get('state', a.get('status', 'N/A'))} | версия: {a.get('human_version', a.get('version', 'N/A'))}" for a in apps) or "Приложения не установлены." - - def get_cpu_temperature(self) -> str: - """Показывает температуру CPU, если она есть в API.""" - d = self._get("temperature", "system/info") or {} - value = d.get("cpu_temperature") or d.get("cpu_temp") or d.get("temperature") - return f"🌡️ CPU: {value} °C" if value is not None else "🌡️ Температура CPU недоступна через API." - - def get_arc_stats(self) -> str: - """Показывает ZFS ARC, если он доступен через API.""" - d = self._first("arc", ["stats", "zfs/arc", "zfs/pool/stats", "pool/statistics"]) - return json.dumps(d, ensure_ascii=False, indent=2, default=str) if d else "📊 ARC-статистика недоступна через API." + """Список установленных приложений (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: - """Показывает виртуальные машины.""" - vms = self._first("vms", ["vm/query", "vm"]) - if not isinstance(vms, list): - return f"⚠️ VM недоступны: {self._last_error}" - return "\n".join(f"{v.get('name', '?')} | {v.get('status', 'N/A')}" for v in vms) or "Виртуальные машины не найдены." + """Список виртуальных машин (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: - """Показывает контейнеры virt/incus.""" - items = self._first("containers", ["virt/instance", "virt/instance/query", "container/query"]) - if not isinstance(items, list): - return f"⚠️ Контейнеры недоступны: {self._last_error}" - return "\n".join(f"{c.get('name', c.get('id', '?'))} | {c.get('status', c.get('state', 'N/A'))}" for c in items) or "Контейнеры не найдены." + """Лёгкие контейнеры / инстансы (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: - """Показывает активные алерты.""" - alerts = self._get("alerts", "alert/list") or [] + 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)] - return "✅ Все системы работают штатно." if not active else "\n".join(f"[{a.get('level', 'INFO')}] {a.get('formatted', a.get('message', '?'))}" for a in active) + 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: - """Формирует полный read-only отчёт.""" - sections = [("СИСТЕМА", self.get_system_status()), ("RAM", self.get_memory_info()), ("ТЕМПЕРАТУРА", self.get_cpu_temperature()), ("ARC", self.get_arc_stats()), ("ПУЛЫ", self.get_pools_status()), ("ДИСКИ", self.get_disks_temp_and_info()), ("ПРИЛОЖЕНИЯ", self.get_apps_status()), ("VM", self.get_vms_status()), ("КОНТЕЙНЕРЫ", self.get_containers_status()), ("АЛЕРТЫ", self.get_active_alerts())] - return "\n\n".join(f"--- {title} ---\n{content}" for title, content in sections) + 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)