core/utils.py - master - HookDev-Arch/ServerMonitor

core/utils.py
136 строк4 KB
HookDev-Arch
upload files
09 окт 2025, 02:25
09 окт 2025, 02:258bd6873
import os
import json
import shutil
import platform
from pathlib import Path

import requests

BASE_DIR = Path(__file__).resolve().parents[1]
DATA_DIR = BASE_DIR / "data"
LOGS_DIR = DATA_DIR / "logs"
SERVERS_FILE = DATA_DIR / "servers.json"

LOGS_DIR.mkdir(parents=True, exist_ok=True)
DATA_DIR.mkdir(parents=True, exist_ok=True)


def get_size(bytes_val):
    """Преобразует байты в удобный формат"""
    try:
        bytes_val = float(bytes_val)
    except Exception:
        return "0 B"
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_val < 1024.0:
            return f"{bytes_val:.2f} {unit}"
        bytes_val /= 1024.0
    return f"{bytes_val:.2f} PB"


def read_json(path: Path, default):
    if not path.exists():
        return default
    try:
        with path.open("r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def write_json(path: Path, data):
    tmp = path.with_suffix(".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    tmp.replace(path)


def detect_gpu_backend():
    """Возвращает 'nvml', 'nvidia-smi' или None"""
    try:
        import pynvml
        try:
            pynvml.nvmlInit()
            return "nvml"
        except Exception:
            pass
    except Exception:
        pass
    if shutil.which("nvidia-smi"):
        return "nvidia-smi"
    return None


GPU_BACKEND = detect_gpu_backend()


# Кэш для IP адреса
_cached_ip = None
_ip_cache_time = 0
_ip_cache_duration = 300  # 5 минут кэширования

def get_public_ip(timeout=2.0):
    """Получает публичный IP с кэшированием и множественными сервисами"""
    import time
    
    global _cached_ip, _ip_cache_time
    
    # Проверяем кэш
    current_time = time.time()
    if _cached_ip and (current_time - _ip_cache_time) < _ip_cache_duration:
        return _cached_ip
    
    # Список сервисов для получения IP (в порядке надёжности)
    ip_services = [
        "https://api.ipify.org",
        "https://ipinfo.io/ip",
        "https://icanhazip.com",
        "https://ifconfig.me/ip",
        "https://checkip.amazonaws.com"
    ]
    
    for service in ip_services:
        try:
            response = requests.get(service, timeout=timeout, headers={
                'User-Agent': 'HookDev-Arch-Dashboard/3.1.2'
            })
            if response.status_code == 200:
                ip = response.text.strip()
                # Простая валидация IP
                if ip and len(ip.split('.')) == 4 and all(0 <= int(x) <= 255 for x in ip.split('.') if x.isdigit()):
                    _cached_ip = ip
                    _ip_cache_time = current_time
                    return ip
        except Exception:
            continue
    
    # Если все сервисы недоступны, возвращаем кэшированное значение или N/A
    return _cached_ip if _cached_ip else "N/A"


def force_refresh_public_ip():
    """Принудительно обновляет кэш публичного IP"""
    global _cached_ip, _ip_cache_time
    _cached_ip = None
    _ip_cache_time = 0
    return get_public_ip()


def load_servers():
    data = read_json(SERVERS_FILE, {"servers": []})
    out = []
    for s in data.get("servers", []):
        out.append({
            "name": s.get("name"),
            "host": s.get("host"),
            "port": s.get("port", 22),
            "user": s.get("user", "root"),
            "password": s.get("password", ""),
            "api_port": s.get("api_port", 3333),
            "agent_path": s.get("agent_path", "~/agent.py")
        })
    return {"servers": out}


def save_servers(data):
    write_json(SERVERS_FILE, data)