duffyduckandClaude Opus 4.8
0efb8d0848
- Brain: project_vms.py (Registry /shared/config/project_vms.json pro Projekt) + Endpoints GET/POST/DELETE /projects/<id>/vms + boot/stop (via SSH aria-wohnung aria-vm auf dem Host; Status aus `aria-vm list`). - ARIA-Tools vm_register/vm_list (aufs Request-Projekt) + Seed-Regel: nach dem VM-Bau registrieren, damit sie in Stefans Desktop-Panel auftaucht. - App: brainApi VM-Methoden + ProjectVm-Typ. Neues DesktopTile — pro Projekt die VM-Liste (leer bis registriert), Start/Stop/Verbinden; Verbinden oeffnet noVNC (VncTile mit dem VNC-Port der VM). Deck rendert DesktopTile statt VncTile. py/tsc clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""
|
|
project_vms — Registry der QEMU-VMs PRO PROJEKT.
|
|
|
|
Persistenz: /shared/config/project_vms.json → { project_id: [ {vm}, ... ] }.
|
|
Eine VM = {name, arch, iso, vnc_display, mem, created_at, updated_at}. Der echte
|
|
Start/Stop laeuft ueber `aria-vm` auf dem Host (SSH aria-wohnung, siehe main.py);
|
|
diese Datei haelt nur die Zuordnung VM ↔ Projekt + die Startparameter, damit die
|
|
Liste im Desktop-Panel der App pro Projekt erscheint (auch wenn leer).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
VMS_FILE = Path(os.environ.get("PROJECT_VMS_FILE", "/shared/config/project_vms.json"))
|
|
NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,40}$")
|
|
VALID_ARCH = {"x86_64", "amd64", "i386", "i686", "x86", "arm", "aarch64",
|
|
"mips", "mipsel", "mips64", "ppc", "ppc64", "riscv64", "sparc"}
|
|
|
|
|
|
def _load() -> dict:
|
|
if not VMS_FILE.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(VMS_FILE.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save(data: dict) -> None:
|
|
VMS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = VMS_FILE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
tmp.replace(VMS_FILE)
|
|
|
|
|
|
def list_vms(project_id: str) -> list[dict]:
|
|
return _load().get(project_id or "", [])
|
|
|
|
|
|
def get_vm(project_id: str, name: str) -> Optional[dict]:
|
|
for v in list_vms(project_id):
|
|
if v.get("name") == name:
|
|
return v
|
|
return None
|
|
|
|
|
|
def add_vm(project_id: str, name: str, arch: str, iso: str = "",
|
|
vnc_display: int = 1, mem: int = 1024) -> dict:
|
|
if not NAME_RE.match(name or ""):
|
|
raise ValueError(f"Ungueltiger VM-Name: {name!r} (nur a-z0-9_-, max 40)")
|
|
if arch not in VALID_ARCH:
|
|
raise ValueError(f"Unbekannte Architektur: {arch!r}")
|
|
data = _load()
|
|
lst = data.setdefault(project_id or "", [])
|
|
now = int(time.time())
|
|
for v in lst:
|
|
if v.get("name") == name:
|
|
v.update({"arch": arch, "iso": iso, "vnc_display": int(vnc_display),
|
|
"mem": int(mem), "updated_at": now})
|
|
_save(data)
|
|
return v
|
|
vm = {"name": name, "arch": arch, "iso": iso, "vnc_display": int(vnc_display),
|
|
"mem": int(mem), "created_at": now, "updated_at": now}
|
|
lst.append(vm)
|
|
_save(data)
|
|
return vm
|
|
|
|
|
|
def remove_vm(project_id: str, name: str) -> bool:
|
|
data = _load()
|
|
lst = data.get(project_id or "")
|
|
if not lst:
|
|
return False
|
|
new = [v for v in lst if v.get("name") != name]
|
|
if len(new) == len(lst):
|
|
return False
|
|
data[project_id or ""] = new
|
|
_save(data)
|
|
return True
|