proxmox-snapshot-service/handbuch/werkstatt/run.py

#!/usr/bin/env python3
"""Startet eine Oberflaeche von pvesnap gegen die Demo-Kulisse.

Aufruf:  run.py <szene> [argumente]

Der eigentliche Trick steckt in _patch(): alle Pfade, die sonst auf einen
echten Proxmox-Host zeigen, werden auf demo/ umgebogen. Der Programmcode
selbst wird nicht angefasst - die Bildschirmfotos zeigen also wirklich das,
was auf dem Host auch zu sehen ist.
"""

from __future__ import annotations

import curses
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
DEMO = os.path.join(HERE, "demo")
sys.path.insert(0, os.environ.get("PVESNAP_SRC", ""))

# Wo die Kulisse haengt, wenn stage.sh sie eingehaengt hat. Ohne Namensraum
# (beim schnellen Nachsehen) wird auf die Verzeichnisse selbst zurueckgefallen.
SNAP_MNT = "/run/pvesnap/mnt/101-auto-taeglich-20260809-023000"
TRANSFER_MNT = "/run/pvesnap/transfer/dumps"
LOKAL = "/root"


def _oder(pfad, ersatz):
    return pfad if os.path.isdir(pfad) and os.listdir(pfad) else ersatz


def _patch():
    from pvesnap import preflight, recovery, transfer, tui

    recovery.STATE_DIR = os.path.join(DEMO, "lib")
    recovery.REGISTRY = os.path.join(DEMO, "lib", "recovery.json")
    recovery.CONF_ROOT = os.path.join(DEMO, "pve", "nodes")

    transfer.TRANSFER_DIR = os.path.join(DEMO, "lib", "transfer")
    transfer.INDEX = os.path.join(DEMO, "lib", "transfer", "index.json")
    transfer.MOUNT_ROOT = "/run/pvesnap/transfer"
    transfer.CONF_ROOT = recovery.CONF_ROOT

    # Loop-Geraete und Einhaengepunkte gibt es hier nicht - also erzaehlen wir
    # sie. "dumps" haengt in der Maschine, "austausch" gerade am Host.
    abbild = lambda name: os.path.join(transfer.TRANSFER_DIR, name + ".img")
    transfer._loop_devices = lambda: {abbild("dumps"): "/dev/loop3",
                                      abbild("austausch"): "/dev/loop5"}
    transfer._mounts = lambda: {"/dev/loop5p1": os.path.join(transfer.MOUNT_ROOT,
                                                             "austausch")}

    with open(os.path.join(DEMO, "pve", ".members"), encoding="utf-8") as handle:
        mitglieder = json.load(handle)
    recovery._ADDRESSES.update({n: e["ip"] for n, e
                                in mitglieder["nodelist"].items()})

    preflight.check_environment = lambda require_pve=True: []
    tui.service_state = lambda: (True, "active")
    tui.service_text = lambda: "Dienst: laeuft"


class Kulissensession:
    """Genau so viel Sitzung, wie der Explorer fuer die Laufwerkswahl braucht.

    Eine echte Session wuerde einen Snapshot einbinden - dafuer braucht es
    Ceph. Der Explorer fasst sie nur an zwei Stellen an: um die Dateisysteme
    aufzuzaehlen und um zu wissen, welches Fenster der Snapshot ist.
    """
    mounts = {"root": 1, "boot": 1}

    @staticmethod
    def mount_entries():
        return [(SNAP_MNT, "/  (ext4, 400G)  - wahrscheinlich das System"),
                (SNAP_MNT + "-boot", "/boot  (ext4, 1G)")]


def _explorer_panes(links_pfad, links_titel, untertitel, links_schreibbar=False,
                    session=None):
    from pvesnap.explorer import Explorer, Pane
    links = Pane(path=links_pfad, root=links_pfad,
                 readonly=not links_schreibbar, title=links_titel)
    rechts = Pane(path=_oder(LOKAL, os.path.join(DEMO, "lokal")), root="/",
                  readonly=False, title="Lokaler Rechner")
    return Explorer(links, rechts, session=session, subtitle=untertitel)


# ---------------------------------------------------------------------------
# Szenen
# ---------------------------------------------------------------------------

def szene_config(stdscr):
    from pvesnap.tui import Editor
    pfad = "/etc/pvesnap.conf"
    editor = Editor(pfad if os.path.exists(pfad)
                    else os.path.join(DEMO, "pvesnap.conf"))
    editor.load()
    editor.run(stdscr)


def szene_recovery(stdscr):
    from pvesnap.proxmox import Proxmox
    from pvesnap.recovery_ui import _overview
    _overview(stdscr, Proxmox())


