diff --git a/truenas_tools.py b/truenas_tools.py deleted file mode 100644 index 495bfd9..0000000 --- a/truenas_tools.py +++ /dev/null @@ -1,319 +0,0 @@ -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._base = self.valves.TRUENAS_URL.rstrip("/") - self._headers = { - "Authorization": f"Bearer {self.valves.TRUENAS_API_KEY}", - "Content-Type": "application/json", - } - self._last_error = "нет данных (запросы ещё не выполнялись)" - try: - import urllib3 - - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - except Exception: - pass - - 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() - ) \ No newline at end of file