From 421ff928a33aec909fa023ef8d8a10370f5f8ec2 Mon Sep 17 00:00:00 2001
From: host <arumyantsev@rummail.ru>
Date: Fri, 24 Jul 2026 02:43:10 +0300
Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?=
 =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?=
 =?UTF-8?q?/=C2=BB?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 truenas_tools (1).py | 327 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 327 insertions(+)
 create mode 100644 truenas_tools (1).py

diff --git a/truenas_tools (1).py b/truenas_tools (1).py
new file mode 100644
index 0000000..3bb9108
--- /dev/null
+++ b/truenas_tools (1).py	
@@ -0,0 +1,327 @@
+import json
+import time
+import requests
+from typing import Dict
+from pydantic import BaseModel, Field
+
+
+class Tools:
+    """
+    Мониторинг TrueNAS через Open WebUI.
+    REST с перебором альтернативных путей.
+    TrueNAS 25.10 (Goldeye).
+    """
+
+    class Valves(BaseModel):
+        TRUENAS_URL: str = Field(
+            default="http://192.168.31.100",
+            description="IP-адрес или домен TrueNAS (http://...)",
+        )
+        TRUENAS_API_KEY: str = Field(
+            default="",
+            description="API-ключ для доступа к TrueNAS (Settings → API Keys)",
+        )
+        CACHE_TTL: int = Field(
+            default=30, description="Время кэширования в секундах (0 = без кэша)"
+        )
+
+    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):
+        self._cache.clear()
+
+    def _rest_get(self, cache_key: str, endpoint: str):
+        now = time.time()
+        if cache_key in self._cache:
+            t, d = self._cache[cache_key]
+            if now - t < self.valves.CACHE_TTL:
+                return d
+        try:
+            r = requests.get(
+                f"{self._base}/api/v2.0/{endpoint}", headers=self._headers, timeout=10
+            )
+            if r.status_code == 200:
+                data = r.json()
+                s = json.dumps(data, default=str)
+                if self.valves.CACHE_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 Exception as e:
+            self._last_error = f"Исключение на {endpoint}: {type(e).__name__}: {e}"
+            return None
+
+    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=10,
+                    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 _try_endpoints(self, cache_key: str, endpoints: list) -> 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 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 "Ошибка парсинга"
+        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}\n"
+            f"📊 Load Average: {load_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
+            rows.append(
+                f"{icon} '{name}' | {status} | {total/1024**4:.2f} TiB, {used_pct:.1f}% занято"
+            )
+        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", "")
+            rows.append(f"💾 {name} ({dtype}) | {model} | {serial} | {size}")
+        return "\n".join(rows)
+
+    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)
+
+    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} | RAM: {mem_s} | автозапуск: {autostart}"
+            )
+        return "\n".join(rows)
+
+    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:
+        return (
+            "══════════════════════════════════════════════════\n"
+            "📋 ПОЛНЫЙ ОТЧЁТ TRUENAS\n"
+            "(REST перебор путей)\n"
+            "══════════════════════════════════════════════════\n\n"
+            "--- СИСТЕМА ---\n" + self.get_system_status() + "\n\n"
+            "--- ПУЛЫ ZFS ---\n" + self.get_pools_status() + "\n\n"
+            "--- ДИСКИ ---\n" + self.get_disks_temp_and_info() + "\n\n"
+            "--- ПРИЛОЖЕНИЯ ---\n" + self.get_apps_status() + "\n\n"
+            "--- ВИРТУАЛЬНЫЕ МАШИНЫ ---\n" + self.get_vms_status() + "\n\n"
+            "--- КОНТЕЙНЕРЫ (virt/incus) ---\n" + self.get_containers_status() + "\n\n"
+            "--- АЛЕРТЫ ---\n" + self.get_active_alerts()
+        )