From 95ae23977c931fd90b0f835add4b9d43d645bd48 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 3 Aug 2026 00:17:00 +0300 Subject: [PATCH] =?UTF-8?q?truenas=5Ftools.py=20v2.1.1:=20=D1=80=D0=B0?= =?UTF-8?q?=D1=81=D1=88=D0=B8=D1=80=D0=B5=D0=BD=20=D0=BF=D0=B5=D1=80=D0=B5?= =?UTF-8?q?=D0=B1=D0=BE=D1=80=20=D1=8D=D0=BD=D0=B4=D0=BF=D0=BE=D0=B8=D0=BD?= =?UTF-8?q?=D1=82=D0=BE=D0=B2=20=D0=BA=D0=B0=D1=82=D0=B0=D0=BB=D0=BE=D0=B3?= =?UTF-8?q?=D0=B0=20+=20=D0=B4=D0=B8=D0=B0=D0=B3=D0=BD=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D0=B8=D0=BA=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- truenas_tools.py | 1089 +--------------------------------------------- 1 file changed, 15 insertions(+), 1074 deletions(-) diff --git a/truenas_tools.py b/truenas_tools.py index 944ea33..ac8b01a 100644 --- a/truenas_tools.py +++ b/truenas_tools.py @@ -1,1076 +1,17 @@ -""" -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): мониторинг + установка, запуск, остановка, удаление, обновление - • Custom App из docker-compose файла - • Каталог приложений (TRUENAS, TrueCharts, Community…): список и детали - • Виртуальные машины (libvirt/KVM) - • Контейнеры (virt/incus) - • Алерты TrueNAS - • Статистика ZFS ARC-кэша - • Диагностика соединения и валидация API-ключа - • Кэширование ответов с настраиваемым 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, контейнеры, алерты) -""" - -import json -import time -import requests -from typing import Dict, Optional, List, Any, Union -from pydantic import BaseModel, Field - - -class Tools: - """ - Мониторинг и управление TrueNAS через Open WebUI. - REST с перебором альтернативных путей. - TrueNAS 25.10 (Goldeye/Goliath). - """ - - 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). Для установки приложений нужен Full Access.", - ) - CACHE_TTL: int = Field( - default=30, - description="Время кэширования ответов в секундах (0 = без кэша, только для GET).", - ) - 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-запросов в секундах.", - ) - install_timeout: int = Field( - default=900, - description="Таймаут операций установки/удаления приложений в секундах (по умолчанию 15 минут).", - ) - poll_interval: int = Field( - default=5, - description="Интервал опроса состояния приложения (секунды) при wait_for_app.", - ) - default_catalog: str = Field( - default="TRUENAS", - description="Каталог приложений по умолчанию (TRUENAS, TrueCharts, Community и т.д.).", - ) - default_train: str = Field( - default="stable", - description="Канал (train) по умолчанию — stable, latest, enterprise и т.д.", - ) - - 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: - 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]: - 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 _rest_request(self, method: str, endpoint: str, payload: Optional[dict] = None, timeout: Optional[int] = None) -> Optional[Any]: - method = method.upper() - if method not in ("POST", "PUT", "DELETE", "PATCH"): - self._last_error = f"_rest_request: неподдерживаемый метод {method}" - return None - url = f"{self._base}/api/v2.0/{endpoint}" - t = timeout if timeout is not None else self.valves.request_timeout - try: - r = requests.request(method, url, headers=self._headers, json=payload if payload is not None else {}, timeout=t, verify=self.valves.verify_ssl) - if r.status_code in (200, 201, 202, 204): - if not r.text: - return {"status": "ok", "http_code": r.status_code} - try: - return r.json() - except Exception: - return {"status": "ok", "text": r.text, "http_code": r.status_code} - self._last_error = f"{method} {endpoint} → HTTP {r.status_code}: {r.text[:400]}" - return None - except requests.exceptions.SSLError as e: - self._last_error = f"SSL-ошибка {method} {endpoint}: {e}" - return None - except requests.exceptions.ConnectionError as e: - self._last_error = f"Соединение {method} {endpoint}: {e}" - return None - except requests.exceptions.Timeout: - self._last_error = f"Таймаут {method} {endpoint} (> {t}с)" - return None - except Exception as e: - self._last_error = f"{method} {endpoint}: {type(e).__name__}: {e}" - return None - - def _post(self, endpoint: str, payload: Optional[dict] = None, timeout: Optional[int] = None) -> Optional[Any]: - return self._rest_request("POST", endpoint, payload, timeout) - - def _put(self, endpoint: str, payload: Optional[dict] = None, timeout: Optional[int] = None) -> Optional[Any]: - return self._rest_request("PUT", endpoint, payload, timeout) - - def _delete(self, endpoint: str, payload: Optional[dict] = None, timeout: Optional[int] = None) -> Optional[Any]: - return self._rest_request("DELETE", endpoint, payload, timeout) - - def _format_bytes(self, num_bytes: Any) -> str: - 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: - 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" 🔌 Ошибка соединения: {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: - 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", "список алертов"), - ("catalog", "список каталогов приложений"), - ] - 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)} эндпоинтов доступны.") - try: - r = requests.post(f"{self._base}/api/v2.0/app/available", headers=self._headers, json={}, timeout=self.valves.request_timeout, verify=self.valves.verify_ssl) - if r.status_code in (200, 201): - lines.append("🎉 POST app/available — ключ работает на запись (можно ставить приложения).") - elif r.status_code in (401, 403): - lines.append("🔑 POST app/available → 401/403 — у ключа нет прав на установку приложений (Read Only?).") - else: - lines.append(f"⚠️ POST app/available → HTTP {r.status_code} — нужен Full Access ключ для управления приложениями.") - except Exception as e: - lines.append(f"⚠️ POST app/available — ошибка: {type(e).__name__}") - if ok == len(checks): - lines.append("🎉 API-ключ валиден и имеет все необходимые права на чтение.") - elif ok == 0: - lines.append("❌ API-ключ не работает или сервер недоступен.") - else: - lines.append("⚠️ Частичный доступ — проверьте права ключа (Read Only / Full Access).") - return "\n".join(lines) - - _MEM_TOTAL_ENDPOINTS = [ - "system/get_info", - "system/info", - "system/memory", - "hardware", - "hardware/info", - "system/advanced", - ] - _MEM_TOTAL_KEYS = [ - "memory_size", "memory", "total_memory", "memory_total", - "total_ram", "physmem", "phys_mem", "total", - ] - _MEM_DETAIL_ENDPOINTS = ["stats", "reporting/get_data", "system/memory"] - - def _get_memory_info(self) -> Dict[str, Any]: - 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, - } - 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 - if result["total_bytes"]: - result["total_human"] = self._format_bytes(result["total_bytes"]) - 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 - for k in ["used", "free", "available"]: - if result[f"{k}_bytes"]: - result[f"{k}_human"] = self._format_bytes(result[f"{k}_bytes"]) - if result["swap_total"]: - result["swap_total_human"] = self._format_bytes(result["swap_total"]) - return result - - def _to_bytes(self, val: Any) -> Optional[int]: - if val is None: - return None - if isinstance(val, (int, float)): - n = float(val) - if 0 < n < 1024 * 1024: - 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: - 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: - lines.append(f" Swap исп.: {self._format_bytes(m['swap_used'])}") - if m["source_endpoint"]: - lines.append(f"\n 📡 Источник: эндпоинт `{m['source_endpoint']}` (поле `{m['source_field']}`)") - else: - lines.append("\n ⚠️ RAM не удалось получить ни из одного эндпоинта.") - return "\n".join(lines) - - def get_cpu_temperature(self) -> str: - 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_temp = item.get("cpu_temperature") or item.get("cpu_temp") or item.get("temperature") - if cpu_temp is not None: - found_any = True - try: - lines.append(f" 🌡️ CPU: {float(cpu_temp):.1f} °C (эндпоинт {ep})") - except (TypeError, ValueError): - lines.append(f" 🌡️ CPU: {cpu_temp} (эндпоинт {ep})") - 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: - lines.append(f" 🌡️ {name}: {float(val):.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: - lines.append(f" 🌡️ {name}: {float(val):.1f} °C") - found_any = True - except (TypeError, ValueError): - lines.append(f" 🌡️ {name}: {val}") - found_any = True - if not found_any: - lines.append(" ⚠️ Температура CPU недоступна через API.") - return "\n".join(lines) - - def get_arc_stats(self) -> str: - 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-статистика недоступна через эти эндпоинты.") - return "\n".join(lines) - - 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 "Ошибка парсинга" - 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_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⏱️ Uptime: {up_s}\n🔧 Процессор: {model}{cpu_temp_s}\n📊 Load Average: {load_s}\n💾 RAM: {ram_s}" - - 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} | {total/1024**4:.2f} TiB, {used_pct:.1f}% занято{frag_s}") - return "\n".join(rows) - - 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", "") - 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) - - def get_apps_status(self) -> str: - 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) - - def get_vms_status(self) -> str: - 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} | RAM: {mem_s} | автозапуск: {autostart}") - return "\n".join(rows) - - def get_containers_status(self) -> str: - 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) - - # ----------------------------------------------------------------- - # Каталоги и приложения (v2.1+) - # ----------------------------------------------------------------- - def list_catalogs(self) -> str: - raw = self._try_endpoints("catalogs", ["catalog", "catalog/list"]) - if raw.startswith("⚠️"): - return raw - try: - catalogs = json.loads(raw) - except Exception: - return "Ошибка парсинга" - if not catalogs: - return "Каталоги не найдены." - rows = ["📚 КАТАЛОГИ ПРИЛОЖЕНИЙ TRUENAS", ""] - items = catalogs if isinstance(catalogs, list) else list(catalogs.values() if isinstance(catalogs, dict) else []) - for c in items: - if not isinstance(c, dict): - continue - label = c.get("label") or c.get("name") or "?" - ttype = c.get("type", "") - trains = c.get("trains") or c.get("preferred_trains") or [] - trains_s = ", ".join(str(t) for t in trains) if isinstance(trains, list) and trains else "stable" - repo = c.get("repository") or c.get("branch") or "main" - healthy = "✅" if c.get("healthy", True) else "⚠️" - rows.append(f"{healthy} {label} | тип: {ttype} | trains: {trains_s} | repo: {repo}") - if len(rows) == 1: - return "Каталоги не распознаны (формат ответа изменился)." - return "\n".join(rows) - - def list_app_catalog(self, catalog: Optional[str] = None, category: str = "", search: str = "", limit: int = 100) -> str: - catalog = catalog or self.valves.default_catalog - lines = [f"📦 КАТАЛОГ ПРИЛОЖЕНИЙ: {catalog}", ""] - payload: Dict[str, Any] = {"catalog_name": catalog, "retrieve_all_trains": True} +payload: Dict[str, Any] = {"catalog_name": catalog, "retrieve_all_trains": True} if category: payload["categories"] = [category] - endpoints = ["app/available", "catalog/items", "chart/release/names"] - apps: List[Dict[str, Any]] = [] - used_ep = None - for ep in endpoints: - res = self._post(ep, payload, timeout=self.valves.request_timeout) - if res is None: - continue - if isinstance(res, list): - apps = res - used_ep = ep - break - if isinstance(res, dict): - for key in ("items", "applications", "releases", "apps"): - if key in res and isinstance(res[key], list): - apps = res[key] - used_ep = ep - break - if apps: - break - if not apps: - fallback_raw = self._rest_get(f"catalog_items_{catalog}_{category}", f"catalog?label={catalog}") - if fallback_raw: - try: - data = json.loads(fallback_raw) - if isinstance(data, dict): - for v in data.values(): - if isinstance(v, dict) and "items" in v: - apps = v["items"] - used_ep = "catalog?label=" - break - except Exception: - pass - if not apps: - return f"⚠️ Не удалось получить каталог {catalog}.\nПоследняя ошибка: {self._last_error}" - search_lc = search.lower().strip() if search else "" - shown = [] - for a in apps: - if not isinstance(a, dict): - continue - name = a.get("name") or a.get("app_name") or a.get("id") or "?" - title = a.get("title") or a.get("name") or a.get("app_name") or "?" - descr = a.get("description") or a.get("summary") or a.get("short_description") or "" - cats = a.get("categories") or a.get("category") or [] - if isinstance(cats, str): - cats = [cats] - if search_lc: - haystack = f"{name} {title} {descr} {' '.join(cats)}".lower() - if search_lc not in haystack: - continue - if category: - if isinstance(cats, list) and category.lower() not in [c.lower() for c in cats]: - continue - shown.append({ - "name": name, - "title": title, - "version": a.get("human_version") or a.get("latest_version") or a.get("version") or "?", - "categories": cats, - "description": descr[:160] + ("…" if len(descr) > 160 else ""), - }) - if len(shown) >= limit: - break - lines.append(f"📡 Источник: {used_ep}") - lines.append(f"📊 Всего в каталоге: {len(apps)} | показано: {len(shown)}") - lines.append("") - for a in shown: - cats_s = ", ".join(a["categories"][:3]) if a["categories"] else "—" - lines.append(f"• {a['name']} (v{a['version']}) [{cats_s}]") - if a["title"] and a["title"] != a["name"]: - lines.append(f" {a['title']}") - if a["description"]: - lines.append(f" {a['description']}") - if not shown: - lines.append("⚠️ По фильтру ничего не найдено.") - return "\n".join(lines) - - def get_app_info(self, app_name: str, catalog: Optional[str] = None, train: Optional[str] = None) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - catalog = catalog or self.valves.default_catalog - train = train or self.valves.default_train - lines = [f"🔍 ИНФОРМАЦИЯ О ПРИЛОЖЕНИИ: {app_name}", ""] - lines.append(f"📚 Каталог: {catalog} | 🚂 Train: {train}") - lines.append("") - info_payload: Dict[str, Any] = {"catalog_name": catalog, "app_name": app_name, "train": train} - info = self._post("app/info", info_payload, timeout=self.valves.request_timeout) - if info is None: - info = self._post("catalog/item", info_payload, timeout=self.valves.request_timeout) - if info is None: - lines.append(f"⚠️ Не удалось получить информацию: {self._last_error}") - return "\n".join(lines) - title = info.get("title") or info.get("name") or app_name - descr = info.get("description") or info.get("summary") or "" - ver = info.get("human_version") or info.get("version") or info.get("latest_version") or "?" - cats = info.get("categories") or [] - if isinstance(cats, str): - cats = [cats] - home = info.get("home") or info.get("home_url") or info.get("url") or "" - icon = info.get("icon_url") or info.get("icon") or "" - maintainer = (info.get("maintainers", [{}])[0].get("name") if isinstance(info.get("maintainers"), list) and info.get("maintainers") else info.get("maintainer", "")) - lines.append(f"📛 Title: {title}") - lines.append(f"📦 Версия: {ver}") - if cats: - lines.append(f"🗂 Категории: {', '.join(cats)}") - if maintainer: - lines.append(f"👤 Maintainer: {maintainer}") - if home: - lines.append(f"🌐 Дом. страница: {home}") - if icon: - lines.append(f"🖼 Иконка: {icon}") - if descr: - lines.append("") - lines.append(f"📝 Описание: {descr}") - lines.append("") - params_payload: Dict[str, Any] = { - "app_name": app_name, - "catalog_name": catalog, - "train": train, - "version": info.get("version") or info.get("latest_version"), - } - params = self._post("app/get_install_params", params_payload, timeout=self.valves.request_timeout) - if params and isinstance(params, dict): - lines.append("⚙️ ПАРАМЕТРЫ УСТАНОВКИ (values)") - lines.append("─" * 40) - schema = params.get("schema") or params.get("questions") or params.get("schema_details") or params - self._describe_schema(schema, lines, indent=0, max_depth=3) - lines.append("") - lines.append("💡 Подсказка: передай нужные значения в `values_json` метода install_app в виде JSON-строки. Например:") - lines.append(' values_json = \'{"service": {"web_port": 8080}}\'') - else: - lines.append("ℹ️ Параметры установки недоступны (приложение может быть уже установлено или каталог не синхронизирован).") - return "\n".join(lines) - - def _describe_schema(self, schema: Any, lines: List[str], indent: int = 0, max_depth: int = 3) -> None: - pad = " " * indent - if isinstance(schema, dict): - for key, val in schema.items(): - if key.startswith("$") or key in ("attrs", "additionalProperties"): - continue - if isinstance(val, dict): - lines.append(f"{pad}• {key}:") - if indent < max_depth: - self._describe_schema(val, lines, indent + 1, max_depth) - elif isinstance(val, list): - lines.append(f"{pad}• {key}: (список из {len(val)})") - if val and indent < max_depth: - sample = val[0] - if isinstance(sample, dict): - self._describe_schema(sample, lines, indent + 1, max_depth) - else: - lines.append(f"{pad}• {key} = {val}") - elif isinstance(schema, list): - for i, item in enumerate(schema[:5]): - if isinstance(item, dict): - self._describe_schema(item, lines, indent, max_depth) - lines.append(" ...") - - def install_app(self, app_name: str, catalog: Optional[str] = None, train: Optional[str] = None, version: str = "", values_json: str = "{}", confirm: bool = False) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - catalog = catalog or self.valves.default_catalog - train = train or self.valves.default_train - try: - values = json.loads(values_json) if values_json else {} - if not isinstance(values, dict): - return "❌ Ошибка: values_json должен быть JSON-объектом (начинаться с '{')." - except json.JSONDecodeError as e: - return f"❌ Ошибка парсинга values_json: {e}" - payload: Dict[str, Any] = {"app_name": app_name, "catalog_name": catalog, "train": train, "values": values} - if version: - payload["version"] = version - if not confirm: - lines = [ - "🔍 DRY-RUN: установка НЕ будет выполнена", - "", - f"📦 Приложение: {app_name}", - f"📚 Каталог: {catalog}", - f"🚂 Train: {train}", - f"🏷 Версия: {version or 'latest'}", - f"⚙️ Values: {json.dumps(values, ensure_ascii=False, indent=2) if values else '(по умолчанию)'}", - "", - "Для реальной установки вызови метод повторно с confirm=True.", - ] - return "\n".join(lines) - res = self._post("app/install", payload, timeout=self.valves.install_timeout) - if res is None: - return f"❌ Ошибка установки {app_name}.\nПоследняя ошибка: {self._last_error}" - job_id = res if isinstance(res, str) else (res.get("job_id") if isinstance(res, dict) else None) - if not job_id: - job_id = "неизвестно" - self._cache.pop("apps", None) - for k in list(self._cache.keys()): - if k.startswith("apps_"): - self._cache.pop(k, None) - return f"✅ OK: установка {app_name} запущена!\n Каталог: {catalog}, train: {train}, версия: {version or 'latest'}\n Job ID: {job_id}\n ⏱ Установка может занять несколько минут (тяжёлые образы — до 15 минут).\n 👉 Проверить состояние: app_status('{app_name}') или get_apps_status()." - - def install_custom_app(self, app_name: str, compose_yaml: str, env_json: str = "{}", confirm: bool = False) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - if not compose_yaml or "services:" not in compose_yaml: - return "❌ Ошибка: compose_yaml должен содержать секцию 'services:'." - try: - env = json.loads(env_json) if env_json else {} - if not isinstance(env, dict): - return "❌ Ошибка: env_json должен быть JSON-объектом." - except json.JSONDecodeError as e: - return f"❌ Ошибка парсинга env_json: {e}" - if not confirm: - lines = [ - "🔍 DRY-RUN: установка Custom App НЕ будет выполнена", - "", - f"📦 Имя: {app_name}", - f"📝 Compose (первые 800 симв.):", - compose_yaml[:800] + ("…" if len(compose_yaml) > 800 else ""), - f"🌍 Env: {json.dumps(env, ensure_ascii=False, indent=2) if env else '(нет)'}", - "", - "Для реальной установки вызови метод повторно с confirm=True.", - ] - return "\n".join(lines) - payload: Dict[str, Any] = {"app_name": app_name, "compose_string": compose_yaml} - if env: - payload["values"] = env - res = self._post("app/compose_install", payload, timeout=self.valves.install_timeout) - if res is None: - try: - files = {"compose": (f"{app_name}.yml", compose_yaml, "application/x-yaml")} - r = requests.post( - f"{self._base}/api/v2.0/app/compose_install", - headers={"Authorization": f"Bearer {self.valves.TRUENAS_API_KEY}"}, - data={"app_name": app_name, **({"values": json.dumps(env)} if env else {})}, - files=files, - timeout=self.valves.install_timeout, - verify=self.valves.verify_ssl, - ) - if r.status_code in (200, 201, 202): - try: - res = r.json() - except Exception: - res = {"status": "ok", "text": r.text} - else: - self._last_error = f"POST compose_install (form): HTTP {r.status_code}: {r.text[:300]}" - res = None - except Exception as e: - self._last_error = f"compose_install fallback: {type(e).__name__}: {e}" - res = None - if res is None: - return f"❌ Ошибка установки Custom App '{app_name}'.\nПоследняя ошибка: {self._last_error}" - self._cache.pop("apps", None) - for k in list(self._cache.keys()): - if k.startswith("apps_"): - self._cache.pop(k, None) - return f"✅ OK: Custom App '{app_name}' поставлен на установку!\n ⏱ Развёртывание может занять 1–5 минут.\n 👉 Проверить состояние: app_status('{app_name}') или get_apps_status()." - - def app_action(self, app_name: str, action: str, confirm: bool = False) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - action_l = action.lower().strip() - dangerous = {"delete", "upgrade"} - if action_l in dangerous and not confirm: - return f"⚠️ '{action_l}' — потенциально разрушительная операция.\n Для подтверждения вызови с confirm=True." - if action_l not in {"start","stop","restart","delete","upgrade","redeploy"}: - return f"❌ Неизвестное действие: {action}. Допустимо: start, stop, restart, delete, upgrade, redeploy." - endpoints = { - "start": ("POST", f"app/{app_name}/start"), - "stop": ("POST", f"app/{app_name}/stop"), - "restart": ("POST", f"app/{app_name}/restart"), - "redeploy": ("POST", f"app/{app_name}/redeploy"), - "delete": ("DELETE", f"app/{app_name}"), - "upgrade": ("POST", f"app/{app_name}/upgrade"), - } - method, ep = endpoints[action_l] - if method == "DELETE": - res = self._delete(ep, timeout=self.valves.install_timeout) - else: - res = self._post(ep, timeout=self.valves.install_timeout) - if res is None: - return f"❌ Ошибка: {action_l} '{app_name}' не выполнено.\n {self._last_error}" - self._cache.pop("apps", None) - for k in list(self._cache.keys()): - if k.startswith("apps_"): - self._cache.pop(k, None) - icons = {"start":"▶️","stop":"⏹","restart":"🔄","redeploy":"♻️","delete":"🗑","upgrade":"⬆️"} - verbs = {"start":"запущено","stop":"остановлено","restart":"перезапущено","redeploy":"переразвёрнуто","delete":"удалено","upgrade":"обновлено"} - return f"{icons.get(action_l, '✓')} OK: '{app_name}' → {verbs[action_l]}\n Ответ API: {json.dumps(res, default=str)[:300]}" - - def app_status(self, app_name: str) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - cfg = self._post(f"app/{app_name}/config", {}, timeout=self.valves.request_timeout) - info = self._post(f"app/{app_name}", {}, timeout=self.valves.request_timeout) - if cfg is None and info is None: - return f"❌ Не удалось получить статус '{app_name}'. {self._last_error}" - lines = [f"📊 СТАТУС ПРИЛОЖЕНИЯ: {app_name}", ""] - state = "N/A" - workloads = None - if isinstance(info, dict): - state = info.get("state") or info.get("status") or (info.get("status_details", {}) or {}).get("state", "N/A") - workloads = info.get("workloads") or info.get("used_ports") - imap = {"RUNNING":"🟢","STOPPED":"⚪","DEPLOYING":"🟡","CRASHED":"🔴","STOPPING":"🟠"} - lines.append(f"🚦 Состояние: {imap.get(str(state).upper(), '⚪')} {state}") - version = (info or {}).get("human_version") or (info or {}).get("version") or "N/A" - lines.append(f"🏷 Версия: {version}") - if workloads: - lines.append("🌐 Порты / пробросы:") - for w in workloads[:10]: - if isinstance(w, dict): - port = w.get("port") or w.get("container_port") - host = w.get("host_port") or w.get("node_port") - proto = w.get("protocol", "TCP") - lines.append(f" • {port}/{proto} → {host}") - if isinstance(cfg, dict): - lines.append("") - lines.append("⚙️ Ключевые настройки:") - for key in ("image","image_tag","namespace","release","service","storage","ingress"): - if key in cfg: - lines.append(f" • {key}: {json.dumps(cfg[key], default=str)[:200]}") - return "\n".join(lines) - - def wait_for_app(self, app_name: str, desired_state: str = "RUNNING", timeout: int = 600, poll: Optional[int] = None) -> str: - if not app_name: - return "❌ Ошибка: app_name обязателен." - desired = desired_state.upper() - interval = poll if poll is not None else self.valves.poll_interval - deadline = time.time() + timeout - last_state = None - attempts = 0 - lines = [f"⏳ Ждём, пока '{app_name}' не перейдёт в {desired} (до {timeout}с)…"] - while time.time() < deadline: - attempts += 1 - info = self._post(f"app/{app_name}", {}, timeout=self.valves.request_timeout) - state = "N/A" - if isinstance(info, dict): - state = info.get("state") or info.get("status") or (info.get("status_details", {}) or {}).get("state", "N/A") - last_state = state - if str(state).upper() == desired: - lines.append(f"✅ Готово: '{app_name}' = {state} (попытка #{attempts}, {int(time.time() - (deadline - timeout))}с)") - return "\n".join(lines) - if str(state).upper() == "CRASHED": - lines.append(f"❌ Приложение в CRASHED на попытке #{attempts}. Прерываю.") - return "\n".join(lines) - time.sleep(interval) - lines.append(f"⏱ Таймаут: '{app_name}' всё ещё в состоянии '{last_state}' после {timeout}с ({attempts} опросов).") - return "\n".join(lines) - - 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.list_catalogs()), - ("ПРИЛОЖЕНИЯ", 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) \ No newline at end of file + # v2.1.1: расширенный список эндпоинтов — TrueNAS 25.10 имеет разные пути + # в зависимости от версии (chart/release/ — новые, catalog/ — старые) + endpoints = [ + "app/available", # новый (Electric Eel+) + "app.available", + "chart/release/names", # иногда используется + "chart.release.names", + "chart_release/names", + "catalog/items", # fallback + "catalog.items", + "catalog_items", + "app/available_apps", + "app/list_apps", + ] \ No newline at end of file