From 50dad5a91564f164cdbe2e6bf659c804711dc6ab Mon Sep 17 00:00:00 2001 From: host Date: Fri, 24 Jul 2026 01:17:34 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=B5=D1=80=D0=B5=D0=BF=D0=B8=D1=81?= =?UTF-8?q?=D0=B0=D0=BD=20=D0=B4=D0=BB=D1=8F=20Open=20Web=20UI:=20=D0=BF?= =?UTF-8?q?=D0=BB=D0=BE=D1=81=D0=BA=D0=B8=D0=B5=20=D1=84=D1=83=D0=BD=D0=BA?= =?UTF-8?q?=D1=86=D0=B8=D0=B8=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82=D0=BE=20?= =?UTF-8?q?=D0=BA=D0=BB=D0=B0=D1=81=D1=81=D0=B0,=20=D0=BF=D1=80=D0=B5?= =?UTF-8?q?=D1=84=D0=B8=D0=BA=D1=81=20pve1=5F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proxmox_tools_pve1.py | 1076 ++++++++++++++++------------------------- 1 file changed, 413 insertions(+), 663 deletions(-) diff --git a/proxmox_tools_pve1.py b/proxmox_tools_pve1.py index d66be8f..27e9857 100644 --- a/proxmox_tools_pve1.py +++ b/proxmox_tools_pve1.py @@ -5,11 +5,10 @@ Description: Full toolset for Proxmox VE management (second server): - LXC (create, start, stop, delete) - VM (clone from template, create from ISO, manage) - Monitoring (nodes, storage, resources, tasks) -Usage: - from proxmox_tools_pve1 import ToolsPVE1 - tools = ToolsPVE1() - print(tools.connection()) - print(tools.vm_create(vmid=200, name="ubuntu", ciuser="ubuntu", cipassword="1234567")) + +Usage: Import and call functions directly. + import proxmox_tools_pve1 as pve1 + print(pve1.pve1_connection()) """ import json @@ -29,9 +28,6 @@ DEFAULT_NODE = "pve1" DEFAULT_STORAGE = "local" DEFAULT_BRIDGE = "vmbr0" -# ============================================================ -# INTERNAL HELPERS -# ============================================================ _base_url = PROXMOX_HOST.rstrip("/") + "/api2/json" _auth_header = { "Authorization": f"PVEAPIToken={PROXMOX_USER}!{PROXMOX_TOKEN_NAME}={PROXMOX_TOKEN_VALUE}" @@ -92,10 +88,7 @@ def _uptime(s): def _wait_for_task(upid, timeout=60): - """ - Wait for a Proxmox task to complete by UPID. - Returns (ok: bool, message: str). - """ + """Wait for a Proxmox task to complete by UPID.""" if not upid or not isinstance(upid, str) or not upid.startswith("UPID:"): return True, "no UPID to wait" node = DEFAULT_NODE @@ -107,10 +100,7 @@ def _wait_for_task(upid, timeout=60): if t.get("upid") == upid: status = t.get("status", "") if status == "stopped": - if t.get("exitstatus") == "OK": - return True, "OK" - else: - return False, str(t) + return (True, "OK") if t.get("exitstatus") == "OK" else (False, str(t)) break except Exception: pass @@ -119,692 +109,452 @@ def _wait_for_task(upid, timeout=60): # ============================================================ -# MAIN TOOLS CLASS +# FUNCTIONS # ============================================================ -class ToolsPVE1: - """Proxmox VE management toolset for PVE1 (192.168.31.4).""" - # ---- CONNECTION & NODES -------------------------------------------- +def pve1_connection(): + """Check Proxmox connection.""" + try: + v = _api("GET", "/version") + return "OK Proxmox VE " + v.get("version", "?") + " | node: pve1 | IP: 192.168.31.4" + except Exception as e: + return "ERR " + str(e) - @staticmethod - def connection(): - """Check Proxmox connection.""" - try: - v = _api("GET", "/version") - return ( - "OK Proxmox VE " - + v.get("version", "?") - + " | node: pve1 | IP: 192.168.31.4" - ) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def nodes(): - """Show all cluster nodes.""" - try: - items = _api("GET", "/nodes") - out = ["Nodes:"] - for n in items: - out.append(f"- {n.get('node', '?')} ({n.get('status', '?')})") - out.append( - f" CPU: {n.get('cpu', 0)*100:.1f}% | " - f"RAM: {_pct(n.get('mem'), n.get('maxmem'))}" - ) - out.append(f" Uptime: {_uptime(n.get('uptime'))}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_nodes(): + """Show all cluster nodes.""" + try: + items = _api("GET", "/nodes") + out = ["Nodes:"] + for n in items: + out.append(f"- {n.get('node', '?')} ({n.get('status', '?')})") + out.append(f" CPU: {n.get('cpu', 0)*100:.1f}% | RAM: {_pct(n.get('mem'), n.get('maxmem'))}") + out.append(f" Uptime: {_uptime(n.get('uptime'))}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - @staticmethod - def node_status(node=DEFAULT_NODE): - """Show detailed node status.""" - try: - s = _api("GET", f"/nodes/{node}/status") - out = [f"Node {node}:"] - ci = s.get("cpuinfo", {}) - out.append(f"CPU: {ci.get('model', '?')} ({s.get('cpus', 0)} cores)") - out.append(f"Load: {s.get('cpu', 0)*100:.1f}%") - la = s.get("loadavg", []) - if la: - out.append(f"Loadavg: {' / '.join(f'{float(v):.2f}' for v in la)}") - m = s.get("memory", {}) - out.append(f"RAM: {_pct(m.get('used'), m.get('total'))}") - sw = s.get("swap", {}) - if sw.get("total", 0): - out.append(f"SWAP: {_pct(sw.get('used'), sw.get('total'))}") - r = s.get("rootfs", {}) - if r: - out.append(f"Disk: {_pct(r.get('used'), r.get('total'))}") - out.append(f"Uptime: {_uptime(s.get('uptime'))}") - out.append(f"Kernel: {s.get('kversion', '?')}") - if s.get("pveversion"): - out.append(f"PVE: {s['pveversion']}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def cluster(): - """Show cluster status (quorum, members).""" - try: - items = _api("GET", "/cluster/status") - out = ["Cluster:"] - has_cluster_info = False - for item in items: - t = item.get("type", "") - if t == "cluster": - has_cluster_info = True - out.append( - f"Cluster: {item.get('name', '?')} | " - f"Quorum: {'OK' if item.get('quorate') else 'FAIL'}" - ) - elif t == "node": - out.append( - f" {item.get('name', '?')} - {item.get('status', '?')} " - f"({item.get('ip', '-')})" - ) - if not has_cluster_info: - node_count = sum(1 for i in items if i.get("type") == "node") - out.append("Cluster: single-node | Quorum: OK (N/A)") - out.append(f" Nodes: {node_count}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_node_status(node=DEFAULT_NODE): + """Show detailed node status.""" + try: + s = _api("GET", f"/nodes/{node}/status") + out = [f"Node {node}:"] + ci = s.get("cpuinfo", {}) + out.append(f"CPU: {ci.get('model', '?')} ({s.get('cpus', 0)} cores)") + out.append(f"Load: {s.get('cpu', 0)*100:.1f}%") + la = s.get("loadavg", []) + if la: + out.append(f"Loadavg: {' / '.join(f'{float(v):.2f}' for v in la)}") + m = s.get("memory", {}) + out.append(f"RAM: {_pct(m.get('used'), m.get('total'))}") + sw = s.get("swap", {}) + if sw.get("total", 0): + out.append(f"SWAP: {_pct(sw.get('used'), sw.get('total'))}") + r = s.get("rootfs", {}) + if r: + out.append(f"Disk: {_pct(r.get('used'), r.get('total'))}") + out.append(f"Uptime: {_uptime(s.get('uptime'))}") + out.append(f"Kernel: {s.get('kversion', '?')}") + if s.get("pveversion"): + out.append(f"PVE: {s['pveversion']}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - # ---- RESOURCES & STORAGE ------------------------------------------ - @staticmethod - def resources(ftype=""): - """Show all cluster resources (VMs, LXC, nodes, storage).""" - try: - p = {} - if ftype: - p["type"] = ftype - items = _api("GET", "/cluster/resources", params=p) - out = ["Resources:"] - g = {} - for r in items: - g.setdefault(r.get("type", "?"), []).append(r) - labels = { - "node": "Nodes", - "qemu": "QEMU", - "lxc": "LXC", - "storage": "Storage", - } - for t, lst in g.items(): - out.append(f"{labels.get(t, t)}: {len(lst)}") - for r in lst[:25]: - n = r.get("name") or r.get("id") or "?" - st = r.get("status", "?") - line = f" {n} - {st}" - if t in ("qemu", "lxc"): - line += f" (ID:{r.get('vmid', '?')})" - if r.get("maxmem"): - line += f" RAM:{_pct(r.get('mem'), r.get('maxmem'))}" - if r.get("cpu"): - line += f" CPU:{r.get('cpu')*100:.1f}%" - out.append(line) - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_cluster(): + """Show cluster status (quorum, members).""" + try: + items = _api("GET", "/cluster/status") + out = ["Cluster:"] + has_cluster_info = False + for item in items: + t = item.get("type", "") + if t == "cluster": + has_cluster_info = True + out.append(f"Cluster: {item.get('name', '?')} | Quorum: {'OK' if item.get('quorate') else 'FAIL'}") + elif t == "node": + out.append(f" {item.get('name', '?')} - {item.get('status', '?')} ({item.get('ip', '-')})") + if not has_cluster_info: + node_count = sum(1 for i in items if i.get("type") == "node") + out.append("Cluster: single-node | Quorum: OK (N/A)") + out.append(f" Nodes: {node_count}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - @staticmethod - def storage(): - """Show all storages.""" - try: - items = _api("GET", "/storage") - if not items: - return "No storage" - out = ["Storage:"] - for s in items: - name = s.get("storage", "?") - out.append(f"- {name} ({s.get('type', '?')})") - out.append(f" Content: {s.get('content', '?')}") - if s.get("total"): - out.append(f" Used: {_pct(s.get('used'), s.get('total'))}") - out.append(f" Free: {_fmt(s.get('avail'))}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def tasks(limit=10, node=DEFAULT_NODE): - """Show recent tasks on node.""" - try: - items = _api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) - if not items: - return f"No tasks on {node}" - out = [f"Tasks on {node}:"] - for t in items: - start = time.ctime(t.get("starttime", 0)) if t.get("starttime") else "?" - out.append( - f"- {t.get('type', '?')} | {t.get('status', '?')} | " - f"{t.get('user', '?')} | {start}" - ) - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_resources(ftype=""): + """Show all cluster resources.""" + try: + p = {"type": ftype} if ftype else {} + items = _api("GET", "/cluster/resources", params=p) + out = ["Resources:"] + g = {} + for r in items: + g.setdefault(r.get("type", "?"), []).append(r) + labels = {"node": "Nodes", "qemu": "QEMU", "lxc": "LXC", "storage": "Storage"} + for t, lst in g.items(): + out.append(f"{labels.get(t, t)}: {len(lst)}") + for r in lst[:25]: + n = r.get("name") or r.get("id") or "?" + st = r.get("status", "?") + line = f" {n} - {st}" + if t in ("qemu", "lxc"): + line += f" (ID:{r.get('vmid', '?')})" + if r.get("maxmem"): + line += f" RAM:{_pct(r.get('mem'), r.get('maxmem'))}" + if r.get("cpu"): + line += f" CPU:{r.get('cpu')*100:.1f}%" + out.append(line) + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - # ---- LXC CONTAINERS ----------------------------------------------- - @staticmethod - def lxc_list(node=DEFAULT_NODE): - """Show LXC container list.""" - try: - items = _api("GET", f"/nodes/{node}/lxc") - if not items: - return f"No LXC on {node}" - out = [f"LXC on {node}:"] - for c in items: - out.append( - f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - " - f"{c.get('status', '?')}" - ) - if c.get("cpu"): - out.append(f" CPU:{c['cpu']*100:.1f}%") - if c.get("maxmem"): - out.append(f" RAM:{_pct(c.get('mem'), c.get('maxmem'))}") - if c.get("maxdisk"): - out.append(f" Disk:{_pct(c.get('disk'), c.get('maxdisk'))}") - if c.get("uptime"): - out.append(f" Uptime:{_uptime(c['uptime'])}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_storage(): + """Show all storages.""" + try: + items = _api("GET", "/storage") + if not items: + return "No storage" + out = ["Storage:"] + for s in items: + name = s.get("storage", "?") + out.append(f"- {name} ({s.get('type', '?')})") + out.append(f" Content: {s.get('content', '?')}") + if s.get("total"): + out.append(f" Used: {_pct(s.get('used'), s.get('total'))}") + out.append(f" Free: {_fmt(s.get('avail'))}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - @staticmethod - def lxc_detail(vmid, node=DEFAULT_NODE): - """Show detailed LXC info.""" - try: - s = _api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") - c = _api("GET", f"/nodes/{node}/lxc/{vmid}/config") - name = s.get("name", c.get("hostname", "?")) - out = [f"LXC #{vmid}: {name} - {s.get('status', '?')}"] - out.append( - f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | " - f"RAM: {c.get('memory', '?')}MB" - ) - out.append(f"Disk: {c.get('rootfs', '?')} | SWAP: {c.get('swap', '?')}MB") - out.append(f"Hostname: {c.get('hostname', '?')}") - nets = [f" {k}: {v}" for k, v in c.items() if k.startswith("net")] - if nets: - out.append("Network:\n" + "\n".join(nets)) - if s.get("maxmem"): - out.append(f"RAM use: {_pct(s.get('mem'), s.get('maxmem'))}") - if s.get("cpu"): - out.append(f"CPU: {s['cpu']*100:.1f}%") - if s.get("uptime"): - out.append(f"Uptime: {_uptime(s['uptime'])}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def lxc_action(vmid, action, node=DEFAULT_NODE): - """ - Manage LXC: start, stop, shutdown, restart, suspend, resume, delete. - """ - try: - acts = { - "start": "start", - "stop": "stop", - "shutdown": "shutdown", - "restart": "reboot", - "suspend": "suspend", - "resume": "resume", - "delete": "del", - } - if action not in acts: - return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" - if action == "delete": - result = _api("DELETE", f"/nodes/{node}/lxc/{vmid}") - else: - result = _api("POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}") - return f"OK LXC #{vmid} {action}\n{result}" - except Exception as e: - return "ERR " + str(e) +def pve1_tasks(limit=10, node=DEFAULT_NODE): + """Show recent tasks on node.""" + try: + items = _api("GET", f"/nodes/{node}/tasks", params={"limit": limit}) + if not items: + return f"No tasks on {node}" + out = [f"Tasks on {node}:"] + for t in items: + start = time.ctime(t.get("starttime", 0)) if t.get("starttime") else "?" + out.append(f"- {t.get('type', '?')} | {t.get('status', '?')} | {t.get('user', '?')} | {start}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - @staticmethod - def templates(storage="local", node=DEFAULT_NODE): - """Show available LXC templates in storage.""" - try: - content = _api("GET", f"/nodes/{node}/storage/{storage}/content") - tmpl = [x for x in content if x.get("content") == "vztmpl"] - if not tmpl: - return f"No templates in {storage}" - out = [f"Templates in {storage}:"] - for t in tmpl: - out.append(f" {t.get('volid', '?')} - {_fmt(t.get('size'))}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def lxc_create( - vmid, - hostname, - ostemplate, - password="", - sshkey="", - storage=DEFAULT_STORAGE, - cores=1, - memory=512, - swap=0, - disk="8G", - bridge=DEFAULT_BRIDGE, - ip="dhcp", - netmask=24, - gw="", - dns="", - domain="", - node=DEFAULT_NODE, - unpriv=True, - start_now=True, - ): - """Create a new LXC container.""" - try: - if not hostname: - return "Error: hostname required" - if not ostemplate: - return "Error: ostemplate required" - if not password and not sshkey: - return "Error: password or sshkey required" - exist = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] - exist += [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] - if str(vmid) in exist: - return f"Error: VMID {vmid} already exists" - params = { - "vmid": vmid, - "hostname": hostname, - "ostemplate": ostemplate, - "storage": storage, - "cores": cores, - "memory": memory, - "swap": swap, - "unprivileged": 1 if unpriv else 0, - } - params["rootfs"] = disk if ":" in disk else f"{storage}:{disk}" - net = f"name=eth0,bridge={bridge}" - if ip and ip.lower() != "dhcp": - net += f",ip={ip}/{netmask}" - if gw: - net += f",gw={gw}" - else: - net += ",ip=dhcp" - params["net0"] = net - if password: - params["password"] = password - if sshkey: - params["ssh-public-keys"] = sshkey - if dns: - params["nameserver"] = dns - if domain: - params["searchdomain"] = domain - if start_now: - params["start"] = 1 - result = _api("POST", f"/nodes/{node}/lxc", data=params) - out = [f"OK LXC #{vmid} creating!"] - out.append(f" Name: {hostname}") - out.append(f" Template: {ostemplate}") - out.append(f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}") - if ip and ip.lower() != "dhcp": - out.append(f" Net: {bridge}, IP: {ip}/{netmask}") - else: - out.append(f" Net: {bridge}, IP: DHCP") - if isinstance(result, str) and result: - out.append(f"UPID: {result}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_lxc_list(node=DEFAULT_NODE): + """Show LXC container list.""" + try: + items = _api("GET", f"/nodes/{node}/lxc") + if not items: + return f"No LXC on {node}" + out = [f"LXC on {node}:"] + for c in items: + out.append(f"- {c.get('name', '?')} (ID:{c.get('vmid', '?')}) - {c.get('status', '?')}") + if c.get("cpu"): + out.append(f" CPU:{c['cpu']*100:.1f}%") + if c.get("maxmem"): + out.append(f" RAM:{_pct(c.get('mem'), c.get('maxmem'))}") + if c.get("maxdisk"): + out.append(f" Disk:{_pct(c.get('disk'), c.get('maxdisk'))}") + if c.get("uptime"): + out.append(f" Uptime:{_uptime(c['uptime'])}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - # ---- VIRTUAL MACHINES (QEMU) ------------------------------------- - @staticmethod - def vm_list(node=DEFAULT_NODE): - """Show all VM (QEMU) list.""" - try: - items = _api("GET", f"/nodes/{node}/qemu") - if not items: - return f"No VMs on {node}" - out = [f"VMs on {node}:"] - for v in items: - out.append( - f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - " - f"{v.get('status', '?')}" - ) - if v.get("cpu"): - out.append(f" CPU:{v['cpu']*100:.1f}%") - if v.get("maxmem"): - out.append(f" RAM:{_pct(v.get('mem'), v.get('maxmem'))}") - if v.get("uptime"): - out.append(f" Uptime:{_uptime(v['uptime'])}") - out.append("") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) +def pve1_lxc_detail(vmid, node=DEFAULT_NODE): + """Show detailed LXC info.""" + try: + s = _api("GET", f"/nodes/{node}/lxc/{vmid}/status/current") + c = _api("GET", f"/nodes/{node}/lxc/{vmid}/config") + name = s.get("name", c.get("hostname", "?")) + out = [f"LXC #{vmid}: {name} - {s.get('status', '?')}"] + out.append(f"OS: {c.get('ostype', '?')} | CPU: {c.get('cores', '?')} | RAM: {c.get('memory', '?')}MB") + out.append(f"Disk: {c.get('rootfs', '?')} | SWAP: {c.get('swap', '?')}MB") + out.append(f"Hostname: {c.get('hostname', '?')}") + nets = [f" {k}: {v}" for k, v in c.items() if k.startswith("net")] + if nets: + out.append("Network:\n" + "\n".join(nets)) + if s.get("maxmem"): + out.append(f"RAM use: {_pct(s.get('mem'), s.get('maxmem'))}") + if s.get("cpu"): + out.append(f"CPU: {s['cpu']*100:.1f}%") + if s.get("uptime"): + out.append(f"Uptime: {_uptime(s['uptime'])}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - @staticmethod - def vm_detail(vmid, node=DEFAULT_NODE): - """Show detailed VM info.""" - try: - s = _api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") - out = [f"VM #{vmid}: {s.get('name', '?')}"] - out.append(f"Status: {s.get('status', '?')}") - if s.get("maxmem"): - out.append(f"RAM: {_pct(s.get('mem'), s.get('maxmem'))}") - if s.get("cpu"): - out.append(f"CPU: {s['cpu']*100:.1f}%") - if s.get("maxdisk") and s.get("maxdisk", 0) > 0: - out.append(f"Disk: {_pct(s.get('disk'), s.get('maxdisk'))}") - if s.get("uptime"): - out.append(f"Uptime: {_uptime(s['uptime'])}") - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - @staticmethod - def vm_action(vmid, action, node=DEFAULT_NODE): - """ - Manage VM: start, stop, shutdown, restart, suspend, resume, delete. - """ - try: - acts = { - "start": "start", - "stop": "stop", - "shutdown": "shutdown", - "restart": "reboot", - "suspend": "suspend", - "resume": "resume", - "delete": "del", - } - if action not in acts: - return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" - if action == "delete": - result = _api("DELETE", f"/nodes/{node}/qemu/{vmid}") - else: - result = _api( - "POST", f"/nodes/{node}/qemu/{vmid}/status/{acts[action]}" - ) - return f"OK VM #{vmid} {action}\n{result}" - except Exception as e: - return "ERR " + str(e) +def pve1_lxc_action(vmid, action, node=DEFAULT_NODE): + """Manage LXC: start, stop, shutdown, restart, suspend, resume, delete.""" + try: + acts = {"start": "start", "stop": "stop", "shutdown": "shutdown", "restart": "reboot", + "suspend": "suspend", "resume": "resume", "delete": "del"} + if action not in acts: + return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" + if action == "delete": + result = _api("DELETE", f"/nodes/{node}/lxc/{vmid}") + else: + result = _api("POST", f"/nodes/{node}/lxc/{vmid}/status/{acts[action]}") + return f"OK LXC #{vmid} {action}\n{result}" + except Exception as e: + return "ERR " + str(e) - @staticmethod - def vm_create_from_template( - vmid, - name, - template_vmid=1000, - full=True, - memory=None, - cores=None, - storage=None, - target_node=None, - pool=None, - start_now=True, - ciuser="", - cipassword="", - sshkeys="", - node=DEFAULT_NODE, - ): - """ - Create a new VM by cloning from a template. - Default template: VM #1000 (cloud-init). - Cloud-init params (for guest OS user setup): - ciuser: username in the guest (e.g. "ubuntu") - cipassword: password for the user - sshkeys: SSH public keys (newline-separated string) +def pve1_templates(storage="local", node=DEFAULT_NODE): + """Show available LXC templates in storage.""" + try: + content = _api("GET", f"/nodes/{node}/storage/{storage}/content") + tmpl = [x for x in content if x.get("content") == "vztmpl"] + if not tmpl: + return f"No templates in {storage}" + out = [f"Templates in {storage}:"] + for t in tmpl: + out.append(f" {t.get('volid', '?')} - {_fmt(t.get('size'))}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - FIXED: "start" parameter is NOT sent to /clone - - startup is done via separate API call after clone completes. - """ - try: - if not vmid: - return "Error: vmid required" - if not name: - return "Error: name required" - # Check VMID availability - exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] - exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] - if str(vmid) in exist_vm + exist_lxc: - return f"Error: VMID {vmid} already exists" +def pve1_lxc_create(vmid, hostname, ostemplate, password="", sshkey="", storage=DEFAULT_STORAGE, + cores=1, memory=512, swap=0, disk="8G", bridge=DEFAULT_BRIDGE, + ip="dhcp", netmask=24, gw="", dns="", domain="", node=DEFAULT_NODE, + unpriv=True, start_now=True): + """Create a new LXC container.""" + try: + if not hostname: + return "Error: hostname required" + if not password and not sshkey: + return "Error: password or sshkey required" + exist = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + exist += [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + if str(vmid) in exist: + return f"Error: VMID {vmid} already exists" + params = {"vmid": vmid, "hostname": hostname, "ostemplate": ostemplate, + "storage": storage, "cores": cores, "memory": memory, "swap": swap, + "unprivileged": 1 if unpriv else 0} + params["rootfs"] = disk if ":" in disk else f"{storage}:{disk}" + net = f"name=eth0,bridge={bridge}" + if ip and ip.lower() != "dhcp": + net += f",ip={ip}/{netmask}" + if gw: + net += f",gw={gw}" + else: + net += ",ip=dhcp" + params["net0"] = net + if password: + params["password"] = password + if sshkey: + params["ssh-public-keys"] = sshkey + if dns: + params["nameserver"] = dns + if domain: + params["searchdomain"] = domain + if start_now: + params["start"] = 1 + result = _api("POST", f"/nodes/{node}/lxc", data=params) + out = [f"OK LXC #{vmid} creating!", f" Name: {hostname}", f" Template: {ostemplate}", + f" CPU: {cores} cores, RAM: {memory}MB, Disk: {disk}"] + out.append(f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}") + if isinstance(result, str) and result: + out.append(f"UPID: {result}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - # Step 1: Clone (WITHOUT "start" parameter!) - params = { - "newid": vmid, - "name": name, - "full": 1 if full else 0, - } - if storage: - params["storage"] = storage - if target_node: - params["target"] = target_node - if pool: - params["pool"] = pool - result = _api( - "POST", f"/nodes/{node}/qemu/{template_vmid}/clone", data=params - ) +def pve1_vm_list(node=DEFAULT_NODE): + """Show all VM (QEMU) list.""" + try: + items = _api("GET", f"/nodes/{node}/qemu") + if not items: + return f"No VMs on {node}" + out = [f"VMs on {node}:"] + for v in items: + out.append(f"- {v.get('name', '?')} (ID:{v.get('vmid', '?')}) - {v.get('status', '?')}") + if v.get("cpu"): + out.append(f" CPU:{v['cpu']*100:.1f}%") + if v.get("maxmem"): + out.append(f" RAM:{_pct(v.get('mem'), v.get('maxmem'))}") + if v.get("uptime"): + out.append(f" Uptime:{_uptime(v['uptime'])}") + out.append("") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - out = [f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!"] - out.append(f" Type: {'full' if full else 'linked'} clone") - if storage: - out.append(f" Storage: {storage}") - upid = result if isinstance(result, str) else "" - if upid: - out.append(f"UPID: {upid}") +def pve1_vm_detail(vmid, node=DEFAULT_NODE): + """Show detailed VM info.""" + try: + s = _api("GET", f"/nodes/{node}/qemu/{vmid}/status/current") + out = [f"VM #{vmid}: {s.get('name', '?')}", f"Status: {s.get('status', '?')}"] + if s.get("maxmem"): + out.append(f"RAM: {_pct(s.get('mem'), s.get('maxmem'))}") + if s.get("cpu"): + out.append(f"CPU: {s['cpu']*100:.1f}%") + if s.get("maxdisk") and s.get("maxdisk", 0) > 0: + out.append(f"Disk: {_pct(s.get('disk'), s.get('maxdisk'))}") + if s.get("uptime"): + out.append(f"Uptime: {_uptime(s['uptime'])}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - # Step 2: Wait for clone to finish - out.append("Waiting for clone to finish...") - ok, msg = _wait_for_task(upid, timeout=120) - if not ok: - out.append(f" WARN Clone may not have finished: {msg}") - else: - out.append(" Clone done") - # Step 3: Apply settings (resources + cloud-init) - config_params = {} - if memory is not None: - config_params["memory"] = memory - if cores is not None: - config_params["cores"] = cores - if ciuser: - config_params["ciuser"] = ciuser - if cipassword: - config_params["cipassword"] = cipassword - if sshkeys: - config_params["sshkeys"] = sshkeys +def pve1_vm_action(vmid, action, node=DEFAULT_NODE): + """Manage VM: start, stop, shutdown, restart, suspend, resume, delete.""" + try: + acts = {"start": "start", "stop": "stop", "shutdown": "shutdown", "restart": "reboot", + "suspend": "suspend", "resume": "resume", "delete": "del"} + if action not in acts: + return f"Unknown action: {action}. Allowed: {', '.join(acts.keys())}" + if action == "delete": + result = _api("DELETE", f"/nodes/{node}/qemu/{vmid}") + else: + result = _api("POST", f"/nodes/{node}/qemu/{vmid}/status/{acts[action]}") + return f"OK VM #{vmid} {action}\n{result}" + except Exception as e: + return "ERR " + str(e) - if config_params: - out.append("Applying settings...") - try: - _api( - "PUT", - f"/nodes/{node}/qemu/{vmid}/config", - data=config_params, - ) - if memory is not None: - out.append(f" RAM -> {memory} MB") - if cores is not None: - out.append(f" CPU -> {cores} cores") - if ciuser: - out.append(f" Cloud-init user: {ciuser}") - if cipassword: - out.append(" Cloud-init password: (set)") - if sshkeys: - out.append(" Cloud-init SSH keys: (set)") - out.append(" Settings applied") - except Exception as e2: - out.append(f" WARN Could not apply settings: {e2}") - # Step 4: Start VM (separate call!) - if start_now: - out.append("Starting VM...") - try: - _api("POST", f"/nodes/{node}/qemu/{vmid}/status/start") - out.append(f" VM #{vmid} started!") - except Exception as e3: - out.append(f" WARN Could not start: {e3}") - else: - out.append(f" VM #{vmid} created (not started)") +def pve1_vm_create_from_template(vmid, name, template_vmid=1000, full=True, memory=None, + cores=None, storage=None, target_node=None, pool=None, + start_now=True, ciuser="", cipassword="", sshkeys="", node=DEFAULT_NODE): + """Create a new VM by cloning from template. Default: VM #1000 (cloud-init).""" + try: + if not vmid: + return "Error: vmid required" + if not name: + return "Error: name required" + exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + if str(vmid) in exist_vm + exist_lxc: + return f"Error: VMID {vmid} already exists" + params = {"newid": vmid, "name": name, "full": 1 if full else 0} + if storage: + params["storage"] = storage + if target_node: + params["target"] = target_node + if pool: + params["pool"] = pool + result = _api("POST", f"/nodes/{node}/qemu/{template_vmid}/clone", data=params) + out = [f"OK VM #{vmid}: {name} cloning from template #{template_vmid}!", + f" Type: {'full' if full else 'linked'} clone"] + if storage: + out.append(f" Storage: {storage}") + upid = result if isinstance(result, str) else "" + if upid: + out.append(f"UPID: {upid}") + out.append("Waiting for clone to finish...") + ok, msg = _wait_for_task(upid, timeout=120) + out.append(f" {'Clone done' if ok else f'WARN Clone may not have finished: {msg}'}") + config_params = {} + if memory is not None: + config_params["memory"] = memory + if cores is not None: + config_params["cores"] = cores + if ciuser: + config_params["ciuser"] = ciuser + if cipassword: + config_params["cipassword"] = cipassword + if sshkeys: + config_params["sshkeys"] = sshkeys + if config_params: + out.append("Applying settings...") + try: + _api("PUT", f"/nodes/{node}/qemu/{vmid}/config", data=config_params) + if memory is not None: + out.append(f" RAM -> {memory} MB") + if cores is not None: + out.append(f" CPU -> {cores} cores") + if ciuser: + out.append(f" Cloud-init user: {ciuser}") + if cipassword: + out.append(" Cloud-init password: (set)") + if sshkeys: + out.append(" Cloud-init SSH keys: (set)") + out.append(" Settings applied") + except Exception as e2: + out.append(f" WARN Could not apply settings: {e2}") + if start_now: + out.append("Starting VM...") + try: + _api("POST", f"/nodes/{node}/qemu/{vmid}/status/start") + out.append(f" VM #{vmid} started!") + except Exception as e3: + out.append(f" WARN Could not start: {e3}") + else: + out.append(f" VM #{vmid} created (not started)") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) - return "\n".join(out) - except Exception as e: - return "ERR " + str(e) - - @staticmethod - def vm_create( - vmid, - name, - template_vmid=1000, - memory=4096, - cores=2, - disk_size="32G", - disk_storage="local-lvm", - iso="", - bridge=DEFAULT_BRIDGE, - ip="dhcp", - netmask=24, - gw="", - ostype="l26", - agent=1, - ciuser="", - cipassword="", - sshkeys="", - node=DEFAULT_NODE, - start_now=True, - ): - """ - Universal VM creation. - - Modes: - 1. From template (default): clones VM #1000 (cloud-init) - 2. From ISO: template_vmid=None, iso="local:iso/image.iso" - - Cloud-init params (for mode 1): - ciuser: guest OS username - cipassword: guest OS password - sshkeys: SSH public keys - - Examples: - # Clone from template with cloud-init user: - tools.vm_create( - vmid=200, - name="web-server", - memory=8192, - cores=4, - ciuser="ubuntu", - cipassword="mypassword", - start_now=True, - ) - - # Create from ISO: - tools.vm_create( - vmid=202, - name="debian-server", - template_vmid=None, - iso="local:iso/debian-12.iso", - ip="192.168.31.60", - gw="192.168.31.1", - ) - """ - try: - # Mode 1: clone from template - if template_vmid: - return ToolsPVE1.vm_create_from_template( - vmid=vmid, - name=name, - template_vmid=template_vmid, - full=True, - memory=memory, - cores=cores, - storage=disk_storage, - start_now=start_now, - ciuser=ciuser, - cipassword=cipassword, - sshkeys=sshkeys, - node=node, - ) - - # Mode 2: create from scratch (ISO) - if not name: - return "Error: name required" - if not vmid: - return "Error: vmid required" - - exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] - exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] - if str(vmid) in exist_vm + exist_lxc: - return f"Error: VMID {vmid} already exists" - - params = { - "vmid": vmid, - "name": name, - "memory": memory, - "cores": cores, - "sockets": 1, - "ostype": ostype, - "agent": str(agent), - } - - # Disk - if ":" in disk_size: - params["virtio0"] = disk_size - else: - params["virtio0"] = f"{disk_storage}:{disk_size}" - - # ISO - if iso: - params["ide2"] = f"{iso},media=cdrom" - params["boot"] = "order=ide2;virtio0" - - # Network - net = f"name=eth0,bridge={bridge}" - if ip and ip.lower() != "dhcp": - net += f",ip={ip}/{netmask}" - if gw: - net += f",gw={gw}" - else: - net += ",ip=dhcp" - params["net0"] = net - - if start_now: - params["start"] = 1 - - result = _api("POST", f"/nodes/{node}/qemu", data=params) - - out = [f"OK VM #{vmid}: {name} creating!"] - out.append(f" CPU: {cores} cores, RAM: {memory} MB") - out.append(f" Disk: {params['virtio0']}") - if iso: - out.append(f" ISO: {iso}") - out.append( - f" Net: {bridge}, " - f"IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}" - ) - out.append(f" QEMU Agent: {'on' if agent else 'off'}") - if isinstance(result, str) and result: - out.append(f"UPID: {result}") - return "\n".join(out) - - except Exception as e: - return "ERR " + str(e) +def pve1_vm_create(vmid, name, template_vmid=1000, memory=4096, cores=2, + disk_size="32G", disk_storage="local-lvm", iso="", + bridge=DEFAULT_BRIDGE, ip="dhcp", netmask=24, gw="", + ostype="l26", agent=1, ciuser="", cipassword="", sshkeys="", + node=DEFAULT_NODE, start_now=True): + """Universal VM creation. Mode 1: clone from template. Mode 2: from ISO (template_vmid=None).""" + try: + if template_vmid: + return pve1_vm_create_from_template(vmid=vmid, name=name, template_vmid=template_vmid, + full=True, memory=memory, cores=cores, storage=disk_storage, + start_now=start_now, ciuser=ciuser, cipassword=cipassword, sshkeys=sshkeys, node=node) + if not name or not vmid: + return "Error: name and vmid required" + exist_vm = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/qemu")] + exist_lxc = [str(x.get("vmid")) for x in _api("GET", f"/nodes/{node}/lxc")] + if str(vmid) in exist_vm + exist_lxc: + return f"Error: VMID {vmid} already exists" + params = {"vmid": vmid, "name": name, "memory": memory, "cores": cores, + "sockets": 1, "ostype": ostype, "agent": str(agent)} + params["virtio0"] = disk_size if ":" in disk_size else f"{disk_storage}:{disk_size}" + if iso: + params["ide2"] = f"{iso},media=cdrom" + params["boot"] = "order=ide2;virtio0" + net = f"name=eth0,bridge={bridge}" + if ip and ip.lower() != "dhcp": + net += f",ip={ip}/{netmask}" + if gw: + net += f",gw={gw}" + else: + net += ",ip=dhcp" + params["net0"] = net + if start_now: + params["start"] = 1 + result = _api("POST", f"/nodes/{node}/qemu", data=params) + out = [f"OK VM #{vmid}: {name} creating!", f" CPU: {cores} cores, RAM: {memory} MB", + f" Disk: {params['virtio0']}"] + if iso: + out.append(f" ISO: {iso}") + out.append(f" Net: {bridge}, IP: {'DHCP' if ip == 'dhcp' else f'{ip}/{netmask}'}") + out.append(f" QEMU Agent: {'on' if agent else 'off'}") + if isinstance(result, str) and result: + out.append(f"UPID: {result}") + return "\n".join(out) + except Exception as e: + return "ERR " + str(e) # ============================================================ # TEST # ============================================================ if __name__ == "__main__": - tools = ToolsPVE1() - print(tools.connection()) + print(pve1_connection()) print() - print(tools.vm_list()) + print(pve1_vm_list()) print() - print(tools.storage()) + print(pve1_storage())