debian-manager/app/core/proxmox.py

"""Proxmox VE: API через proxmoxer, ноды, VM/LXC, управление питанием."""
from __future__ import annotations

from typing import List, Optional

from app.db.database import get_session
from app.db.models import Setting


def _get(key: str) -> str:
    with get_session() as s:
        row = s.get(Setting, key)
        return row.value if row else ""


def _set(key: str, value: str) -> None:
    with get_session() as s:
        row = s.get(Setting, key) or Setting(key=key)
        row.value = value
        s.add(row)
        s.commit()


def get_config() -> dict:
    return {
        "host": _get("pve_host"),
        "user": _get("pve_user") or "root@pam",
        "token_name": _get("pve_token_name"),
        "token_value": _get("pve_token_value"),
        "verify_ssl": _get("pve_verify_ssl") == "1",
    }


def save_config(host: str, user: str, token_name: str, token_value: str,
                verify_ssl: bool) -> None:
    _set("pve_host", host.rstrip("/"))
    _set("pve_user", user)
    _set("pve_token_name", token_name)
    _set("pve_token_value", token_value)
    _set("pve_verify_ssl", "1" if verify_ssl else "0")


def _client():
    """Создать proxmoxer-клиент. Бросает если не настроено."""
    from proxmoxer import ProxmoxAPI
    cfg = get_config()
    if not cfg["host"] or not cfg["token_name"] or not cfg["token_value"]:
        raise RuntimeError("Proxmox не настроен — заполните хост и API-токен")
    # убрать http:// или https:// если пользователь ввёл
    host = cfg["host"].replace("http://", "").replace("https://", "").rstrip("/")
    return ProxmoxAPI(
        host, user=cfg["user"],
        token_name=cfg["token_name"], token_value=cfg["token_value"],
        verify_ssl=cfg["verify_ssl"], timeout=15,
    )


def test_connection() -> tuple[bool, str]:
    try:
        pve = _client()
        version = pve.version.get()
        return True, f"Proxmox VE {version.get('version', '?')}"
    except Exception as e:
        return False, str(e)


def nodes() -> List[dict]:
    return _client().nodes.get()


def node_status(node: str) -> dict:
    return _client().nodes(node).status.get()


def vms(node: str) -> List[dict]:
    """Все VM и LXC на ноде."""
    pve = _client()
    qemu = [{"type": "qemu", **x} for x in pve.nodes(node).qemu.get()]
    lxc = [{"type": "lxc", **x} for x in pve.nodes(node).lxc.get()]
    return qemu + lxc


def vm_status(node: str, vmid: int, vm_type: str) -> dict:
    pve = _client()
    if vm_type == "qemu":
        return pve.nodes(node).qemu(vmid).status.current.get()
    return pve.nodes(node).lxc(vmid).status.current.get()


def vm_action(node: str, vmid: int, vm_type: str, action: str) -> str:
    """start/stop/shutdown/reboot/reset. Возвращает UPID задачи."""
    assert action in ("start", "stop", "shutdown", "reboot", "reset")
    pve = _client()
    api = pve.nodes(node).qemu(vmid) if vm_type == "qemu" else pve.nodes(node).lxc(vmid)
    return getattr(api.status, action).post()


def vm_config(node: str, vmid: int, vm_type: str) -> dict:
    pve = _client()
    if vm_type == "qemu":
        return pve.nodes(node).qemu(vmid).config.get()
    return pve.nodes(node).lxc(vmid).config.get()


def vm_rrd(node: str, vmid: int, vm_type: str, timeframe: str = "hour") -> list:
    """Метрики из RRD (cpu, mem, netin, netout...)."""
    pve = _client()
    api = pve.nodes(node).qemu(vmid) if vm_type == "qemu" else pve.nodes(node).lxc(vmid)
    return api.rrd.get(timeframe=timeframe, cf="AVERAGE")


def get_vm_ip(node: str, vmid: int, vm_type: str) -> Optional[str]:
    """IP из QEMU Guest Agent или из конфига LXC."""
    try:
        pve = _client()
        if vm_type == "qemu":
            data = pve.nodes(node).qemu(vmid).agent("network-get-interfaces").get()
            for iface in data.get("result", []):
                for addr in iface.get("ip-addresses", []):
                    ip = addr.get("ip-address", "")
                    if addr.get("ip-address-type") == "ipv4" and not ip.startswith("127."):
                        return ip
        else:
            cfg = pve.nodes(node).lxc(vmid).config.get()
            net0 = cfg.get("net0", "")
            # "name=eth0,bridge=vmbr0,ip=192.168.1.10/24,gw=..."
            for part in net0.split(","):
                if part.startswith("ip=") and part != "ip=dhcp":
                    return part[3:].split("/")[0]
    except Exception:
        pass
    return None