"""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"), ("Wiederholungen bei belegter Sperre", str(globals_.retries), "int", "retries"), ("Wartezeit vor Wiederholung", format_duration(globals_.retry_delay), "duration", "retry_delay"), ("Pause zwischen zwei VMs", format_duration(globals_.pause_between, zero="keine"), "duration", "pause_between"), ("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 == "int": raw = self._prompt(win, label, str(getattr(globals_, attribute))) if raw is None: return try: setattr(globals_, attribute, max(0, int(raw.strip() or 0))) self.dirty = True except ValueError: self._message(win, "Bitte eine ganze Zahl eingeben.", error=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