# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART)."""

from __future__ import annotations

import logging
import re
from typing import Any

from napalm_proxmox import utils

logger = logging.getLogger(__name__)

_JsonDict = dict[str, Any]


class ProxmoxSystemMixin:
    """Mixin providing system-level NAPALM methods."""

    # ------------------------------------------------------------------ #
    # get_environment
    # ------------------------------------------------------------------ #

    def get_environment(self) -> _JsonDict:
        """Return environment status (CPU, memory, temperature)."""
        status = self._get_node_status()
        env: _JsonDict = {
            "fans": {},
            "temperature": {},
            "power": {},
            "cpu": {},
            "memory": {"available_ram": 0, "used_ram": 0},
        }

        # CPU
        cpu_usage = status.get("cpu", 0.0)
        env["cpu"]["0"] = {"%usage": round(float(cpu_usage) * 100, 2)}

        # Memory (Proxmox reports in bytes)
        mem = status.get("memory", {})
        total = int(mem.get("total", 0) or 0)
        used = int(mem.get("used", 0) or 0)
        env["memory"]["available_ram"] = total
        env["memory"]["used_ram"] = used

        # Temperature (from node sensors if available)
        try:
            sensors = self._node_api().hardware.sensors.get() or []
        except Exception as exc:
            logger.debug("Failed to fetch hardware sensors: %s", exc)
            sensors = []
        for sensor in sensors:
            name = sensor.get("name", "unknown")
            value = sensor.get("value", None)
            if value is not None:
                try:
                    temp_c = float(value)
                    env["temperature"][name] = {
                        "temperature": temp_c,
                        "is_alert": temp_c >= 80.0,
                        "is_critical": temp_c >= 95.0,
                    }
                except (TypeError, ValueError):
                    pass

        return env

    # ------------------------------------------------------------------ #
    # get_ntp_servers / get_ntp_stats
    # ------------------------------------------------------------------ #

    def get_ntp_servers(self) -> dict[str, _JsonDict]:
        """Return configured NTP servers."""
        ntp = self._get_node_ntp()
        servers: dict[str, _JsonDict] = {}
        # Proxmox reports a comma-separated or space-separated server list
        raw = ntp.get("server", "") or ntp.get("servers", "")
        for srv in re.split(r"[\s,]+", raw):
            srv = srv.strip()
            if srv:
                servers[srv] = {}
        return servers

    def get_ntp_stats(self) -> list[_JsonDict]:
        """Return NTP synchronisation statistics from chronyc/ntpq output."""
        raw = self._exec_ssh_command(
            "chronyc -n tracking 2>/dev/null || ntpq -pn 2>/dev/null || true"
        )
        stats: list[_JsonDict] = []
        for line in raw.splitlines():
            line = line.strip()
            # ntpq -pn format: *remote  refid  st  t  when  poll  reach  delay  offset  jitter
            m = re.match(
                r"^([\*\+\-\s])([\d.]+)\s+([\d.]+)\s+(\d+)\s+\S+\s+(\S+)\s+(\d+)\s+(\d+)\s+([\d.]+)\s+([-\d.]+)\s+([\d.]+)",
                line,
            )
            if m:
                synced = m.group(1).strip() == "*"
                stats.append(
                    {
                        "remote": m.group(2),
                        "referenceid": m.group(3),
                        "synchronized": synced,
                        "stratum": int(m.group(4)),
                        "type": "",
                        "when": m.group(5),
                        "hostpoll": int(m.group(6)),
                        "reachability": int(m.group(7)),
                        "delay": float(m.group(8)),
                        "offset": float(m.group(9)),
                        "jitter": float(m.group(10)),
                    }
                )
        return stats

    # ------------------------------------------------------------------ #
    # get_snmp_information
    # ------------------------------------------------------------------ #

    def get_snmp_information(self) -> _JsonDict:
        """Return SNMP information.

        Proxmox does not expose SNMP configuration via the REST API.
        We read /etc/snmp/snmpd.conf via exec if available.
        """
        raw = self._exec_ssh_command(
            "cat /etc/snmp/snmpd.conf 2>/dev/null || true"
        )
        communities: dict[str, _JsonDict] = {}
        location = ""
        contact = ""
        for line in raw.splitlines():
            line = line.strip()
            if line.startswith("#") or not line:
                continue
            # rocommunity <community> [source]
            m = re.match(r"^(ro|rw)community\s+(\S+)", line)
            if m:
                mode = "ro" if m.group(1) == "ro" else "rw"
                community = m.group(2)
                communities[community] = {"acl": "N/A", "mode": mode}
            m_loc = re.match(r"^sysLocation\s+(.+)", line)
            if m_loc:
                location = m_loc.group(1).strip()
            m_con = re.match(r"^sysContact\s+(.+)", line)
            if m_con:
                contact = m_con.group(1).strip()

        return {
            "chassis_id": self._node_name,
            "community": communities,
            "contact": contact,
            "location": location,
        }

    # ------------------------------------------------------------------ #
    # get_users
    # ------------------------------------------------------------------ #

    def get_users(self) -> dict[str, _JsonDict]:
        """Return users configured on the Proxmox node.

        Reads from both the Proxmox access/users API and local /etc/passwd.
        """
        result: dict[str, _JsonDict] = {}
        try:
            pve_users = self._api.access.users.get() or []  # type: ignore[union-attr]
        except Exception as exc:
            logger.debug("Failed to fetch Proxmox users: %s", exc)
            pve_users = []

        for user in pve_users:
            uid = user.get("userid", "")
            if not uid:
                continue
            # Proxmox roles: Administrator -> 15, otherwise 1
            groups = user.get("groups", "") or ""
            level = 1
            try:
                roles = self._api.access.users(uid).get() or {}  # type: ignore[union-attr]
                if "Administrator" in str(roles):
                    level = 15
            except Exception as exc:
                logger.debug("Failed to fetch roles for user %s: %s", uid, exc)
            result[uid] = {
                "level": level,
                "password": "",
                "sshkeys": [],
            }

        # Merge local OS users from /etc/passwd
        raw = self._exec_ssh_command("getent passwd 2>/dev/null || cat /etc/passwd")
        for line in raw.splitlines():
            parts = line.split(":")
            if len(parts) < 7:
                continue
            uname, _, uid_str, *_ = parts
            try:
                uid_int = int(uid_str)
            except ValueError:
                continue
            if uname not in result and uid_int < 1000 or uid_int == 0:
                result[uname] = {
                    "level": 15 if uid_int == 0 else 0,
                    "password": "",
                    "sshkeys": [],
                }

        return result

    # ------------------------------------------------------------------ #
    # Packages (Debian APT)
    # ------------------------------------------------------------------ #

    def get_packages(self) -> list[_JsonDict]:
        """Return installed Debian packages with available-update info.

        Installed list comes from ``dpkg-query`` via SSH (the Proxmox API
        ``/apt/installed`` endpoint is not implemented on PVE 8.x).
        Available updates come from the Proxmox API ``/apt/update``.
        """
        # Available updates from Proxmox API (keyed by package name)
        upgradable: dict[str, str] = {}
        try:
            for upd in self._api.nodes(self._node_name).apt.update.get():
                pkg = upd.get("Package", "")
                if pkg:
                    upgradable[pkg] = upd.get("Version", "")
        except Exception as exc:
            logger.debug("Failed to fetch available APT updates: %s", exc)

        # Installed packages via SSH dpkg-query
        raw = self._exec_ssh_command(
            "dpkg-query -W -f='${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}\\n'"
            " 2>/dev/null"
        )
        result: list[_JsonDict] = []
        for line in raw.splitlines():
            parts = line.strip().split("\t")
            if len(parts) < 2:
                continue
            name = parts[0]
            version = parts[1] if len(parts) > 1 else ""
            status = parts[2] if len(parts) > 2 else "installed"
            size_kb = parts[3] if len(parts) > 3 else "0"
            if not name or status != "installed":
                continue
            size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0
            result.append({
                "name": name,
                "version": version,
                "installed": True,
                "description": "",
                "size": size_bytes,
                "source": "pve",
                "upgrade_version": upgradable.get(name, ""),
            })
        return result

    # ------------------------------------------------------------------ #
    # Device warnings
    # ------------------------------------------------------------------ #

    def get_device_warnings(self) -> list[_JsonDict]:
        """Return warnings for the Proxmox node.

        Currently detects:
        - lldpd not installed (LLDP neighbor discovery unavailable)
        - Available package updates (via Proxmox APT API)
        - Missing / invalid subscription
        """
        warnings: list[_JsonDict] = []

        # 0. LLDP daemon
        try:
            lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
            if not lldpd_path:
                warnings.append({
                    "code": "lldpd_not_installed",
                    "severity": "warning",
                    "action": "install_lldpd",
                })
        except Exception as exc:
            logger.debug("Failed to check for lldpd: %s", exc)

        # 1. Available package updates
        try:
            updates = self.get_available_updates()
            if updates:
                warnings.append({
                    "code": "updates_available",
                    "severity": "warning",
                    "title": (
                        f"{len(updates)} package update"
                        f"{'s' if len(updates) != 1 else ''} available"
                    ),
                    "message": None,
                    "action": None,
                    "meta": {
                        "count": len(updates),
                        "packages": [u["name"] for u in updates],
                    },
                })
        except Exception as exc:
            logger.debug("Failed to check available updates: %s", exc)

        # 2. Subscription status
        try:
            sub = self._get_node_subscription()
            status = sub.get("status", "")
            if status in ("NotFound", "Invalid", "Expired"):
                warnings.append({
                    "code": "no_subscription",
                    "severity": "warning",
                    "action": None,
                    "meta": {"status": status},
                })
        except Exception as exc:
            logger.debug("Failed to check subscription status: %s", exc)

        return warnings

    # ------------------------------------------------------------------ #
    # Services (systemd)
    # ------------------------------------------------------------------ #

    def get_services(self) -> list[_JsonDict]:
        """Return systemd services with running and enabled state.

        Uses two ``systemctl`` invocations combined in a single SSH command:
        - ``list-unit-files`` for the static enabled/disabled state
        - ``list-units`` for the live running state
        """
        raw = self._exec_ssh_command(
            "{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;"
            " echo '---UNITS---';"
            " systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;"
            " } || true"
        )

        # Parse enabled state from list-unit-files
        enabled_map: dict[str, bool] = {}
        section = "files"
        for line in raw.splitlines():
            if line.strip() == "---UNITS---":
                section = "units"
                continue
            parts = line.strip().split(None, 1)
            if len(parts) < 1:
                continue
            unit = parts[0].lstrip("\u25cf").strip()
            if not unit.endswith(".service"):
                continue
            name = unit[: -len(".service")]
            if section == "files":
                state = parts[1].strip() if len(parts) > 1 else ""
                enabled_map[name] = state in ("enabled", "enabled-runtime", "static")

        # Parse running state from list-units
        running_map: dict[str, bool] = {}
        section = "files"
        for line in raw.splitlines():
            if line.strip() == "---UNITS---":
                section = "units"
                continue
            if section != "units":
                continue
            parts = line.strip().lstrip("\u25cf").strip().split(None, 4)
            if len(parts) < 4:
                continue
            unit = parts[0]
            if not unit.endswith(".service"):
                continue
            name = unit[: -len(".service")]
            sub_state = parts[3]
            running_map[name] = sub_state == "running"

        all_names = sorted(set(enabled_map) | set(running_map))
        return [
            {
                "name": name,
                "running": running_map.get(name, False),
                "enabled": enabled_map.get(name, False),
                "pid": 0,
            }
            for name in all_names
        ]

    def manage_service(self, name: str, action: str) -> _JsonDict:
        """Start / stop / restart / enable / disable a systemd service."""
        if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name):
            raise ValueError(f"Invalid service name: {name!r}")
        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
            raise ValueError(f"Invalid action: {action!r}")
        output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true")
        return {"success": True, "output": output}

    # ------------------------------------------------------------------ #
    # Available updates
    # ------------------------------------------------------------------ #

    def get_available_updates(self) -> list[_JsonDict]:
        """Return list of upgradable packages from the Proxmox APT API."""
        updates: list[_JsonDict] = []
        try:
            for upd in self._api.nodes(self._node_name).apt.update.get():
                pkg = upd.get("Package", "")
                if not pkg:
                    continue
                updates.append({
                    "name": pkg,
                    "current_version": upd.get("OldVersion", ""),
                    "new_version": upd.get("Version", ""),
                })
        except Exception as exc:
            logger.debug("Failed to fetch available updates: %s", exc)
        return sorted(updates, key=lambda u: u["name"])

    def apply_updates(self, packages: list[str]) -> _JsonDict:
        """Upgrade the given packages via ``apt-get install`` over SSH."""
        for pkg in packages:
            if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
                raise ValueError(f"Invalid package name: {pkg!r}")
        pkg_args = " ".join(packages)
        output = self._exec_ssh_command(
            f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true"
        )
        return {"success": True, "output": output}

    # ------------------------------------------------------------------ #
    # SNMP / Health
    # ------------------------------------------------------------------ #

    def get_snmp_config(self):
        """Return SNMP agent config if snmpd is installed and running on the node.

        Uses _exec_ssh_command (Proxmox API exec or SSH) to inspect the node.
        Returns a SNMPConfigDict or None.
        """
        try:
            from napalm_device_types.models import SNMPConfigDict
        except ImportError:
            return None

        running = (
            self._exec_ssh_command("systemctl is-active snmpd 2>/dev/null || true").strip()
            == "active"
        )
        if not running:
            return None

        community = "public"
        port = 161
        try:
            conf = self._exec_ssh_command(
                "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5"
            )
            for line in conf.splitlines():
                parts = line.split()
                if not parts:
                    continue
                kw = parts[0].lower()
                if kw in ("rocommunity", "rwcommunity") and len(parts) >= 2:
                    community = parts[1]
                    break
                elif kw == "com2sec" and len(parts) >= 4:
                    community = parts[3]
                    break
        except Exception as exc:
            logger.debug("Failed to parse snmpd.conf: %s", exc)

        return SNMPConfigDict(running=True, community=community, port=port, version="2c")

    def run_device_action(self, action: str) -> dict:
        """Execute a named administrative action on the Proxmox node."""
        if action == "fix_snmp":
            return self._action_fix_snmp()
        if action == "install_lldpd":
            return self._action_install_lldpd()
        raise NotImplementedError(f"Unknown action: {action!r}")

    def _action_install_lldpd(self) -> dict:
        """Install, enable and start lldpd on the Proxmox node.

        Proxmox runs Debian/Linux underneath, so this is a plain apt install
        followed by enabling the systemd service. lldpd's defaults (listen on
        all interfaces) are sufficient to discover the directly-connected
        switch via the management bridge (e.g. vmbr0).
        """
        lines: list[str] = []

        install_out = self._exec_ssh_command(
            "DEBIAN_FRONTEND=noninteractive apt-get install -y lldpd 2>&1 | tail -5"
        )
        lines.append(f"[install] {install_out.strip()[-200:]}")

        enable_out = self._exec_ssh_command(
            "systemctl enable --now lldpd 2>&1 || service lldpd start 2>&1 || true"
        )
        lines.append(f"[service] {enable_out.strip()[-200:]}")

        verify = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
        success = bool(verify)
        if success:
            lines.append("[ok] lldpd installed. Neighbors will appear after a short warm-up period.")
        else:
            lines.append("[warn] lldpd does not appear to be installed after the attempt.")

        return {"success": success, "output": "\n".join(lines)}

    def _action_fix_snmp(self) -> dict:
        """Install, configure and start snmpd on the Proxmox node.

        Proxmox runs Debian/Linux underneath. _exec_ssh_command runs as root
        (either via Proxmox API execute endpoint or SSH with root credentials),
        so no sudo is needed.
        """
        import base64 as _b64
        lines: list[str] = []

        # 1. Install snmpd and snmp client tools
        install_out = self._exec_ssh_command(
            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1 | tail -5"
        )
        lines.append(f"[install] {install_out.strip()[-200:]}")

        # 2. Detect the IP this connection comes from (for firewall rule)
        netork_ip = ""
        try:
            raw = self._exec_ssh_command(
                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
            ).strip()
            if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"):
                netork_ip = raw
        except Exception as exc:
            logger.debug("Failed to detect network origin IP: %s", exc)

        # 3. Write snmpd.conf via /tmp (no permission issues)
        conf_str = (
            "agentAddress udp:161\n"
            "rocommunity public\n"
            "sysLocation    Managed by netOrk\n"
            "sysContact     netork@localhost\n"
        )
        conf_b64 = _b64.b64encode(conf_str.encode()).decode()
        self._exec_ssh_command(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf")
        self._exec_ssh_command(
            "mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && "
            "chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf"
        )
        verify = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null").strip()
        if "agentAddress" in verify and "rocommunity" in verify:
            lines.append("[config] Wrote /etc/snmp/snmpd.conf - agentAddress udp:161, rocommunity public.")
        else:
            lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}")

        # 4. Open firewall if ufw is present
        if netork_ip:
            try:
                ufw = self._exec_ssh_command("command -v ufw 2>/dev/null").strip()
                if ufw:
                    parts = netork_ip.rsplit(".", 1)
                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
                    fw_out = self._exec_ssh_command(
                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1"
                    )
                    lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}")
            except Exception as exc:
                lines.append(f"[firewall] skipped - {exc}")

        # 5. Stop and restart snmpd cleanly (no DBus needed for stop+start)
        self._exec_ssh_command(
            "service snmpd stop 2>/dev/null; pkill -9 snmpd 2>/dev/null; true"
        )
        import time as _time
        _time.sleep(1)
        start_out = self._exec_ssh_command(
            "service snmpd start 2>&1 || systemctl start snmpd 2>&1 || true"
        )
        lines.append(f"[service] {start_out.strip()[-200:]}")

        # 6. Verify via local probe
        _time.sleep(2)
        probe_out = self._exec_ssh_command(
            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true"
        ).strip()
        _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
        success = any(t in probe_out for t in _snmp_types)
        if success:
            lines.append("[ok] SNMP probe successful - community 'public' is working.")
        else:
            lines.append(f"[warn] SNMP probe failed - output: {probe_out[:200]}")

        return {"success": success, "output": "\n".join(lines)}

    # ------------------------------------------------------------------ #
    # set_hostname
    # ------------------------------------------------------------------ #

    def set_hostname(self, new_hostname: str) -> None:
        """Set the system hostname on the Proxmox node.

        Performs all steps required for a clean rename on a Debian/Proxmox host:

        1. ``/etc/hostname``  — short hostname only.
        2. ``/etc/hosts``     — old hostname replaced by new (short or FQDN).
        3. ``/etc/mailname``  — FQDN, if the file exists.
        4. Postfix ``myhostname`` — updated via ``postconf -e`` if Postfix is
           installed.
        5. Runtime hostname  — applied immediately via ``hostname(1)`` so SSH
           and the Proxmox API see the new name without a reboot.
        6. ``/etc/pve/nodes/<old>`` → ``/etc/pve/nodes/<new>`` rename so
           VM/CT configs are found after the reboot under the new node name.
        7. Proxmox node TLS certificate — regenerated via
           ``pvenode cert create --overwrite``.

        **Important**: the Proxmox *node name* (shown in the web UI and stored
        in ``/etc/pve/nodes/<name>/``) only changes after a full reboot.
        Until then the web UI still shows the old node name, but the OS-level
        hostname and the SSL certificate already reflect the new value.

        Accepts a bare hostname (``pve-see``) or an FQDN
        (``pve-see.see.local``).
        """
        import base64 as _b64

        if "." in new_hostname:
            short_hostname, domain = new_hostname.split(".", 1)
            fqdn = new_hostname
        else:
            short_hostname = new_hostname
            domain = None
            fqdn = new_hostname

        # Derive old short hostname for substitution in /etc/hosts
        old_short = self._exec_ssh_command(
            "cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null"
        ).strip().split(".")[0]

        # ── 1. /etc/hostname ─────────────────────────────────────────────
        hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode()
        self._exec_ssh_command(f"echo {hostname_b64} | base64 -d > /etc/hostname")

        # ── 2. /etc/hosts ────────────────────────────────────────────────
        # Read the file, substitute in Python (safe for all characters),
        # write back via base64 to avoid shell-escaping issues.
        if old_short and old_short != short_hostname:
            hosts_raw = self._exec_ssh_command("cat /etc/hosts 2>/dev/null")
            # Replace whole-word occurrences of the old short hostname.
            # Both "hostname" and "hostname.domain" forms are covered because
            # the new value is the full fqdn when a domain was given.
            import re as _re
            new_replace = fqdn if domain else short_hostname
            hosts_new = _re.sub(
                r"\b" + _re.escape(old_short) + r"\b",
                new_replace,
                hosts_raw,
            )
            hosts_b64 = _b64.b64encode(hosts_new.encode()).decode()
            self._exec_ssh_command(
                f"echo {hosts_b64} | base64 -d > /tmp/_netork_hosts"
                f" && mv /tmp/_netork_hosts /etc/hosts"
                f" && chmod 644 /etc/hosts"
            )

        # ── 3. /etc/mailname ─────────────────────────────────────────────
        mailname_b64 = _b64.b64encode(f"{fqdn}\n".encode()).decode()
        self._exec_ssh_command(
            f"[ -f /etc/mailname ]"
            f" && echo {mailname_b64} | base64 -d > /etc/mailname"
            f" || true"
        )

        # ── 4. Postfix myhostname ─────────────────────────────────────────
        self._exec_ssh_command(
            f"command -v postconf >/dev/null 2>&1"
            f" && postconf -e 'myhostname = {fqdn}'"
            f" || true"
        )

        # ── 5. Apply hostname at runtime ─────────────────────────────────
        self._exec_ssh_command(f"hostname '{short_hostname}'")

        # ── 6. Migrate /etc/pve/nodes/<old> → /etc/pve/nodes/<new> ──────
        # Proxmox stores VM and CT configs under /etc/pve/nodes/<nodename>/.
        # This rename MUST happen before the reboot while pve-cluster is
        # running and pmxcfs is mounted — pmxcfs supports the rename live.
        # Without this step the node boots under the new hostname, finds an
        # empty /etc/pve/nodes/<new>/ and all VMs appear missing.
        if old_short and old_short != short_hostname:
            pve_old = f"/etc/pve/nodes/{old_short}"
            pve_new = f"/etc/pve/nodes/{short_hostname}"
            migrate_out = self._exec_ssh_command(
                f"if [ -d '{pve_old}' ] && [ ! -e '{pve_new}' ]; then"
                f"  mv '{pve_old}' '{pve_new}'"
                f"  && echo 'migrated {pve_old} -> {pve_new}';"
                f"elif [ -d '{pve_new}' ]; then"
                f"  echo 'target already exists, skipping migration';"
                f"else"
                f"  echo 'source {pve_old} not found, skipping migration';"
                f"fi"
            )
            logger.info("Proxmox node dir migration: %s", migrate_out)

        # ── 7. Regenerate Proxmox node TLS certificate ───────────────────
        # Use pvenode cert create --overwrite rather than pvecm updatecerts -f.
        # pvecm updatecerts -f restarts pve-cluster which briefly unmounts the
        # cluster filesystem (/etc/pve) and can crash running VMs.
        # pvenode cert create --overwrite only touches the node's own cert and
        # restarts pveproxy (safe while VMs are running).
        cert_out = self._exec_ssh_command(
            "pvenode cert create --overwrite 1 2>&1"
            " || { systemctl restart pveproxy 2>&1; echo 'restarted pveproxy'; }"
            " || true"
        )
        logger.info("Proxmox cert regeneration: %s", cert_out[:200])

        logger.info(
            "Proxmox set_hostname: %s → %s (fqdn: %s). "
            "Reboot required for node name to update in web UI / cluster.",
            old_short, short_hostname, fqdn,
        )

    # ------------------------------------------------------------------ #
    # _send_command — generic SSH command helper used by netOrk core
    # ------------------------------------------------------------------ #

    def _send_command(self, command: str) -> str:
        """Execute *command* on the node and return its output.

        Delegates to _exec_ssh_command so that netOrk's generic helpers
        (e.g. the reboot action) work without knowing the driver internals.
        """
        return self._exec_ssh_command(command)

    def get_disk_smart(self) -> dict:
        """Return SMART health data for all disks on this Proxmox node.

        Combines the disk list (model, size, wearout) with per-disk SMART
        data (health, temperature, percentage used).

        Returns a dict keyed by device path, e.g. {"/dev/nvme0n1": {...}}.
        """
        import re as _re

        result: dict = {}
        try:
            disks = self._node_api().disks.list.get() or []
        except Exception as exc:
            logger.debug("Failed to list disks: %s", exc)
            return result

        for disk in disks:
            dev = disk.get("devpath") or disk.get("dev")
            if not dev:
                continue
            entry: dict = {
                "model":   disk.get("model", ""),
                "serial":  disk.get("serial", ""),
                "type":    disk.get("type", ""),
                "size":    disk.get("size", 0),
                "health":  disk.get("health", "unknown").lower(),
                "wearout": disk.get("wearout"),          # NVMe wear indicator 0-100
                "temperature": None,
                "percentage_used": None,
                "available_spare": None,
                "reallocated_sectors": None,
                "power_on_hours": None,
            }
            try:
                smart = self._node_api().disks.smart.get(disk=dev) or {}
                # health from SMART endpoint may be more accurate
                if smart.get("health"):
                    entry["health"] = smart["health"].lower()

                text = smart.get("text", "")
                # Parse temperature
                m = _re.search(r"Temperature[^:]*:\s*(\d+)\s*Celsius", text)
                if m:
                    entry["temperature"] = int(m.group(1))
                # NVMe-specific
                m = _re.search(r"Percentage Used:\s*(\d+)%", text)
                if m:
                    entry["percentage_used"] = int(m.group(1))
                m = _re.search(r"Available Spare:\s*(\d+)%", text)
                if m:
                    entry["available_spare"] = int(m.group(1))
                m = _re.search(r"Power On Hours:\s*([\d,]+)", text)
                if m:
                    entry["power_on_hours"] = int(m.group(1).replace(",", ""))
                # HDD-specific SMART attributes
                for attr in smart.get("attributes", []):
                    name = attr.get("name", "").lower()
                    raw  = attr.get("raw", "")
                    try:
                        raw_int = int(str(raw).split()[0])
                    except (ValueError, TypeError):
                        raw_int = None
                    if "temperature" in name and raw_int is not None:
                        entry["temperature"] = raw_int
                    elif "reallocated" in name and "sector" in name and raw_int is not None:
                        entry["reallocated_sectors"] = raw_int
                    elif "power_on" in name and raw_int is not None:
                        entry["power_on_hours"] = raw_int
            except Exception as exc:
                logger.debug("Failed to fetch SMART data for %s: %s", dev, exc)
            result[dev] = entry

        return result

    def get_system_config(self) -> dict[str, Any]:
        """Return system-level configuration for this Proxmox node.

        Collected from ``/cluster/status`` and the node config API:

        * ``cluster_name`` (str | None) — Proxmox cluster name; ``None`` when
          the node is standalone (not part of a cluster).
        * ``cluster_nodes`` (list[str]) — hostnames of all online cluster nodes.
        * ``hostname`` (str) — this node's name as reported by the cluster.
        * ``timezone`` (str | None) — e.g. ``"Europe/Berlin"``.
        * ``ntp_servers`` (list[str]).
        * ``ssh_port`` (int) — always 22 for Proxmox nodes.
        * ``ssh_password_auth`` (bool) — ``False`` as a safe default.
        """
        cfg: dict[str, Any] = {
            "cluster_name": None,
            "cluster_nodes": [],
            "hostname": self._node_name,
            "timezone": None,
            "ntp_servers": [],
            "ssh_port": 22,
            "ssh_password_auth": False,
        }

        # Cluster name and node list from /cluster/status
        try:
            status = self._api.cluster.status.get() or []
            for entry in status:
                if entry.get("type") == "cluster":
                    cfg["cluster_name"] = entry.get("name")
                elif entry.get("type") == "node" and entry.get("online"):
                    cfg["cluster_nodes"].append(entry.get("name", ""))
        except Exception as exc:
            logger.debug("Failed to fetch cluster status for system config: %s", exc)

        # Timezone from node time API
        try:
            time_cfg = self._node_api().time.get() or {}
            cfg["timezone"] = time_cfg.get("timezone") or None
        except Exception as exc:
            logger.debug("Failed to fetch node time config: %s", exc)

        return cfg