duffyduckandClaude Opus 5
8f1bf037cd
Nach dem Speichern musste man bisher in die Konsole wechseln, um die Konfiguration mit "systemctl reload pvesnap" scharf zu schalten. Die Gruppenliste kann das jetzt selbst: * 'r' laedt den Dienst neu; ungespeicherte Aenderungen werden vorher zum Speichern angeboten, ein gestoppter Dienst zum Starten, und wenn das Neuladen scheitert, wird ein Neustart angeboten. * Nach 's' fragt der Editor direkt, ob neu geladen werden soll. * Die Kopfzeile zeigt laufend den Zustand des Dienstes. * Die Tastenleiste bricht auf schmalen Terminals auf zwei Zeilen um, statt hinten abgeschnitten zu werden. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1050 lines
42 KiB
Python
1050 lines
42 KiB
Python
"""ncurses-Editor fuer die pvesnap-Konfiguration.
|
|
|
|
Aufruf: pvesnap config
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
import os
|
|
import subprocess
|
|
|
|
from .config import (WEEKDAY_NAMES, Config, ConfigError, Group, clone_group,
|
|
load_config, parse_time_of_day, save_config)
|
|
from .naming import effective_slug
|
|
from .proxmox import Proxmox, ProxmoxError, pvesh_available
|
|
from .schedule import describe
|
|
from .util import (format_duration, format_vmid_list, parse_duration,
|
|
parse_list, parse_vmid_list, truncate)
|
|
|
|
SCHEDULE_CHOICES = [
|
|
("interval", "Intervall (z.B. alle 30 Minuten)"),
|
|
("hourly", "stuendlich"),
|
|
("daily", "taeglich"),
|
|
("weekly", "woechentlich"),
|
|
("monthly", "monatlich"),
|
|
("yearly", "jaehrlich"),
|
|
]
|
|
|
|
TYPE_CHOICES = [
|
|
([], "alle (VMs und Container)"),
|
|
(["qemu"], "nur VMs (QEMU/KVM)"),
|
|
(["lxc"], "nur Container (LXC)"),
|
|
]
|
|
|
|
C_HEADER, C_FOOTER, C_SEL, C_WARN, C_OK, C_DIM = 1, 2, 3, 4, 5, 6
|
|
|
|
SERVICE = "pvesnap.service"
|
|
|
|
# Tastenleiste der Gruppenliste: (Taste, lange Beschriftung, kurze Beschriftung)
|
|
GROUP_KEYS = [
|
|
("Enter", "Bearbeiten", "Bearb."),
|
|
("n", "Neu", "Neu"),
|
|
("c", "Kopieren", "Kopie"),
|
|
("d", "Loeschen", "Loesch"),
|
|
("Leer", "An/Aus", "An/Aus"),
|
|
("g", "Global", "Global"),
|
|
("v", "VMs", "VMs"),
|
|
("s", "Speichern", "Speich."),
|
|
("r", "Dienst neu laden", "Reload"),
|
|
("q", "Ende", "Ende"),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# systemd
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _systemctl(*args, **kwargs):
|
|
"""Ruft systemctl auf und gibt (Rueckgabewert, Ausgabe) zurueck."""
|
|
try:
|
|
proc = subprocess.run(["systemctl"] + list(args), stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, timeout=kwargs.get("timeout", 30))
|
|
return proc.returncode, proc.stdout.decode("utf-8", "replace").strip()
|
|
except FileNotFoundError:
|
|
return 127, "systemctl nicht gefunden"
|
|
except subprocess.TimeoutExpired:
|
|
return 124, "Zeitueberschreitung beim Aufruf von systemctl"
|
|
|
|
|
|
def service_state():
|
|
"""('geladen?', 'Zustand') des pvesnap-Dienstes, z.B. (True, 'active')."""
|
|
code, output = _systemctl("show", "-p", "LoadState", "-p", "ActiveState", SERVICE)
|
|
if code not in (0, 3):
|
|
return None, output.splitlines()[0] if output else "unbekannt"
|
|
values = {}
|
|
for line in output.splitlines():
|
|
if "=" in line:
|
|
key, _, value = line.partition("=")
|
|
values[key.strip()] = value.strip()
|
|
loaded = values.get("LoadState") == "loaded"
|
|
return loaded, values.get("ActiveState", "unbekannt")
|
|
|
|
|
|
def service_text():
|
|
loaded, state = service_state()
|
|
if loaded is None:
|
|
return "Dienst: %s" % state
|
|
if not loaded:
|
|
return "Dienst: nicht installiert"
|
|
return {"active": "Dienst: laeuft",
|
|
"inactive": "Dienst: gestoppt",
|
|
"failed": "Dienst: FEHLER",
|
|
"activating": "Dienst: startet"}.get(state, "Dienst: %s" % state)
|
|
|
|
|
|
def _keybar(width, items):
|
|
"""Tastenleiste als Liste von Zeilen, passend zur Breite des Terminals.
|
|
|
|
Reicht der Platz nicht fuer eine Zeile, wird auf zwei Zeilen umgebrochen;
|
|
bei ganz schmalen Terminals bleiben nur noch die Tasten uebrig.
|
|
"""
|
|
def join(entries, index):
|
|
return " | ".join("%s %s" % (entry[0], entry[index]) for entry in entries)
|
|
|
|
long_text = join(items, 1)
|
|
if len(long_text) + 2 <= width:
|
|
return [long_text]
|
|
|
|
half = (len(items) + 1) // 2
|
|
first, second = join(items[:half], 1), join(items[half:], 1)
|
|
if max(len(first), len(second)) + 2 <= width:
|
|
return [first, second]
|
|
|
|
short_text = join(items, 2)
|
|
if len(short_text) + 2 <= width:
|
|
return [short_text]
|
|
|
|
first, second = join(items[:half], 2), join(items[half:], 2)
|
|
if max(len(first), len(second)) + 2 <= width:
|
|
return [first, second]
|
|
return [" ".join(entry[0] for entry in items)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Zeichen-Helfer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _put(win, y, x, text, attr=0):
|
|
"""addstr, das bei zu kleinem Terminal nicht abstuerzt."""
|
|
height, width = win.getmaxyx()
|
|
if y < 0 or y >= height or x >= width:
|
|
return
|
|
try:
|
|
win.addstr(y, x, str(text)[: max(0, width - x - 1)], attr)
|
|
except curses.error:
|
|
pass
|
|
|
|
|
|
def _fill(win, y, text, attr):
|
|
height, width = win.getmaxyx()
|
|
if y < 0 or y >= height:
|
|
return
|
|
try:
|
|
win.addstr(y, 0, str(text).ljust(width - 1)[: width - 1], attr)
|
|
except curses.error:
|
|
pass
|
|
|
|
|
|
def _read_key(win):
|
|
try:
|
|
return win.get_wch()
|
|
except curses.error:
|
|
return None
|
|
except KeyboardInterrupt:
|
|
return "\x1b"
|
|
|
|
|
|
def _is_enter(key):
|
|
return key in ("\n", "\r", curses.KEY_ENTER)
|
|
|
|
|
|
def _is_escape(key):
|
|
return key == "\x1b"
|
|
|
|
|
|
def _is_backspace(key):
|
|
return key in ("\x7f", "\b", "\x08", curses.KEY_BACKSPACE)
|
|
|
|
|
|
class Editor:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.config = None
|
|
self.dirty = False
|
|
self._guests = None
|
|
self._inventory_error = ""
|
|
self._service_cache = None
|
|
|
|
# -- Daten ------------------------------------------------------------
|
|
|
|
def load(self):
|
|
if os.path.exists(self.path):
|
|
self.config = load_config(self.path)
|
|
else:
|
|
self.config = Config(path=self.path)
|
|
self.dirty = True
|
|
|
|
def guests(self, refresh=False):
|
|
"""Inventar der VMs; wird nur bei Bedarf einmal geholt."""
|
|
if self._guests is not None and not refresh:
|
|
return self._guests
|
|
self._guests = []
|
|
self._inventory_error = ""
|
|
if not pvesh_available():
|
|
self._inventory_error = ("'pvesh' nicht gefunden - VM-Liste nur auf einem "
|
|
"Proxmox-Host verfuegbar.")
|
|
return self._guests
|
|
try:
|
|
self._guests = Proxmox().inventory(refresh=True)
|
|
except ProxmoxError as exc:
|
|
self._inventory_error = str(exc)
|
|
return self._guests
|
|
|
|
# -- Rahmen -----------------------------------------------------------
|
|
|
|
def _frame(self, win, title, subtitle, keys):
|
|
win.erase()
|
|
height, width = win.getmaxyx()
|
|
_fill(win, 0, " pvesnap - %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
marker = "*" if self.dirty else " "
|
|
_fill(win, 1, " %s%s" % (marker, subtitle), curses.color_pair(C_DIM))
|
|
lines = list(keys) if isinstance(keys, (list, tuple)) else [keys]
|
|
for offset, line in enumerate(reversed(lines)):
|
|
_fill(win, height - 1 - offset, " " + line, curses.color_pair(C_FOOTER))
|
|
return height, width
|
|
|
|
def _message(self, win, text, error=False):
|
|
height, _ = win.getmaxyx()
|
|
attr = curses.color_pair(C_WARN if error else C_OK) | curses.A_BOLD
|
|
_fill(win, height - 2, " " + str(text) + " [Taste druecken]", attr)
|
|
win.refresh()
|
|
_read_key(win)
|
|
|
|
def _confirm(self, win, question):
|
|
height, _ = win.getmaxyx()
|
|
_fill(win, height - 2, " %s [j/n]" % question,
|
|
curses.color_pair(C_WARN) | curses.A_BOLD)
|
|
win.refresh()
|
|
while True:
|
|
key = _read_key(win)
|
|
if key in ("j", "J", "y", "Y"):
|
|
return True
|
|
if key in ("n", "N") or _is_escape(key):
|
|
return False
|
|
|
|
def _prompt(self, win, label, initial=""):
|
|
"""Einzeilige Eingabe am unteren Rand. None = abgebrochen."""
|
|
height, width = win.getmaxyx()
|
|
buffer = list(str(initial))
|
|
curses.curs_set(1)
|
|
try:
|
|
while True:
|
|
text = "".join(buffer)
|
|
prompt = " %s: " % label
|
|
visible = width - len(prompt) - 2
|
|
shown = text[-visible:] if visible > 0 and len(text) > visible else text
|
|
_fill(win, height - 2, prompt + shown, curses.color_pair(C_SEL))
|
|
try:
|
|
win.move(height - 2, min(width - 2, len(prompt) + len(shown)))
|
|
except curses.error:
|
|
pass
|
|
win.refresh()
|
|
|
|
key = _read_key(win)
|
|
if key is None:
|
|
continue
|
|
if _is_enter(key):
|
|
return "".join(buffer).strip()
|
|
if _is_escape(key):
|
|
return None
|
|
if _is_backspace(key):
|
|
if buffer:
|
|
buffer.pop()
|
|
continue
|
|
if key == "\x15": # Strg-U
|
|
buffer = []
|
|
continue
|
|
if isinstance(key, str) and key.isprintable():
|
|
buffer.append(key)
|
|
finally:
|
|
curses.curs_set(0)
|
|
|
|
def _choose(self, win, title, entries, current=0):
|
|
"""Kleines Auswahlfenster. entries: Liste von (wert, beschriftung)."""
|
|
height, width = win.getmaxyx()
|
|
box_h = min(len(entries) + 4, height - 2)
|
|
box_w = min(max([len(title)] + [len(e[1]) for e in entries]) + 8, width - 2)
|
|
top = max(0, (height - box_h) // 2)
|
|
left = max(0, (width - box_w) // 2)
|
|
box = curses.newwin(box_h, box_w, top, left)
|
|
box.keypad(True)
|
|
index = max(0, min(current, len(entries) - 1))
|
|
|
|
while True:
|
|
box.erase()
|
|
box.box()
|
|
_put(box, 0, 2, " %s " % title, curses.A_BOLD)
|
|
for row, (_value, label) in enumerate(entries[: box_h - 4]):
|
|
attr = curses.color_pair(C_SEL) if row == index else 0
|
|
_put(box, row + 2, 2, " " + label.ljust(box_w - 5), attr)
|
|
box.refresh()
|
|
|
|
key = _read_key(box)
|
|
if key in (curses.KEY_UP, "k"):
|
|
index = (index - 1) % len(entries)
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
index = (index + 1) % len(entries)
|
|
elif _is_enter(key):
|
|
return entries[index][0]
|
|
elif _is_escape(key) or key == "q":
|
|
return None
|
|
|
|
# -- Hauptschleife ----------------------------------------------------
|
|
|
|
def run(self, stdscr):
|
|
curses.curs_set(0)
|
|
if hasattr(curses, "set_escdelay"):
|
|
curses.set_escdelay(25)
|
|
try:
|
|
curses.start_color()
|
|
curses.use_default_colors()
|
|
curses.init_pair(C_HEADER, curses.COLOR_WHITE, curses.COLOR_BLUE)
|
|
curses.init_pair(C_FOOTER, curses.COLOR_BLACK, curses.COLOR_WHITE)
|
|
curses.init_pair(C_SEL, curses.COLOR_BLACK, curses.COLOR_CYAN)
|
|
curses.init_pair(C_WARN, curses.COLOR_WHITE, curses.COLOR_RED)
|
|
curses.init_pair(C_OK, curses.COLOR_BLACK, curses.COLOR_GREEN)
|
|
curses.init_pair(C_DIM, -1, -1)
|
|
except curses.error:
|
|
pass
|
|
stdscr.keypad(True)
|
|
self.screen_groups(stdscr)
|
|
|
|
# -- Bildschirm: Gruppenliste ----------------------------------------
|
|
|
|
def screen_groups(self, win):
|
|
index = 0
|
|
while True:
|
|
groups = self.config.groups
|
|
index = max(0, min(index, max(0, len(groups) - 1)))
|
|
_, width_now = win.getmaxyx()
|
|
status = self._service_text()
|
|
# Der Dienststatus soll immer sichtbar bleiben - notfalls wird der
|
|
# Pfad vorne gekuerzt.
|
|
room = max(12, width_now - len(status) - 15)
|
|
path_text = (self.path if len(self.path) <= room
|
|
else "…" + self.path[-(room - 1):])
|
|
height, width = self._frame(
|
|
win, "Gruppen", "Datei: %s [%s]" % (path_text, status),
|
|
_keybar(width_now, GROUP_KEYS))
|
|
|
|
_put(win, 3, 2, "Gruppe".ljust(18) + "Aktiv " + "Zeitplan".ljust(30)
|
|
+ "Behalte".ljust(20) + "Auswahl", curses.A_BOLD)
|
|
_put(win, 4, 2, "-" * max(0, width - 4), curses.color_pair(C_DIM))
|
|
|
|
if not groups:
|
|
_put(win, 6, 4, "Noch keine Gruppe angelegt - mit 'n' eine erste anlegen.")
|
|
_put(win, 8, 4, "Beispiel: 'stuendlich' alle 1h, 24 Snapshots behalten;")
|
|
_put(win, 9, 4, " 'taeglich' um 02:30, 14 Tage Vorhaltezeit.")
|
|
visible = max(1, height - 8)
|
|
start = max(0, min(index - visible // 2, max(0, len(groups) - visible)))
|
|
|
|
for row, group in enumerate(groups[start:start + visible]):
|
|
position = start + row
|
|
attr = curses.color_pair(C_SEL) if position == index else 0
|
|
keep = "%s / %s" % (group.keep_count or "unbegr.",
|
|
format_duration(group.keep_time))
|
|
line = ("%s%s%s%s%s"
|
|
% (truncate(group.name, 17).ljust(18),
|
|
("ja" if group.enabled else "nein").ljust(7),
|
|
truncate(describe(group), 29).ljust(30),
|
|
truncate(keep, 19).ljust(20),
|
|
truncate(self._selection_text(group), max(4, width - 80))))
|
|
_fill_row(win, 5 + row, 2, line, width, attr)
|
|
|
|
problems = self.config.validate()
|
|
if problems:
|
|
_fill(win, height - 2, " %d Hinweis(e) - 'p' zeigt Details"
|
|
% len(problems), curses.color_pair(C_WARN))
|
|
|
|
win.refresh()
|
|
key = _read_key(win)
|
|
|
|
if key in (curses.KEY_UP, "k"):
|
|
index -= 1
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
index += 1
|
|
elif key == curses.KEY_HOME:
|
|
index = 0
|
|
elif key == curses.KEY_END:
|
|
index = len(groups) - 1
|
|
elif _is_enter(key) and groups:
|
|
self.screen_group(win, groups[index])
|
|
elif key == "n":
|
|
self._new_group(win)
|
|
index = len(self.config.groups) - 1
|
|
elif key == "c" and groups:
|
|
name = self._prompt(win, "Name der Kopie", groups[index].name + "-kopie")
|
|
if name:
|
|
self.config.groups.append(clone_group(groups[index], name))
|
|
self.dirty = True
|
|
elif key == "d" and groups:
|
|
if self._confirm(win, "Gruppe '%s' wirklich loeschen?" % groups[index].name):
|
|
del self.config.groups[index]
|
|
self.dirty = True
|
|
elif key == " " and groups:
|
|
groups[index].enabled = not groups[index].enabled
|
|
self.dirty = True
|
|
elif key == "g":
|
|
self.screen_globals(win)
|
|
elif key == "v":
|
|
self.screen_overview(win)
|
|
elif key == "p":
|
|
self._show_problems(win)
|
|
elif key == "s":
|
|
self._save(win)
|
|
elif key in ("r", "R"):
|
|
self._reload_service(win)
|
|
elif key in ("q", "Q") or _is_escape(key):
|
|
if self.dirty and not self._confirm(win, "Ungespeicherte Aenderungen verwerfen?"):
|
|
continue
|
|
return
|
|
|
|
def _selection_text(self, group):
|
|
parts = []
|
|
if group.all:
|
|
parts.append("alle VMs")
|
|
if group.vmids:
|
|
parts.append("IDs: " + format_vmid_list(group.vmids))
|
|
if group.names:
|
|
parts.append("Namen: " + ", ".join(group.names))
|
|
if group.tags:
|
|
parts.append("Tags: " + ", ".join(group.tags))
|
|
if group.pools:
|
|
parts.append("Pools: " + ", ".join(group.pools))
|
|
return "; ".join(parts) or "(nichts ausgewaehlt)"
|
|
|
|
def _new_group(self, win):
|
|
name = self._prompt(win, "Name der neuen Gruppe", "")
|
|
if not name:
|
|
return
|
|
if self.config.group(name):
|
|
self._message(win, "Eine Gruppe mit diesem Namen gibt es bereits.", error=True)
|
|
return
|
|
group = Group(name=name, interval=3600, keep_count=24, keep_time=2 * 86400)
|
|
self.config.groups.append(group)
|
|
self.dirty = True
|
|
self.screen_group(win, group)
|
|
|
|
def _show_problems(self, win):
|
|
problems = self.config.validate()
|
|
height, width = self._frame(win, "Hinweise zur Konfiguration", self.path,
|
|
"beliebige Taste = zurueck")
|
|
if not problems:
|
|
_put(win, 4, 4, "Alles in Ordnung.", curses.color_pair(C_OK))
|
|
for row, problem in enumerate(problems[: height - 6]):
|
|
_put(win, 4 + row, 4, truncate("- " + problem, width - 6))
|
|
win.refresh()
|
|
_read_key(win)
|
|
|
|
def _save(self, win, offer_reload=True):
|
|
problems = self.config.validate()
|
|
if problems:
|
|
if not self._confirm(win, "%d Hinweis(e) - trotzdem speichern?" % len(problems)):
|
|
return
|
|
try:
|
|
target = save_config(self.config, self.path)
|
|
self.dirty = False
|
|
except OSError as exc:
|
|
self._message(win, "Speichern fehlgeschlagen: %s" % exc, error=True)
|
|
return
|
|
|
|
# Direkt anbieten, die Aenderung auch scharf zu schalten.
|
|
loaded, state = service_state()
|
|
if offer_reload and loaded and state == "active":
|
|
if self._confirm(win, "Gespeichert. Dienst jetzt neu laden?"):
|
|
self._reload_service(win)
|
|
return
|
|
self._message(win, "Gespeichert: %s (wirksam nach 'r' oder "
|
|
"'systemctl reload pvesnap')" % target)
|
|
else:
|
|
self._message(win, "Gespeichert: %s" % target)
|
|
|
|
# -- Dienst steuern ---------------------------------------------------
|
|
|
|
def _service_text(self, refresh=False):
|
|
if self._service_cache is None or refresh:
|
|
self._service_cache = service_text()
|
|
return self._service_cache
|
|
|
|
def _reload_service(self, win):
|
|
"""Taste 'r': Konfiguration speichern und den Dienst neu laden."""
|
|
if self.dirty:
|
|
if self._confirm(win, "Es gibt ungespeicherte Aenderungen - erst speichern?"):
|
|
self._save(win, offer_reload=False)
|
|
if self.dirty:
|
|
return # Speichern abgebrochen oder fehlgeschlagen
|
|
|
|
loaded, state = service_state()
|
|
if loaded is None:
|
|
self._message(win, "systemd nicht erreichbar: %s" % state, error=True)
|
|
return
|
|
if not loaded:
|
|
self._message(win, "Der Dienst %s ist nicht installiert - bitte install.sh "
|
|
"ausfuehren." % SERVICE, error=True)
|
|
return
|
|
if os.geteuid() != 0:
|
|
self._message(win, "Dazu werden root-Rechte benoetigt - bitte 'sudo pvesnap "
|
|
"config' verwenden.", error=True)
|
|
return
|
|
|
|
if state == "active":
|
|
code, output = self._run_service_action(win, "reload", "neu geladen")
|
|
if code != 0:
|
|
first = (output.splitlines() or ["Fehler %d" % code])[0]
|
|
if self._confirm(win, "Fehlgeschlagen: %s - stattdessen neu starten?"
|
|
% truncate(first, 50)):
|
|
self._run_service_action(win, "restart", "neu gestartet")
|
|
else:
|
|
if self._confirm(win, "Der Dienst laeuft nicht - jetzt starten?"):
|
|
self._run_service_action(win, "start", "gestartet")
|
|
|
|
def _run_service_action(self, win, action, past_tense):
|
|
height, _ = win.getmaxyx()
|
|
_fill(win, height - 2, " systemctl %s %s ..." % (action, SERVICE),
|
|
curses.color_pair(C_SEL))
|
|
win.refresh()
|
|
code, output = _systemctl(action, SERVICE)
|
|
self._service_cache = None
|
|
if code == 0:
|
|
self._message(win, "Dienst %s - %s" % (past_tense, self._service_text(refresh=True)))
|
|
else:
|
|
first = (output.splitlines() or ["Rueckgabewert %d" % code])[0]
|
|
self._message(win, "systemctl %s fehlgeschlagen: %s" % (action, first), error=True)
|
|
return code, output
|
|
|
|
# -- Bildschirm: eine Gruppe -----------------------------------------
|
|
|
|
def screen_group(self, win, group):
|
|
index = 0
|
|
top = 0
|
|
while True:
|
|
fields = self._group_fields(group)
|
|
index = max(0, min(index, len(fields) - 1))
|
|
while fields[index][0] == "-" and index < len(fields) - 1:
|
|
index += 1
|
|
|
|
height, width = self._frame(
|
|
win, "Gruppe: %s" % group.name,
|
|
"Kurzname im Snapshot: %s-%s-JJJJMMTT-HHMMSS"
|
|
% (self.config.globals.prefix,
|
|
effective_slug(self.config.globals.prefix, group.slug)),
|
|
"Enter Aendern | Leer Umschalten | v VM-Auswahl | q zurueck")
|
|
|
|
visible = height - 5
|
|
if index < top:
|
|
top = index
|
|
if index >= top + visible:
|
|
top = index - visible + 1
|
|
|
|
for row in range(visible):
|
|
position = top + row
|
|
if position >= len(fields):
|
|
break
|
|
kind, label, value, _handler = fields[position]
|
|
if kind == "-":
|
|
_put(win, 3 + row, 2, "-- %s " % label + "-" * max(0, width - len(label) - 8),
|
|
curses.color_pair(C_DIM) | curses.A_BOLD)
|
|
continue
|
|
attr = curses.color_pair(C_SEL) if position == index else 0
|
|
_fill_row(win, 3 + row, 2,
|
|
truncate(label, 30).ljust(32) + truncate(value, max(4, width - 40)),
|
|
width, attr)
|
|
|
|
win.refresh()
|
|
key = _read_key(win)
|
|
|
|
if key in (curses.KEY_UP, "k"):
|
|
index = self._step(fields, index, -1)
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
index = self._step(fields, index, +1)
|
|
elif key == curses.KEY_NPAGE:
|
|
index = self._step(fields, min(index + 10, len(fields) - 1), +1, land=True)
|
|
elif key == curses.KEY_PPAGE:
|
|
index = self._step(fields, max(index - 10, 0), -1, land=True)
|
|
elif _is_enter(key) or key == " ":
|
|
handler = fields[index][3]
|
|
if handler:
|
|
handler(win)
|
|
self.dirty = True
|
|
elif key == "v":
|
|
self._pick_vms(win, group)
|
|
elif key in ("q", "Q") or _is_escape(key):
|
|
return
|
|
|
|
@staticmethod
|
|
def _step(fields, index, direction, land=False):
|
|
position = index if land else index + direction
|
|
while 0 <= position < len(fields) and fields[position][0] == "-":
|
|
position += direction
|
|
if position < 0 or position >= len(fields):
|
|
return index
|
|
return position
|
|
|
|
def _group_fields(self, group):
|
|
"""Liste von (art, beschriftung, angezeigter_wert, handler)."""
|
|
fields = []
|
|
|
|
def text_field(label, getter, setter, hint=""):
|
|
def handler(win):
|
|
value = self._prompt(win, hint or label, getter())
|
|
if value is not None:
|
|
setter(value)
|
|
fields.append(("f", label, getter(), handler))
|
|
|
|
def bool_field(label, getter, setter):
|
|
def handler(_win):
|
|
setter(not getter())
|
|
fields.append(("f", label, "ja" if getter() else "nein", handler))
|
|
|
|
def int_field(label, getter, setter, zero_text=""):
|
|
def handler(win):
|
|
raw = self._prompt(win, label, str(getter()))
|
|
if raw is None:
|
|
return
|
|
try:
|
|
setter(max(0, int(raw.strip() or 0)))
|
|
except ValueError:
|
|
self._message(win, "Bitte eine ganze Zahl eingeben.", error=True)
|
|
value = getter()
|
|
fields.append(("f", label, zero_text if (value == 0 and zero_text) else str(value),
|
|
handler))
|
|
|
|
def duration_field(label, getter, setter):
|
|
def handler(win):
|
|
raw = self._prompt(win, label + " (z.B. 30m, 6h, 7d, 0=unbegrenzt)",
|
|
format_duration(getter(), zero="0"))
|
|
if raw is None:
|
|
return
|
|
try:
|
|
setter(parse_duration(raw, default_unit="m"))
|
|
except ValueError as exc:
|
|
self._message(win, str(exc), error=True)
|
|
fields.append(("f", label, format_duration(getter()), handler))
|
|
|
|
def list_field(label, getter, setter, hint=""):
|
|
def handler(win):
|
|
raw = self._prompt(win, hint or label, ", ".join(getter()))
|
|
if raw is not None:
|
|
setter(parse_list(raw))
|
|
fields.append(("f", label, ", ".join(getter()) or "-", handler))
|
|
|
|
# --- Allgemein ---
|
|
fields.append(("-", "Allgemein", "", None))
|
|
|
|
def set_name(value):
|
|
if value:
|
|
group.name = value
|
|
text_field("Name", lambda: group.name, set_name)
|
|
bool_field("Aktiv", lambda: group.enabled,
|
|
lambda v: setattr(group, "enabled", v))
|
|
|
|
# --- Zeitplan ---
|
|
fields.append(("-", "Zeitplan", "", None))
|
|
|
|
current_kind = group.schedule or "interval"
|
|
|
|
def change_kind(win):
|
|
values = [c[0] for c in SCHEDULE_CHOICES]
|
|
chosen = self._choose(win, "Zeitplan", SCHEDULE_CHOICES,
|
|
values.index(current_kind))
|
|
if chosen is None:
|
|
return
|
|
if chosen == "interval":
|
|
group.schedule = ""
|
|
group.interval = group.interval or 3600
|
|
else:
|
|
group.schedule = chosen
|
|
group.interval = 0
|
|
|
|
fields.append(("f", "Art des Zeitplans",
|
|
dict(SCHEDULE_CHOICES)[current_kind], change_kind))
|
|
|
|
if current_kind == "interval":
|
|
duration_field("Intervall", lambda: group.interval,
|
|
lambda v: setattr(group, "interval", max(60, v)))
|
|
bool_field("An der Uhr ausrichten", lambda: group.align,
|
|
lambda v: setattr(group, "align", v))
|
|
elif current_kind == "hourly":
|
|
int_field("Minute (0-59)", lambda: group.minute,
|
|
lambda v: setattr(group, "minute", min(59, v)))
|
|
else:
|
|
def set_time(value):
|
|
try:
|
|
group.at = "%02d:%02d" % parse_time_of_day(value)
|
|
except ValueError:
|
|
pass
|
|
text_field("Uhrzeit (HH:MM)", lambda: group.at, set_time)
|
|
if current_kind == "weekly":
|
|
def change_day(win):
|
|
entries = [(i, name) for i, name in enumerate(
|
|
["Montag", "Dienstag", "Mittwoch", "Donnerstag",
|
|
"Freitag", "Samstag", "Sonntag"])]
|
|
chosen = self._choose(win, "Wochentag", entries, group.day_of_week)
|
|
if chosen is not None:
|
|
group.day_of_week = chosen
|
|
fields.append(("f", "Wochentag",
|
|
WEEKDAY_NAMES[group.day_of_week], change_day))
|
|
if current_kind in ("monthly", "yearly"):
|
|
int_field("Tag im Monat (1-31)", lambda: group.day_of_month,
|
|
lambda v: setattr(group, "day_of_month", max(1, min(31, v))))
|
|
if current_kind == "yearly":
|
|
int_field("Monat (1-12)", lambda: group.month,
|
|
lambda v: setattr(group, "month", max(1, min(12, v))))
|
|
|
|
fields.append(("f", "Naechster Termin (Vorschau)", _preview_next(group), None))
|
|
|
|
# --- Vorhaltezeit ---
|
|
fields.append(("-", "Vorhaltezeit", "", None))
|
|
int_field("Anzahl behalten (je VM)", lambda: group.keep_count,
|
|
lambda v: setattr(group, "keep_count", v), zero_text="unbegrenzt")
|
|
duration_field("Maximales Alter", lambda: group.keep_time,
|
|
lambda v: setattr(group, "keep_time", v))
|
|
int_field("Mindestens behalten", lambda: group.keep_min,
|
|
lambda v: setattr(group, "keep_min", v), zero_text="0 (keine Untergrenze)")
|
|
|
|
# --- Auswahl ---
|
|
fields.append(("-", "Welche VMs?", "", None))
|
|
bool_field("Alle VMs/Container", lambda: group.all,
|
|
lambda v: setattr(group, "all", v))
|
|
|
|
fields.append(("f", "VMs aus Liste waehlen ...",
|
|
format_vmid_list(group.vmids) or "-",
|
|
lambda win: self._pick_vms(win, group)))
|
|
list_field("Namensmuster (z.B. web-*)", lambda: group.names,
|
|
lambda v: setattr(group, "names", v))
|
|
list_field("Tags", lambda: group.tags,
|
|
lambda v: setattr(group, "tags", [t.lower() for t in v]))
|
|
list_field("Pools", lambda: group.pools,
|
|
lambda v: setattr(group, "pools", v))
|
|
|
|
def change_types(win):
|
|
labels = [(index, text) for index, (_v, text) in enumerate(TYPE_CHOICES)]
|
|
current = 0
|
|
for index, (value, _text) in enumerate(TYPE_CHOICES):
|
|
if sorted(value) == sorted(group.types):
|
|
current = index
|
|
chosen = self._choose(win, "Welche Gasttypen?", labels, current)
|
|
if chosen is not None:
|
|
group.types = list(TYPE_CHOICES[chosen][0])
|
|
current_type_text = "alle (VMs und Container)"
|
|
for value, text in TYPE_CHOICES:
|
|
if sorted(value) == sorted(group.types):
|
|
current_type_text = text
|
|
fields.append(("f", "Gasttypen", current_type_text, change_types))
|
|
|
|
def set_exclude_vmids(win):
|
|
raw = self._prompt(win, "Ausgeschlossene VMIDs (z.B. 900,905-910)",
|
|
format_vmid_list(group.exclude_vmids))
|
|
if raw is None:
|
|
return
|
|
try:
|
|
group.exclude_vmids = parse_vmid_list(raw)
|
|
except ValueError as exc:
|
|
self._message(win, str(exc), error=True)
|
|
fields.append(("f", "Ausgeschlossene VMIDs",
|
|
format_vmid_list(group.exclude_vmids) or "-", set_exclude_vmids))
|
|
list_field("Ausgeschlossene Namensmuster", lambda: group.exclude_names,
|
|
lambda v: setattr(group, "exclude_names", v))
|
|
list_field("Ausgeschlossene Tags", lambda: group.exclude_tags,
|
|
lambda v: setattr(group, "exclude_tags", [t.lower() for t in v]))
|
|
|
|
matched = self._match_count(group)
|
|
fields.append(("f", "Trifft aktuell zu auf", matched, None))
|
|
|
|
# --- Optionen ---
|
|
fields.append(("-", "Snapshot-Optionen", "", None))
|
|
bool_field("RAM mitsichern (nur laufende VMs)", lambda: group.vmstate,
|
|
lambda v: setattr(group, "vmstate", v))
|
|
bool_field("Gestoppte ueberspringen", lambda: group.skip_stopped,
|
|
lambda v: setattr(group, "skip_stopped", v))
|
|
text_field("Beschreibung (Vorlage)",
|
|
lambda: group.description or self.config.globals.description,
|
|
lambda v: setattr(group, "description", v),
|
|
hint="Beschreibung, Platzhalter: {group} {vmid} {name} {datetime} "
|
|
"{keep_time} {keep_count}")
|
|
return fields
|
|
|
|
def _match_count(self, group):
|
|
guests = self.guests()
|
|
if not guests:
|
|
return self._inventory_error or "unbekannt"
|
|
from .engine import select_guests
|
|
selected = select_guests(group, guests)
|
|
if not selected:
|
|
return "0 Gaeste"
|
|
preview = ", ".join("%d" % g.vmid for g in selected[:8])
|
|
return "%d Gast/Gaeste (%s%s)" % (len(selected), preview,
|
|
" ..." if len(selected) > 8 else "")
|
|
|
|
# -- Bildschirm: VM-Auswahl ------------------------------------------
|
|
|
|
def _pick_vms(self, win, group):
|
|
guests = self.guests()
|
|
if not guests:
|
|
self._message(win, self._inventory_error or "Keine VMs gefunden.", error=True)
|
|
return
|
|
|
|
chosen = set(group.vmids)
|
|
index, top, filter_text = 0, 0, ""
|
|
|
|
while True:
|
|
shown = [g for g in guests if self._matches_filter(g, filter_text)]
|
|
index = max(0, min(index, max(0, len(shown) - 1)))
|
|
height, width = self._frame(
|
|
win, "VMs fuer Gruppe '%s'" % group.name,
|
|
"%d ausgewaehlt%s" % (len(chosen),
|
|
(" Filter: " + filter_text) if filter_text else ""),
|
|
"Leer Auswaehlen | a Alle | n Keine | i Umkehren | / Filter | "
|
|
"Enter Uebernehmen | q Abbruch")
|
|
|
|
_put(win, 3, 2, " VMID Typ Name".ljust(46) + "Node".ljust(14)
|
|
+ "Status".ljust(10) + "Tags", curses.A_BOLD)
|
|
visible = max(1, height - 6)
|
|
if index < top:
|
|
top = index
|
|
if index >= top + visible:
|
|
top = index - visible + 1
|
|
|
|
for row in range(visible):
|
|
position = top + row
|
|
if position >= len(shown):
|
|
break
|
|
guest = shown[position]
|
|
attr = curses.color_pair(C_SEL) if position == index else 0
|
|
mark = "[x]" if guest.vmid in chosen else "[ ]"
|
|
line = ("%s %s %s %s%s%s%s"
|
|
% (mark,
|
|
str(guest.vmid).rjust(6),
|
|
("LXC" if guest.type == "lxc" else "VM ").ljust(4),
|
|
truncate(guest.name or "-", 26).ljust(28),
|
|
truncate(guest.node, 12).ljust(14),
|
|
truncate(guest.status, 9).ljust(10),
|
|
truncate(",".join(guest.tags) or "-", 20)))
|
|
_fill_row(win, 4 + row, 2, line, width, attr)
|
|
|
|
win.refresh()
|
|
key = _read_key(win)
|
|
|
|
if key in (curses.KEY_UP, "k"):
|
|
index -= 1
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
index += 1
|
|
elif key == curses.KEY_NPAGE:
|
|
index += visible
|
|
elif key == curses.KEY_PPAGE:
|
|
index -= visible
|
|
elif key == curses.KEY_HOME:
|
|
index = 0
|
|
elif key == curses.KEY_END:
|
|
index = len(shown) - 1
|
|
elif key == " " and shown:
|
|
vmid = shown[index].vmid
|
|
if vmid in chosen:
|
|
chosen.discard(vmid)
|
|
else:
|
|
chosen.add(vmid)
|
|
index += 1
|
|
elif key == "a":
|
|
chosen.update(g.vmid for g in shown)
|
|
elif key == "n":
|
|
chosen.difference_update(g.vmid for g in shown)
|
|
elif key == "i":
|
|
for guest in shown:
|
|
if guest.vmid in chosen:
|
|
chosen.discard(guest.vmid)
|
|
else:
|
|
chosen.add(guest.vmid)
|
|
elif key == "/":
|
|
entered = self._prompt(win, "Filter (Name, VMID oder Tag)", filter_text)
|
|
filter_text = entered or ""
|
|
index = top = 0
|
|
elif key == "r":
|
|
self.guests(refresh=True)
|
|
guests = self._guests
|
|
elif _is_enter(key):
|
|
group.vmids = sorted(chosen)
|
|
self.dirty = True
|
|
return
|
|
elif key in ("q", "Q") or _is_escape(key):
|
|
return
|
|
index = max(0, min(index, max(0, len(shown) - 1)))
|
|
|
|
@staticmethod
|
|
def _matches_filter(guest, text):
|
|
if not text:
|
|
return True
|
|
needle = text.lower()
|
|
return (needle in (guest.name or "").lower()
|
|
or needle in str(guest.vmid)
|
|
or needle in " ".join(guest.tags)
|
|
or needle in (guest.pool or "").lower()
|
|
or needle in (guest.node or "").lower())
|
|
|
|
# -- Bildschirm: globale Einstellungen -------------------------------
|
|
|
|
def screen_globals(self, win):
|
|
globals_ = self.config.globals
|
|
index = 0
|
|
while True:
|
|
fields = [
|
|
("Praefix im Snapshot-Namen", globals_.prefix, "text", "prefix"),
|
|
("Pruefintervall des Dienstes", format_duration(globals_.check_interval),
|
|
"duration", "check_interval"),
|
|
("Zustandsdatei", globals_.state_file, "text", "state_file"),
|
|
("Protokollstufe", globals_.log_level, "level", "log_level"),
|
|
("Zusaetzliche Logdatei", globals_.log_file or "-", "text", "log_file"),
|
|
("Zeitlimit je Snapshot-Task", format_duration(globals_.task_timeout),
|
|
"duration", "task_timeout"),
|
|
("Beim Start sofort ausfuehren", "ja" if globals_.run_on_start else "nein",
|
|
"bool", "run_on_start"),
|
|
("Testlauf (nichts wirklich tun)", "ja" if globals_.dry_run else "nein",
|
|
"bool", "dry_run"),
|
|
("Standard-Beschreibung", globals_.description, "text", "description"),
|
|
]
|
|
index = max(0, min(index, len(fields) - 1))
|
|
height, width = self._frame(win, "Globale Einstellungen", self.path,
|
|
"Enter Aendern | q zurueck")
|
|
for row, (label, value, _kind, _key) in enumerate(fields):
|
|
attr = curses.color_pair(C_SEL) if row == index else 0
|
|
_fill_row(win, 4 + row, 2,
|
|
truncate(label, 32).ljust(34) + truncate(str(value), max(4, width - 42)),
|
|
width, attr)
|
|
_put(win, 6 + len(fields), 2,
|
|
"Platzhalter der Beschreibung: {group} {vmid} {name} {node} {type} "
|
|
"{datetime} {date} {time} {keep_time} {keep_count} {schedule}",
|
|
curses.color_pair(C_DIM))
|
|
win.refresh()
|
|
|
|
key = _read_key(win)
|
|
if key in (curses.KEY_UP, "k"):
|
|
index -= 1
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
index += 1
|
|
elif _is_enter(key) or key == " ":
|
|
label, _value, kind, attribute = fields[index]
|
|
self._edit_global(win, label, kind, attribute)
|
|
elif key in ("q", "Q") or _is_escape(key):
|
|
return
|
|
|
|
def _edit_global(self, win, label, kind, attribute):
|
|
globals_ = self.config.globals
|
|
if kind == "bool":
|
|
setattr(globals_, attribute, not getattr(globals_, attribute))
|
|
self.dirty = True
|
|
return
|
|
if kind == "level":
|
|
entries = [(lvl, lvl) for lvl in ("DEBUG", "INFO", "WARNING", "ERROR")]
|
|
chosen = self._choose(win, "Protokollstufe", entries)
|
|
if chosen:
|
|
globals_.log_level = chosen
|
|
self.dirty = True
|
|
return
|
|
if kind == "duration":
|
|
raw = self._prompt(win, label, format_duration(getattr(globals_, attribute),
|
|
zero="0"))
|
|
if raw is None:
|
|
return
|
|
try:
|
|
setattr(globals_, attribute, parse_duration(raw, default_unit="s"))
|
|
self.dirty = True
|
|
except ValueError as exc:
|
|
self._message(win, str(exc), error=True)
|
|
return
|
|
|
|
raw = self._prompt(win, label, getattr(globals_, attribute))
|
|
if raw is None:
|
|
return
|
|
setattr(globals_, attribute, raw)
|
|
self.dirty = True
|
|
|
|
# -- Bildschirm: Uebersicht ------------------------------------------
|
|
|
|
def screen_overview(self, win):
|
|
from .engine import select_guests
|
|
guests = self.guests()
|
|
top = 0
|
|
while True:
|
|
height, width = self._frame(win, "Uebersicht: welche VM in welcher Gruppe?",
|
|
self._inventory_error or "%d Gaeste" % len(guests),
|
|
"Pfeiltasten Blaettern | r Neu laden | q zurueck")
|
|
_put(win, 3, 2, "VMID".ljust(8) + "Name".ljust(26) + "Status".ljust(10) + "Gruppen",
|
|
curses.A_BOLD)
|
|
rows = []
|
|
for guest in guests:
|
|
names = [g.name for g in self.config.groups if guest in select_guests(g, [guest])]
|
|
rows.append((guest, names))
|
|
|
|
visible = max(1, height - 6)
|
|
top = max(0, min(top, max(0, len(rows) - visible)))
|
|
for row in range(visible):
|
|
position = top + row
|
|
if position >= len(rows):
|
|
break
|
|
guest, names = rows[position]
|
|
attr = 0 if names else curses.color_pair(C_DIM) | curses.A_DIM
|
|
_put(win, 4 + row, 2,
|
|
truncate(str(guest.vmid), 7).ljust(8)
|
|
+ truncate(guest.name or "-", 24).ljust(26)
|
|
+ truncate(guest.status, 9).ljust(10)
|
|
+ truncate(", ".join(names) or "(keine Gruppe)", max(4, width - 48)),
|
|
attr)
|
|
win.refresh()
|
|
|
|
key = _read_key(win)
|
|
if key == curses.KEY_DOWN:
|
|
top += 1
|
|
elif key == curses.KEY_UP:
|
|
top -= 1
|
|
elif key == curses.KEY_NPAGE:
|
|
top += visible
|
|
elif key == curses.KEY_PPAGE:
|
|
top -= visible
|
|
elif key == "r":
|
|
guests = self.guests(refresh=True)
|
|
elif key in ("q", "Q") or _is_escape(key):
|
|
return
|
|
top = max(0, top)
|
|
|
|
|
|
def _preview_next(group):
|
|
"""Zeigt den naechsten Termin, damit man die Einstellung sofort pruefen kann."""
|
|
from datetime import datetime
|
|
|
|
from .schedule import next_due
|
|
try:
|
|
moment = next_due(group, None, datetime.now())
|
|
except ValueError:
|
|
return "-"
|
|
tage = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
|
|
return "%s %s (%s)" % (tage[moment.weekday()],
|
|
moment.strftime("%d.%m.%Y %H:%M"), describe(group))
|
|
|
|
|
|
def _fill_row(win, y, x, text, width, attr):
|
|
_put(win, y, x, str(text).ljust(max(0, width - x - 1))[: max(0, width - x - 1)], attr)
|
|
|
|
|
|
def run_editor(path):
|
|
editor = Editor(path)
|
|
try:
|
|
editor.load()
|
|
except ConfigError as exc:
|
|
print(str(exc))
|
|
return 2
|
|
curses.wrapper(editor.run)
|
|
if editor.dirty:
|
|
print("Achtung: Aenderungen wurden NICHT gespeichert.")
|
|
return 1
|
|
return 0
|