duffyduckandClaude Opus 5
d9d9345d03
Zwei Luecken, die seit dem Umbau offenstanden.
1. Transfer-Laufwerke waren aus dem Explorer nicht erreichbar
F2 zeigte nur die Dateisysteme des Snapshots. Jetzt stehen die
Transfer-Laufwerke in derselben Liste - aus Sicht des Bedieners ist es
dieselbe Frage ("wo soll ich hinschauen?"), und der Weg ueber
pvesnap-recovery entfaellt.
Das gewaehlte Laufwerk landet im rechten Fenster, links bleibt der
Snapshot. Damit laesst sich direkt aus einem Snapshot auf das
Austauschmedium kopieren, das anschliessend in die wiederhergestellte
Maschine wandert - ohne Zwischenlager auf dem Host.
Ausgehaengt wird beim Verlassen, und zwar nur, was wir selbst eingehaengt
haben: ein Laufwerk, das schon vorher am Host hing, gehoert jemand
anderem. Das finally faengt auch Absturz und Strg-C ab - bliebe es
eingehaengt, gaelte es spaeter als belegt.
2. cleanup fand geschuetzte Snapshots ohne Klon nicht
Loescht jemand eine Wiederherstellung in der Proxmox-Oberflaeche, raeumt
Proxmox den Klon durchaus mit ab - aber nicht den Schutz seines
Quell-Snapshots (rbd snap protect, den setzt es beim Klonen selbst).
Zurueck bleibt ein geschuetzter Snapshot ohne Klon: die Vorhaltezeit
scheitert an ihm jede Nacht mit "snapshot is protected", und
find_orphans() findet prinzipiell nichts, weil es den Datentraeger, nach
dem es sucht, wirklich nicht mehr gibt. Dasselbe entsteht nach einem von
Hand ausgefuehrten rbd flatten.
find_stale_protections() sucht deshalb direkt nach der Ursache: jeder
geschuetzte Snapshot, an dem kein Klon haengt. Ein solcher Schutz hat
keinen Zweck - er existiert einzig dafuer, dass ein Klon seine Grundlage
behaelt. Geloest wird nur der Schutz, geloescht wird nichts. Vor jedem
Loesen wird ein zweites Mal nachgesehen, ob inzwischen doch ein Klon
daran haengt.
cleanup laeuft damit in zwei Durchgaengen und liegt neu auch in der
Oberflaeche auf Taste c - man sucht sonst lange nach einem Befehl, den
man nur im Notfall braucht.
Getestet gegen die Kulisse der Handbuch-Werkstatt: die rbd-Attrappe
kennt jetzt snap ls und children und enthaelt beide Zustaende
nebeneinander - ein geschuetzter Snapshot mit Klon (muss in Ruhe
gelassen werden) und einer ohne (muss gefunden werden).
Handbuch und README nachgezogen, zwei Bildschirmfotos dazu.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1088 lines
40 KiB
Python
1088 lines
40 KiB
Python
"""pvesnap-explorer - Dateien aus einem Snapshot holen.
|
|
|
|
Zwei Fenster wie im Midnight Commander: links der geoeffnete Snapshot
|
|
(schreibgeschuetzt), rechts der lokale Rechner. Markieren, F5 - fertig.
|
|
|
|
Drei Ebenen, aus denen man mit q jeweils eine zurueckkommt:
|
|
Gastauswahl -> Snapshot-Auswahl -> Dateibrowser. Beim Verlassen des
|
|
Browsers wird der Snapshot wieder ausgehaengt.
|
|
|
|
pvesnap-explorer Gast und Snapshot auswaehlen
|
|
pvesnap-explorer 802 Snapshot von VM 802 auswaehlen
|
|
pvesnap-explorer 802 <snapname> direkt oeffnen
|
|
pvesnap-explorer --path /mnt beliebiges Verzeichnis (ohne Proxmox)
|
|
pvesnap-explorer --cleanup Reste eines Absturzes aufraeumen
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import curses
|
|
import errno
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
from . import __version__
|
|
from . import transfer
|
|
from .curses_util import (C_DIM, C_DIR, C_FOOTER, C_HEADER, C_MARK, C_OK, C_SEL,
|
|
C_WARN, ask, choose, confirm, fill, init_colors, is_enter,
|
|
is_escape, keybar, message, prompt, put, read_key)
|
|
from .proxmox import Proxmox, ProxmoxError, pvesh_available
|
|
from .snapfs import (Session, SnapfsError, cleanup_leftovers, human_bytes,
|
|
list_snapshots)
|
|
from .util import truncate
|
|
|
|
log = logging.getLogger("pvesnap.explorer")
|
|
|
|
KEYS = [
|
|
("Tab", "Fenster", "Fenst."),
|
|
("Enter", "Oeffnen", "Oeffn."),
|
|
("Leer", "Markieren", "Mark."),
|
|
("F5", "Kopieren", "Kopie"),
|
|
("F3", "Ansehen", "Ansehen"),
|
|
("F7", "Neuer Ordner", "Ordner"),
|
|
("F2", "Laufwerk", "Laufw."),
|
|
("F10", "Zurueck", "Zurueck"),
|
|
]
|
|
|
|
COPY_BUFFER = 1024 * 1024
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Eintraege und Fenster
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def within(root, path):
|
|
"""Liegt `path` innerhalb von `root`?
|
|
|
|
Wichtig ist der Sonderfall root="/" (das lokale Fenster darf ueberall
|
|
hin): dort ist root + "/" gleich "//", worauf kein einziger Pfad passt.
|
|
"""
|
|
root = os.path.realpath(root)
|
|
path = os.path.realpath(path)
|
|
prefix = root if root.endswith(os.sep) else root + os.sep
|
|
return path == root or path.startswith(prefix)
|
|
|
|
|
|
@dataclass
|
|
class Entry:
|
|
name: str
|
|
path: str
|
|
is_dir: bool = False
|
|
size: int = 0
|
|
mtime: float = 0.0
|
|
is_link: bool = False
|
|
is_special: bool = False
|
|
is_up: bool = False
|
|
|
|
@property
|
|
def sort_key(self):
|
|
return (0 if self.is_up else (1 if self.is_dir else 2), self.name.lower())
|
|
|
|
|
|
@dataclass
|
|
class Pane:
|
|
path: str
|
|
root: str = "/"
|
|
readonly: bool = False
|
|
title: str = ""
|
|
entries: list = field(default_factory=list)
|
|
index: int = 0
|
|
top: int = 0
|
|
marks: set = field(default_factory=set)
|
|
error: str = ""
|
|
|
|
def reload(self):
|
|
previous = self.current.name if self.entries and 0 <= self.index < len(self.entries) else ""
|
|
self.entries = []
|
|
self.error = ""
|
|
|
|
real = os.path.realpath(self.path)
|
|
if real != os.path.realpath(self.root):
|
|
parent = os.path.dirname(real) or "/"
|
|
self.entries.append(Entry(name="..", path=parent, is_dir=True, is_up=True))
|
|
|
|
try:
|
|
with os.scandir(self.path) as scan:
|
|
for item in scan:
|
|
self.entries.append(self._entry(item))
|
|
except OSError as exc:
|
|
self.error = "%s: %s" % (self.path, exc.strerror or exc)
|
|
|
|
self.entries.sort(key=lambda e: e.sort_key)
|
|
self.marks = {name for name in self.marks
|
|
if any(e.name == name for e in self.entries)}
|
|
self.index = 0
|
|
for position, entry in enumerate(self.entries):
|
|
if entry.name == previous:
|
|
self.index = position
|
|
break
|
|
self._clamp()
|
|
|
|
@staticmethod
|
|
def _entry(item):
|
|
try:
|
|
info = item.stat(follow_symlinks=False)
|
|
mode = info.st_mode
|
|
is_link = stat.S_ISLNK(mode)
|
|
if is_link:
|
|
try:
|
|
target = item.stat(follow_symlinks=True)
|
|
is_dir = stat.S_ISDIR(target.st_mode)
|
|
except OSError:
|
|
is_dir = False
|
|
special = False
|
|
else:
|
|
is_dir = stat.S_ISDIR(mode)
|
|
special = not (is_dir or stat.S_ISREG(mode))
|
|
return Entry(name=item.name, path=item.path, is_dir=is_dir,
|
|
size=0 if is_dir else info.st_size, mtime=info.st_mtime,
|
|
is_link=is_link, is_special=special)
|
|
except OSError:
|
|
return Entry(name=item.name, path=item.path)
|
|
|
|
# -- Navigation -------------------------------------------------------
|
|
|
|
@property
|
|
def current(self):
|
|
if 0 <= self.index < len(self.entries):
|
|
return self.entries[self.index]
|
|
return Entry(name="", path=self.path)
|
|
|
|
def _clamp(self):
|
|
self.index = max(0, min(self.index, max(0, len(self.entries) - 1)))
|
|
|
|
def move(self, delta):
|
|
self.index += delta
|
|
self._clamp()
|
|
|
|
def open_current(self):
|
|
entry = self.current
|
|
if not entry.is_dir:
|
|
return False
|
|
target = os.path.realpath(entry.path)
|
|
if not within(self.root, target):
|
|
return False # nicht aus dem Snapshot herauslaufen
|
|
self.path = target
|
|
self.marks.clear()
|
|
self.reload()
|
|
return True
|
|
|
|
def go(self, path):
|
|
if os.path.isdir(path):
|
|
self.path = os.path.realpath(path)
|
|
self.marks.clear()
|
|
self.reload()
|
|
return True
|
|
return False
|
|
|
|
def toggle_mark(self):
|
|
entry = self.current
|
|
if entry.is_up or not entry.name:
|
|
return
|
|
if entry.name in self.marks:
|
|
self.marks.discard(entry.name)
|
|
else:
|
|
self.marks.add(entry.name)
|
|
|
|
def selection(self):
|
|
"""Markierte Eintraege, sonst der Eintrag unter dem Balken."""
|
|
if self.marks:
|
|
return [e for e in self.entries if e.name in self.marks]
|
|
entry = self.current
|
|
return [] if (entry.is_up or not entry.name) else [entry]
|
|
|
|
def free_space(self):
|
|
try:
|
|
stats = os.statvfs(self.path)
|
|
return stats.f_bavail * stats.f_frsize
|
|
except OSError:
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Kopieren
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Aborted(Exception):
|
|
"""Der Benutzer hat abgebrochen."""
|
|
|
|
|
|
class Copier:
|
|
"""Kopiert Dateien und Verzeichnisse mit Fortschritt und Abbruch."""
|
|
|
|
def __init__(self, win, progress):
|
|
self.win = win
|
|
self.progress = progress # Rueckruf(text, done, total)
|
|
self.total = 0
|
|
self.done = 0
|
|
self.overwrite = None # None | "alle" | "keine"
|
|
self.copied = 0
|
|
self.skipped = []
|
|
self.failed = []
|
|
|
|
# -- Groesse vorab ----------------------------------------------------
|
|
|
|
def measure(self, entries):
|
|
total = 0
|
|
for entry in entries:
|
|
if entry.is_dir and not entry.is_link:
|
|
for base, _dirs, files in os.walk(entry.path, onerror=None):
|
|
for name in files:
|
|
try:
|
|
info = os.lstat(os.path.join(base, name))
|
|
if stat.S_ISREG(info.st_mode):
|
|
total += info.st_size
|
|
except OSError:
|
|
pass
|
|
elif not entry.is_special:
|
|
total += entry.size
|
|
self.total = total
|
|
return total
|
|
|
|
# -- Ausfuehren -------------------------------------------------------
|
|
|
|
def copy(self, entries, target):
|
|
for entry in entries:
|
|
self._copy_any(entry.path, os.path.join(target, entry.name))
|
|
|
|
def _copy_any(self, source, destination):
|
|
self._check_abort()
|
|
try:
|
|
info = os.lstat(source)
|
|
except OSError as exc:
|
|
self.failed.append("%s: %s" % (source, exc.strerror or exc))
|
|
return
|
|
|
|
if stat.S_ISLNK(info.st_mode):
|
|
self._copy_symlink(source, destination)
|
|
elif stat.S_ISDIR(info.st_mode):
|
|
self._copy_dir(source, destination)
|
|
elif stat.S_ISREG(info.st_mode):
|
|
self._copy_file(source, destination, info)
|
|
else:
|
|
# Geraetedateien, Sockets, FIFOs - im Gast-Dateisystem haeufig,
|
|
# ausserhalb aber sinnlos.
|
|
self.skipped.append("%s (Spezialdatei)" % source)
|
|
|
|
def _copy_symlink(self, source, destination):
|
|
try:
|
|
target = os.readlink(source)
|
|
if os.path.lexists(destination):
|
|
if not self._may_overwrite(destination):
|
|
return
|
|
os.remove(destination)
|
|
os.symlink(target, destination)
|
|
self.copied += 1
|
|
except OSError as exc:
|
|
self.failed.append("%s: %s" % (source, exc.strerror or exc))
|
|
|
|
def _copy_dir(self, source, destination):
|
|
try:
|
|
os.makedirs(destination, exist_ok=True)
|
|
except OSError as exc:
|
|
self.failed.append("%s: %s" % (destination, exc.strerror or exc))
|
|
return
|
|
try:
|
|
names = sorted(os.listdir(source))
|
|
except OSError as exc:
|
|
self.failed.append("%s: %s" % (source, exc.strerror or exc))
|
|
return
|
|
for name in names:
|
|
self._copy_any(os.path.join(source, name), os.path.join(destination, name))
|
|
try:
|
|
shutil.copystat(source, destination)
|
|
except OSError:
|
|
pass
|
|
|
|
def _copy_file(self, source, destination, info):
|
|
if os.path.lexists(destination) and not self._may_overwrite(destination):
|
|
self.done += info.st_size
|
|
return
|
|
self.progress(os.path.basename(source), self.done, self.total)
|
|
try:
|
|
with open(source, "rb") as src, open(destination, "wb") as dst:
|
|
while True:
|
|
self._check_abort()
|
|
chunk = src.read(COPY_BUFFER)
|
|
if not chunk:
|
|
break
|
|
dst.write(chunk)
|
|
self.done += len(chunk)
|
|
self.progress(os.path.basename(source), self.done, self.total)
|
|
shutil.copystat(source, destination)
|
|
self.copied += 1
|
|
except Aborted:
|
|
self._remove_partial(destination)
|
|
raise
|
|
except OSError as exc:
|
|
self._remove_partial(destination)
|
|
if exc.errno == errno.ENOSPC:
|
|
raise SnapfsError("Kein Platz mehr auf dem Ziel-Dateisystem.")
|
|
self.failed.append("%s: %s" % (source, exc.strerror or exc))
|
|
|
|
@staticmethod
|
|
def _remove_partial(path):
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
|
|
# -- Rueckfragen ------------------------------------------------------
|
|
|
|
def _may_overwrite(self, destination):
|
|
if self.overwrite == "alle":
|
|
return True
|
|
if self.overwrite == "keine":
|
|
self.skipped.append("%s (vorhanden)" % destination)
|
|
return False
|
|
answer = ask(self.win,
|
|
"%s gibt es schon: [j]a ueberschreiben, [n]ein, "
|
|
"[a]lle, [k]eine" % truncate(os.path.basename(destination), 40),
|
|
"jnak", default="n")
|
|
if answer == "a":
|
|
self.overwrite = "alle"
|
|
return True
|
|
if answer == "k":
|
|
self.overwrite = "keine"
|
|
if answer == "j":
|
|
return True
|
|
self.skipped.append("%s (vorhanden)" % destination)
|
|
return False
|
|
|
|
def _check_abort(self):
|
|
self.win.nodelay(True)
|
|
try:
|
|
key = self.win.getch()
|
|
except curses.error:
|
|
key = -1
|
|
finally:
|
|
self.win.nodelay(False)
|
|
if key == 27: # ESC
|
|
raise Aborted()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Oberflaeche
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Explorer:
|
|
def __init__(self, left, right, session=None, subtitle=""):
|
|
self.panes = [left, right]
|
|
self.active = 0
|
|
self.session = session
|
|
self.subtitle = subtitle
|
|
self.status = ""
|
|
# Transfer-Laufwerke, die *wir* eingehaengt haben. Nur die haengen wir
|
|
# beim Verlassen wieder aus - ein Laufwerk, das schon vorher am Host
|
|
# hing, gehoert jemand anderem.
|
|
self.mounted_transfers = set()
|
|
|
|
@property
|
|
def source(self):
|
|
return self.panes[self.active]
|
|
|
|
@property
|
|
def target(self):
|
|
return self.panes[1 - self.active]
|
|
|
|
# -- Zeichnen ---------------------------------------------------------
|
|
|
|
def draw(self, win):
|
|
win.erase()
|
|
height, width = win.getmaxyx()
|
|
keys = keybar(width, KEYS)
|
|
list_top = 3
|
|
list_bottom = height - 3 - len(keys)
|
|
rows = max(1, list_bottom - list_top)
|
|
|
|
fill(win, 0, " pvesnap-explorer - %s" % self.subtitle,
|
|
curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
|
|
middle = width // 2
|
|
for row in range(1, list_bottom + 1):
|
|
put(win, row, middle, "│", curses.color_pair(C_DIM))
|
|
|
|
self._draw_pane(win, self.panes[0], 0, middle, list_top, rows, self.active == 0)
|
|
self._draw_pane(win, self.panes[1], middle + 1, width - middle - 2,
|
|
list_top, rows, self.active == 1)
|
|
|
|
entry = self.source.current
|
|
info = self.status or (entry.path if entry.name else self.source.path)
|
|
fill(win, list_bottom + 1, " " + truncate(info, width - 2),
|
|
curses.color_pair(C_DIM))
|
|
|
|
for offset, line in enumerate(reversed(keys)):
|
|
fill(win, height - 1 - offset, " " + line, curses.color_pair(C_FOOTER))
|
|
|
|
def _draw_pane(self, win, pane, x, width, top, rows, active):
|
|
if width < 8:
|
|
return
|
|
header = pane.title or pane.path
|
|
attr = curses.color_pair(C_SEL) if active else curses.color_pair(C_DIM)
|
|
put(win, 1, x, (" " + truncate(header, width - 2)).ljust(width), attr)
|
|
put(win, 2, x, truncate(pane.path, width - 1).ljust(width),
|
|
curses.A_BOLD if active else 0)
|
|
|
|
if pane.error:
|
|
put(win, top, x + 1, truncate(pane.error, width - 2),
|
|
curses.color_pair(C_WARN))
|
|
return
|
|
|
|
if pane.index < pane.top:
|
|
pane.top = pane.index
|
|
if pane.index >= pane.top + rows:
|
|
pane.top = pane.index - rows + 1
|
|
pane.top = max(0, min(pane.top, max(0, len(pane.entries) - rows)))
|
|
|
|
for row in range(rows):
|
|
position = pane.top + row
|
|
if position >= len(pane.entries):
|
|
break
|
|
entry = pane.entries[position]
|
|
marked = entry.name in pane.marks
|
|
|
|
if active and position == pane.index:
|
|
attribute = curses.color_pair(C_SEL)
|
|
elif marked:
|
|
attribute = curses.color_pair(C_MARK) | curses.A_BOLD
|
|
elif entry.is_dir:
|
|
attribute = curses.color_pair(C_DIR)
|
|
else:
|
|
attribute = 0
|
|
|
|
put(win, top + row, x, self._format(entry, width, marked), attribute)
|
|
|
|
total = len(pane.entries)
|
|
put(win, top + rows, x, " %d Eintraege%s"
|
|
% (total, " %d markiert" % len(pane.marks) if pane.marks else ""),
|
|
curses.color_pair(C_DIM))
|
|
|
|
@staticmethod
|
|
def _format(entry, width, marked):
|
|
"""Eine Zeile mit genau `width` Zeichen - laenger wuerde ins andere
|
|
Fenster hineinschreiben."""
|
|
mark = "*" if marked else " "
|
|
suffix = "@" if entry.is_link else ""
|
|
|
|
def sized(text):
|
|
return text.ljust(width)[:width]
|
|
|
|
if entry.is_up:
|
|
return sized("%s.. <hoch>" % mark)
|
|
|
|
if entry.is_dir:
|
|
size = "<DIR>"
|
|
elif entry.is_special:
|
|
size = "<SPEZ>"
|
|
else:
|
|
size = human_bytes(entry.size)
|
|
|
|
when = (datetime.fromtimestamp(entry.mtime).strftime("%d.%m.%y %H:%M")
|
|
if entry.mtime else "")
|
|
|
|
# 1 Markierung + Name + 1 + 8 Groesse + 1 + 14 Datum
|
|
room = width - 25
|
|
if room < 8:
|
|
return sized("%s%s" % (mark, truncate(entry.name + suffix, max(1, width - 2))))
|
|
name = truncate(entry.name + suffix, room).ljust(room)
|
|
return sized("%s%s %s %s" % (mark, name, size.rjust(8), when.rjust(14)))
|
|
|
|
# -- Hauptschleife ----------------------------------------------------
|
|
|
|
def run(self, win):
|
|
for pane in self.panes:
|
|
pane.reload()
|
|
try:
|
|
self._loop(win)
|
|
finally:
|
|
# Auch bei Absturz oder Strg-C: ein Transfer-Laufwerk, das am Host
|
|
# eingehaengt bleibt, gilt spaeter als belegt und laesst sich dann
|
|
# nicht in eine Maschine haengen.
|
|
self._release_transfers()
|
|
|
|
def _loop(self, win):
|
|
while True:
|
|
self.draw(win)
|
|
win.refresh()
|
|
key = read_key(win)
|
|
rows = max(1, win.getmaxyx()[0] - 7)
|
|
self.status = ""
|
|
pane = self.source
|
|
|
|
if key == "\t" or key == curses.KEY_BTAB:
|
|
self.active = 1 - self.active
|
|
elif key in (curses.KEY_UP, "k"):
|
|
pane.move(-1)
|
|
elif key in (curses.KEY_DOWN, "j"):
|
|
pane.move(1)
|
|
elif key == curses.KEY_NPAGE:
|
|
pane.move(rows)
|
|
elif key == curses.KEY_PPAGE:
|
|
pane.move(-rows)
|
|
elif key == curses.KEY_HOME:
|
|
pane.index = 0
|
|
elif key == curses.KEY_END:
|
|
pane.index = len(pane.entries) - 1
|
|
elif is_enter(key):
|
|
if not pane.open_current():
|
|
self._view(win, pane.current)
|
|
elif key in (curses.KEY_BACKSPACE, "\x7f", curses.KEY_LEFT):
|
|
if pane.entries and pane.entries[0].is_up:
|
|
pane.index = 0
|
|
pane.open_current()
|
|
elif key == " ":
|
|
pane.toggle_mark()
|
|
pane.move(1)
|
|
elif key == "*":
|
|
for entry in pane.entries:
|
|
if entry.is_up:
|
|
continue
|
|
if entry.name in pane.marks:
|
|
pane.marks.discard(entry.name)
|
|
else:
|
|
pane.marks.add(entry.name)
|
|
elif key in ("a", "A"):
|
|
pane.marks = {e.name for e in pane.entries if not e.is_up}
|
|
elif key in ("u", "U"):
|
|
pane.marks.clear()
|
|
elif key in (curses.KEY_F5, "c", "C"):
|
|
self._copy(win)
|
|
elif key in (curses.KEY_F3, "v", "V"):
|
|
self._view(win, pane.current)
|
|
elif key in (curses.KEY_F7, "n", "N"):
|
|
self._mkdir(win)
|
|
elif key in (curses.KEY_F2, "m", "M"):
|
|
self._switch_mount(win)
|
|
elif key in (curses.KEY_F6, "g", "G"):
|
|
self._goto(win)
|
|
elif key in ("r", "R", curses.KEY_F9):
|
|
for item in self.panes:
|
|
item.reload()
|
|
elif key in ("h", "?", curses.KEY_F1):
|
|
self._help(win)
|
|
elif key in (curses.KEY_F10, "q", "Q"):
|
|
return
|
|
elif is_escape(key):
|
|
if confirm(win, "Snapshot schliessen und zurueck zur Auswahl?"):
|
|
return
|
|
|
|
# -- Aktionen ---------------------------------------------------------
|
|
|
|
def _copy(self, win):
|
|
source, target = self.source, self.target
|
|
if target.readonly:
|
|
message(win, "Das Ziel ist schreibgeschuetzt - ein Snapshot laesst sich "
|
|
"nicht beschreiben.", error=True)
|
|
return
|
|
|
|
entries = source.selection()
|
|
if not entries:
|
|
message(win, "Nichts ausgewaehlt.", error=True)
|
|
return
|
|
|
|
def progress(name, done, total):
|
|
height, width = win.getmaxyx()
|
|
share = (done / total * 100) if total else 100
|
|
bar_width = max(10, width - 46)
|
|
filled = int(bar_width * share / 100)
|
|
fill(win, height - 2,
|
|
" %s [%s%s] %3d%% %s / %s (ESC bricht ab)"
|
|
% (truncate(name, 24).ljust(24), "#" * filled,
|
|
"." * (bar_width - filled), share,
|
|
human_bytes(done), human_bytes(total)),
|
|
curses.color_pair(C_SEL))
|
|
win.refresh()
|
|
|
|
copier = Copier(win, progress)
|
|
progress("Groesse ermitteln ...", 0, 0)
|
|
total = copier.measure(entries)
|
|
|
|
free = target.free_space()
|
|
if free and total > free:
|
|
if not confirm(win, "Es werden %s benoetigt, frei sind nur %s - trotzdem?"
|
|
% (human_bytes(total), human_bytes(free))):
|
|
return
|
|
|
|
what = ("%d Eintraege" % len(entries)) if len(entries) > 1 else entries[0].name
|
|
if not confirm(win, "%s (%s) nach %s kopieren?"
|
|
% (truncate(what, 30), human_bytes(total), target.path)):
|
|
return
|
|
|
|
started = time.time()
|
|
try:
|
|
copier.copy(entries, target.path)
|
|
summary = ("%d Objekt(e) kopiert, %s in %.1fs"
|
|
% (copier.copied, human_bytes(copier.done), time.time() - started))
|
|
except Aborted:
|
|
summary = "Abgebrochen - %d Objekt(e) waren fertig" % copier.copied
|
|
except SnapfsError as exc:
|
|
summary = str(exc)
|
|
|
|
source.marks.clear()
|
|
target.reload()
|
|
|
|
if copier.failed or copier.skipped:
|
|
self._report(win, summary, copier)
|
|
else:
|
|
message(win, summary)
|
|
|
|
def _report(self, win, summary, copier):
|
|
height, width = win.getmaxyx()
|
|
lines = [summary, ""]
|
|
if copier.failed:
|
|
lines.append("Fehlgeschlagen (%d):" % len(copier.failed))
|
|
lines += [" " + item for item in copier.failed[:10]]
|
|
if copier.skipped:
|
|
lines.append("Uebersprungen (%d):" % len(copier.skipped))
|
|
lines += [" " + item for item in copier.skipped[:10]]
|
|
|
|
win.erase()
|
|
fill(win, 0, " Ergebnis", curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
for row, line in enumerate(lines[: height - 3]):
|
|
put(win, row + 2, 2, truncate(line, width - 4))
|
|
fill(win, height - 1, " beliebige Taste = weiter", curses.color_pair(C_FOOTER))
|
|
win.refresh()
|
|
read_key(win)
|
|
|
|
def _mkdir(self, win):
|
|
pane = self.source
|
|
if pane.readonly:
|
|
message(win, "Im Snapshot laesst sich nichts anlegen.", error=True)
|
|
return
|
|
name = prompt(win, "Neues Verzeichnis", "")
|
|
if not name:
|
|
return
|
|
try:
|
|
os.makedirs(os.path.join(pane.path, name), exist_ok=False)
|
|
pane.reload()
|
|
except OSError as exc:
|
|
message(win, "Nicht angelegt: %s" % (exc.strerror or exc), error=True)
|
|
|
|
def _goto(self, win):
|
|
pane = self.source
|
|
path = prompt(win, "Verzeichnis wechseln nach", pane.path)
|
|
if not path:
|
|
return
|
|
if not pane.go(os.path.expanduser(path)):
|
|
message(win, "Kein Verzeichnis: %s" % path, error=True)
|
|
|
|
def _switch_mount(self, win):
|
|
"""Laufwerk waehlen - Dateisystem des Snapshots oder Transfer-Laufwerk.
|
|
|
|
Beides steht in derselben Liste, weil es aus Sicht des Bedieners
|
|
dieselbe Frage ist: "wo soll ich hinschauen?". Ein Transfer-Laufwerk
|
|
hier zu erreichen spart den Umweg ueber pvesnap-recovery - man kann
|
|
aus dem Snapshot direkt auf das Austauschmedium kopieren, das
|
|
anschliessend in die wiederhergestellte Maschine wandert.
|
|
"""
|
|
entries = []
|
|
|
|
filesystems = self.session.mount_entries() if self.session else []
|
|
if len(filesystems) > 1:
|
|
for path, text in filesystems:
|
|
entries.append((("fs", path), "Snapshot: %s" % text))
|
|
|
|
for volume in self._transfer_volumes():
|
|
state = "" if volume.free or volume.mountpoint \
|
|
else " [%s]" % volume.state_text
|
|
entries.append((("transfer", volume.name),
|
|
"Transfer: %-14s %8s %s%s"
|
|
% (volume.name, human_bytes(volume.size),
|
|
truncate(volume.note, 26), state)))
|
|
|
|
if not entries:
|
|
message(win, "Es ist nur ein Dateisystem eingebunden, und "
|
|
"Transfer-Laufwerke gibt es noch keine.")
|
|
return
|
|
|
|
chosen = choose(win, "Welches Laufwerk?", entries)
|
|
if not chosen:
|
|
return
|
|
kind, value = chosen
|
|
if kind == "fs":
|
|
self.panes[0].root = value
|
|
self.panes[0].go(value)
|
|
else:
|
|
self._open_transfer(win, value)
|
|
|
|
def _transfer_volumes(self):
|
|
try:
|
|
return transfer.load()
|
|
except transfer.TransferError as exc:
|
|
self.status = str(exc)
|
|
return []
|
|
|
|
def _open_transfer(self, win, name):
|
|
"""Ein Transfer-Laufwerk einhaengen und ins Arbeitsfenster holen.
|
|
|
|
Ins Arbeitsfenster, nicht ins Snapshot-Fenster: links steht der
|
|
Snapshot, aus dem geholt wird, rechts das Ziel. Genau dorthin gehoert
|
|
ein Austauschmedium. Gibt es keinen Snapshot (Aufruf aus der
|
|
Laufwerksverwaltung heraus), landet es im gerade aktiven Fenster.
|
|
"""
|
|
target = self.panes[1] if self.session else self.source
|
|
try:
|
|
already = bool(transfer.get(name).mountpoint)
|
|
path = transfer.host_mount(name)
|
|
except transfer.TransferError as exc:
|
|
message(win, str(exc), error=True)
|
|
return
|
|
if not already:
|
|
self.mounted_transfers.add(name)
|
|
|
|
target.root = "/"
|
|
target.marks.clear()
|
|
if not target.go(path):
|
|
message(win, "Eingehaengt unter %s, aber nicht lesbar." % path, error=True)
|
|
return
|
|
target.title = "Transfer: %s" % name
|
|
self.status = "%s eingehaengt unter %s" % (name, path)
|
|
|
|
def _release_transfers(self):
|
|
"""Was wir eingehaengt haben, geben wir auch wieder frei."""
|
|
for name in sorted(self.mounted_transfers):
|
|
try:
|
|
transfer.host_umount(name)
|
|
except transfer.TransferError as exc:
|
|
log.warning("Transfer-Laufwerk %s blieb eingehaengt: %s", name, exc)
|
|
self.mounted_transfers.clear()
|
|
|
|
def _view(self, win, entry):
|
|
if not entry.name or entry.is_dir or entry.is_special:
|
|
return
|
|
try:
|
|
with open(entry.path, "rb") as handle:
|
|
raw = handle.read(1024 * 1024)
|
|
except OSError as exc:
|
|
message(win, "Nicht lesbar: %s" % (exc.strerror or exc), error=True)
|
|
return
|
|
|
|
binary = b"\0" in raw[:8192]
|
|
if binary:
|
|
lines = _hexdump(raw[: 64 * 1024])
|
|
else:
|
|
lines = raw.decode("utf-8", "replace").splitlines() or ["(leer)"]
|
|
|
|
top = 0
|
|
while True:
|
|
height, width = win.getmaxyx()
|
|
rows = height - 3
|
|
win.erase()
|
|
fill(win, 0, " %s %s%s" % (truncate(entry.path, width - 30),
|
|
human_bytes(entry.size),
|
|
" [binaer, Hex-Ansicht]" if binary else ""),
|
|
curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
for row in range(rows):
|
|
position = top + row
|
|
if position >= len(lines):
|
|
break
|
|
put(win, row + 1, 0, truncate(lines[position].rstrip("\n"), width - 1))
|
|
fill(win, height - 1,
|
|
" Pfeiltasten/Bild blaettern | Pos1/Ende | q zurueck (Zeile %d von %d)"
|
|
% (min(top + 1, len(lines)), len(lines)), curses.color_pair(C_FOOTER))
|
|
win.refresh()
|
|
|
|
key = read_key(win)
|
|
if key in (curses.KEY_DOWN, "j"):
|
|
top += 1
|
|
elif key in (curses.KEY_UP, "k"):
|
|
top -= 1
|
|
elif key == curses.KEY_NPAGE or key == " ":
|
|
top += rows
|
|
elif key == curses.KEY_PPAGE:
|
|
top -= rows
|
|
elif key == curses.KEY_HOME:
|
|
top = 0
|
|
elif key == curses.KEY_END:
|
|
top = len(lines) - rows
|
|
elif key in ("q", "Q", curses.KEY_F3, curses.KEY_F10) or is_escape(key):
|
|
return
|
|
top = max(0, min(top, max(0, len(lines) - 1)))
|
|
|
|
def _help(self, win):
|
|
lines = [
|
|
"Tab zwischen den Fenstern wechseln",
|
|
"Pfeiltasten bewegen, Enter oeffnet Verzeichnis oder zeigt Datei",
|
|
"Backspace ein Verzeichnis hoch",
|
|
"Leertaste markieren, * umkehren, a alle, u keine",
|
|
"F5 / c markiertes ins andere Fenster kopieren",
|
|
"F3 / v Datei ansehen (Text oder Hex)",
|
|
"F7 / n neues Verzeichnis (nur im lokalen Fenster)",
|
|
"F6 / g Verzeichnis direkt eingeben",
|
|
"F2 / m Laufwerk waehlen: anderes Dateisystem des Snapshots",
|
|
" oder ein Transfer-Laufwerk ins rechte Fenster",
|
|
"r neu einlesen",
|
|
"F10 / q zurueck zur Snapshot-Auswahl",
|
|
"",
|
|
"Der Snapshot ist schreibgeschuetzt eingebunden - es kann nichts",
|
|
"kaputtgehen. Kopiert wird immer vom aktiven ins andere Fenster.",
|
|
"Geraetedateien und Sockets werden uebersprungen, symbolische",
|
|
"Verweise bleiben Verweise.",
|
|
"",
|
|
"Transfer-Laufwerke (F2) sind Austauschmedien fuer Maschinen ohne",
|
|
"Netz: hier befuellen, spaeter in die Wiederherstellung haengen.",
|
|
"Beim Verlassen werden sie wieder ausgehaengt.",
|
|
"",
|
|
"Es gibt drei Ebenen: Gastauswahl -> Snapshot-Auswahl -> Dateien.",
|
|
"F10 oder q geht jeweils eine Ebene zurueck; der Snapshot wird",
|
|
"dabei wieder ausgehaengt. Das rechte Fenster behaelt sein",
|
|
"Verzeichnis, auch wenn man den naechsten Snapshot oeffnet.",
|
|
]
|
|
height, width = win.getmaxyx()
|
|
win.erase()
|
|
fill(win, 0, " Tastenbelegung", curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
for row, line in enumerate(lines[: height - 3]):
|
|
put(win, row + 2, 3, truncate(line, width - 5))
|
|
fill(win, height - 1, " beliebige Taste = zurueck", curses.color_pair(C_FOOTER))
|
|
win.refresh()
|
|
read_key(win)
|
|
|
|
|
|
def _hexdump(raw, width=16):
|
|
lines = []
|
|
for offset in range(0, len(raw), width):
|
|
chunk = raw[offset:offset + width]
|
|
hexpart = " ".join("%02x" % byte for byte in chunk).ljust(width * 3)
|
|
text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
|
|
lines.append("%08x %s |%s|" % (offset, hexpart, text))
|
|
return lines
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auswahl von Gast und Snapshot
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _pick_guest(win, guests):
|
|
entries = [(g, "%s %-6d %-24s %-8s %s"
|
|
% ("LXC" if g.type == "lxc" else "VM ", g.vmid,
|
|
truncate(g.name or "-", 24), g.status, g.node))
|
|
for g in guests]
|
|
return choose(win, "Welcher Gast? (Esc beendet)", entries)
|
|
|
|
|
|
def _pick_snapshot(win, snapshots):
|
|
entries = []
|
|
for snap in snapshots:
|
|
when = (datetime.fromtimestamp(snap.snaptime).strftime("%d.%m.%Y %H:%M")
|
|
if snap.snaptime else "?")
|
|
if not getattr(snap, "complete", True):
|
|
continue
|
|
entries.append((snap.name, "%-34s %s %s"
|
|
% (truncate(snap.name, 34), when,
|
|
truncate(snap.description.replace("\n", " "), 40))))
|
|
return choose(win, "Welcher Snapshot? (Esc = zurueck zur Gastauswahl)", entries)
|
|
|
|
|
|
def _wait_screen(win, text):
|
|
height, width = win.getmaxyx()
|
|
win.erase()
|
|
fill(win, 0, " pvesnap-explorer", curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
put(win, height // 2, max(2, (width - len(text)) // 2), text, curses.A_BOLD)
|
|
win.refresh()
|
|
|
|
|
|
def _show_notes(win, notes, title="Hinweise"):
|
|
if not notes:
|
|
return
|
|
height, width = win.getmaxyx()
|
|
win.erase()
|
|
fill(win, 0, " %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
|
|
for row, note in enumerate(notes[: height - 3]):
|
|
put(win, row + 2, 2, truncate("- " + note, width - 4), curses.color_pair(C_WARN))
|
|
fill(win, height - 1, " beliebige Taste = weiter", curses.color_pair(C_FOOTER))
|
|
win.refresh()
|
|
read_key(win)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ablauf
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _main(stdscr, args):
|
|
curses.curs_set(0)
|
|
if hasattr(curses, "set_escdelay"):
|
|
curses.set_escdelay(25)
|
|
init_colors()
|
|
stdscr.keypad(True)
|
|
|
|
local = Pane(path=os.path.realpath(args.local), root="/", readonly=False,
|
|
title="Lokaler Rechner")
|
|
|
|
if args.path:
|
|
left = Pane(path=os.path.realpath(args.path), root=os.path.realpath(args.path),
|
|
readonly=True, title="Verzeichnis: %s" % args.path)
|
|
Explorer(left, local, subtitle=args.path).run(stdscr)
|
|
return 0
|
|
|
|
_wait_screen(stdscr, "Frage Proxmox nach VMs und Containern ...")
|
|
proxmox = Proxmox()
|
|
guests = proxmox.inventory()
|
|
if not guests:
|
|
message(stdscr, "Keine VMs oder Container gefunden.", error=True)
|
|
return 1
|
|
|
|
# Drei Ebenen, aus denen man jeweils eine zurueckkommt:
|
|
# Gastauswahl -> Snapshot-Auswahl -> Dateibrowser
|
|
# Der Snapshot wird beim Verlassen des Browsers wieder ausgehaengt, damit
|
|
# man gleich den naechsten oeffnen kann.
|
|
wanted_vmid, wanted_snapshot = args.vmid, args.snapshot
|
|
|
|
while True:
|
|
if wanted_vmid:
|
|
guest = next((g for g in guests if g.vmid == wanted_vmid), None)
|
|
wanted_vmid = None
|
|
if guest is None:
|
|
message(stdscr, "Gast %d nicht gefunden." % args.vmid, error=True)
|
|
return 2
|
|
else:
|
|
guest = _pick_guest(stdscr, guests)
|
|
if guest is None:
|
|
return 0
|
|
|
|
while True:
|
|
_wait_screen(stdscr, "Lese Snapshots von %s ..." % guest.label)
|
|
try:
|
|
snapshots = list_snapshots(proxmox, guest)
|
|
except ProxmoxError as exc:
|
|
message(stdscr, str(exc), error=True)
|
|
break
|
|
|
|
broken = [s for s in snapshots if not s.complete]
|
|
snapshots = [s for s in snapshots if s.complete]
|
|
|
|
if wanted_snapshot:
|
|
name = next((s.name for s in snapshots if s.name == wanted_snapshot), None)
|
|
if name is None:
|
|
message(stdscr, "Snapshot %r nicht gefunden."
|
|
% wanted_snapshot, error=True)
|
|
wanted_snapshot = None
|
|
if name is None:
|
|
continue
|
|
elif not snapshots:
|
|
text = "%s hat keine Snapshots." % guest.label
|
|
if broken:
|
|
text += (" %d halbfertige(r) Eintrag/Eintraege aus abgebrochenen "
|
|
"Laeufen sind vorhanden - aufraeumen mit: %s delsnapshot "
|
|
"%d <name> --force"
|
|
% (len(broken), "pct" if guest.type == "lxc" else "qm",
|
|
guest.vmid))
|
|
message(stdscr, text, error=True)
|
|
break
|
|
else:
|
|
name = _pick_snapshot(stdscr, snapshots)
|
|
if name is None:
|
|
break # zurueck zur Gastauswahl
|
|
|
|
_browse(stdscr, proxmox, guest, name, local)
|
|
|
|
# Beim Zurueckkommen kann sich das Inventar geaendert haben.
|
|
_wait_screen(stdscr, "Frage Proxmox nach VMs und Containern ...")
|
|
try:
|
|
guests = proxmox.inventory(refresh=True)
|
|
except ProxmoxError as exc:
|
|
message(stdscr, str(exc), error=True)
|
|
return 3
|
|
|
|
|
|
def _browse(stdscr, proxmox, guest, snapname, local):
|
|
"""Einen Snapshot einbinden, durchsehen und wieder aushaengen."""
|
|
session = Session(proxmox, guest, snapname)
|
|
try:
|
|
_wait_screen(stdscr,
|
|
"Binde Snapshot '%s' ein - das kann etwas dauern ..." % snapname)
|
|
session.open()
|
|
session.prepare()
|
|
except (SnapfsError, ProxmoxError) as exc:
|
|
session.close()
|
|
_show_notes(stdscr, str(exc).splitlines() + session.notes, "Fehlgeschlagen")
|
|
return
|
|
|
|
try:
|
|
_show_notes(stdscr, session.notes)
|
|
entries = session.mount_entries()
|
|
if not entries:
|
|
_show_notes(stdscr,
|
|
["Es wurde kein Dateisystem eingehaengt."] + session.notes,
|
|
"Nichts zu zeigen")
|
|
return
|
|
|
|
start = entries[0][0]
|
|
if len(entries) > 1:
|
|
start = choose(stdscr, "Welches Dateisystem zuerst?", entries)
|
|
if start is None:
|
|
return # zurueck zur Snapshot-Auswahl
|
|
|
|
left = Pane(path=start, root=start, readonly=True,
|
|
title="%s @ %s" % (guest.label, snapname))
|
|
Explorer(left, local, session=session,
|
|
subtitle="%s - Snapshot %s" % (guest.label, snapname)).run(stdscr)
|
|
finally:
|
|
_wait_screen(stdscr, "Haenge Snapshot wieder aus ...")
|
|
session.close()
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(
|
|
prog="pvesnap-explorer",
|
|
description="Dateien aus einem Proxmox-Snapshot holen (zwei Fenster wie im "
|
|
"Midnight Commander).")
|
|
parser.add_argument("vmid", nargs="?", type=int, help="VMID des Gastes")
|
|
parser.add_argument("snapshot", nargs="?", help="Name des Snapshots")
|
|
parser.add_argument("-l", "--local", default=os.getcwd(),
|
|
help="Startverzeichnis im rechten Fenster (Vorgabe: aktuelles)")
|
|
parser.add_argument("-p", "--path",
|
|
help="beliebiges Verzeichnis oeffnen, ohne Proxmox zu fragen")
|
|
parser.add_argument("--cleanup", action="store_true",
|
|
help="haengengebliebene Einbindungen aufraeumen und beenden")
|
|
parser.add_argument("-v", "--verbose", action="store_true")
|
|
parser.add_argument("-V", "--version", action="version",
|
|
version="pvesnap-explorer %s" % __version__)
|
|
return parser
|
|
|
|
|
|
def main(argv=None):
|
|
args = build_parser().parse_args(argv)
|
|
|
|
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(levelname)s %(message)s")
|
|
|
|
if args.cleanup:
|
|
removed = cleanup_leftovers(verbose=True)
|
|
print("%d Eintrag/Eintraege aufgeraeumt." % len(removed))
|
|
return 0
|
|
|
|
cleanup_leftovers(only_dead=True)
|
|
|
|
if not args.path:
|
|
if not pvesh_available():
|
|
print("'pvesh' nicht gefunden - der Explorer laeuft auf dem Proxmox-Host.\n"
|
|
"Zum Ausprobieren: pvesnap-explorer --path /ein/verzeichnis",
|
|
file=sys.stderr)
|
|
return 3
|
|
if os.geteuid() != 0:
|
|
print("Der Explorer braucht root-Rechte (Snapshots einbinden und mounten).",
|
|
file=sys.stderr)
|
|
return 4
|
|
|
|
# Waehrend der Oberflaeche darf nichts ins Terminal protokolliert werden.
|
|
handlers = logging.getLogger().handlers[:]
|
|
if not args.verbose:
|
|
logging.getLogger().handlers = []
|
|
try:
|
|
return curses.wrapper(_main, args)
|
|
except KeyboardInterrupt:
|
|
return 130
|
|
finally:
|
|
logging.getLogger().handlers = handlers
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
sys.exit(main())
|