def szene_transfer(stdscr):
    from pvesnap import transfer_ui
    from pvesnap.recovery_ui import _browse_transfer
    transfer_ui.screen(stdscr, browse=_browse_transfer)


def szene_explorer(stdscr):
    _explorer_panes(_oder(SNAP_MNT, os.path.join(DEMO, "snapshot")),
                    "VM 101 (db01)  @  auto-taeglich-20260809-023000",
                    "VM 101 (db01)  -  Snapshot auto-taeglich-20260809-023000",
                    session=Kulissensession()).run(stdscr)


def szene_explorer_transfer(stdscr):
    _explorer_panes(_oder(TRANSFER_MNT, os.path.join(DEMO, "transfer-dumps")),
                    "Transfer: dumps  (DUMPS)",
                    "Transfer-Laufwerk dumps - beide Seiten beschreibbar",
                    links_schreibbar=True).run(stdscr)


def szene_gastauswahl(stdscr):
    from pvesnap.explorer import _pick_guest
    from pvesnap.proxmox import Proxmox
    _pick_guest(stdscr, Proxmox().inventory())


def szene_snapshotauswahl(stdscr):
    from pvesnap.explorer import _pick_snapshot
    from pvesnap.proxmox import Proxmox
    from pvesnap.snapfs import list_snapshots
    proxmox = Proxmox()
    gast = next(g for g in proxmox.inventory() if g.vmid == 101)
    _pick_snapshot(stdscr, [s for s in list_snapshots(proxmox, gast) if s.complete])


def szene_neu(stdscr):
    """Die Optionsmaske des Assistenten - ohne etwas anzulegen."""
    from pvesnap.proxmox import Proxmox
    from pvesnap.recovery_ui import _options
    from pvesnap.recovery import Spec
    proxmox = Proxmox()
    gast = next(g for g in proxmox.inventory() if g.vmid == 101)
    spec = Spec(mode="live")
    spec.snapshot = "auto-stuendlich-20260809-100000"
    spec.newid = 9103
    spec.node = "pve1"
    spec.transfers = ["werkzeuge"]
    _options(stdscr, gast, spec)


def _vorhaben(stdscr, vmid, snapshot, **werte):
    from pvesnap.curses_util import init_colors
    from pvesnap.proxmox import Proxmox
    from pvesnap.recovery import Spec, plan
    from pvesnap.recovery_ui import _pages, summary_lines
    init_colors()
    proxmox = Proxmox()
    gast = next(g for g in proxmox.inventory() if g.vmid == vmid)
    spec = Spec(mode=werte.pop("mode", "live"))
    for schluessel, wert in werte.items():
        setattr(spec, schluessel, wert)
    _pages(stdscr, summary_lines(plan(proxmox, gast, snapshot, spec)),
           "Vorhaben pruefen", "j = anlegen | n = abbrechen")


def szene_vorhaben(stdscr):
    """Der Normalfall: abgeschottet hineinschauen, mit Arbeitsspeicher."""
    _vorhaben(stdscr, 101, "auto-stuendlich-20260809-100000",
              newid=9103, transfers=["werkzeuge"])


def szene_vorhaben_dongle(stdscr):
    """Die Windows-Maschine mit SPICE und USB - fuer das Schutzmodul."""
    _vorhaben(stdscr, 105, "auto-taeglich-20260809-023000",
              mode="live", newid=9104, transfers=["werkzeuge"], spice=True, usb=2)


def szene_laufwerkswahl(stdscr):
    from pvesnap import transfer_ui
    transfer_ui.choose(stdscr, preselected=["werkzeuge"])


SZENEN = {
    "config": szene_config,
    "recovery": szene_recovery,
    "transfer": szene_transfer,
    "explorer": szene_explorer,
    "explorer-transfer": szene_explorer_transfer,
    "gastauswahl": szene_gastauswahl,
    "snapshotauswahl": szene_snapshotauswahl,
    "neu": szene_neu,
    "vorhaben": szene_vorhaben,
    "vorhaben-dongle": szene_vorhaben_dongle,
    "laufwerkswahl": szene_laufwerkswahl,
}


def main():
    if len(sys.argv) < 2 or sys.argv[1] not in SZENEN:
        print("Szenen: %s" % ", ".join(sorted(SZENEN)), file=sys.stderr)
        return 1
    _patch()
    from pvesnap.curses_util import init_colors

    szene = SZENEN[sys.argv[1]]

    def start(stdscr):
        curses.curs_set(0)
        if hasattr(curses, "set_escdelay"):
            curses.set_escdelay(25)
        init_colors()
        stdscr.keypad(True)
        return szene(stdscr)

    try:
        curses.wrapper(start)
    except Exception:                     # beim Bauen will man den Grund sehen
        import traceback
        traceback.print_exc()
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())