duffyduckandClaude Opus 5
700492391a
Auf dem Testhost (Ceph RBD) meldete lsblk fuer jede Partition eines
eingebundenen Snapshots "kein Dateisystem" - auch fuer eine 31,5-GB-
Partition, auf der offensichtlich ext4 liegt. Nichts liess sich einhaengen.
Grund: lsblk nimmt FSTYPE aus der udev-Datenbank, und die ist bei frisch
gemappten rbd-Geraeten leer, weil udev dort keine blkid-Probe faehrt. Ein
direktes "blkid -p" auf denselben Geraeten liefert dagegen sauber vfat und
ext4. Der Typ wird jetzt so ermittelt, wenn lsblk nichts weiss.
Ausserdem: die Auswahl der Dateisysteme zeigte nur abgeschnittene
Mountpfade, die sich alle glichen, und vorausgewaehlt war das erste
gefundene - in der Praxis gern die 200-MB-EFI-Partition. Jetzt stehen dort
Geraet, Typ, Groesse und ein Hinweis ("Linux-Wurzelverzeichnis",
"Startpartition"), und die wahrscheinlichste Wurzel steht oben.
Auf pvetest01 gegen echte Snapshots geprueft: einbinden, alle drei
Dateisysteme mounten, Explorer, Web-Oberflaeche mit Download und ZIP,
Ausbruchversuch abgewiesen, restloses Aufraeumen auch nach hartem Abbruch.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
933 lines
33 KiB
Python
933 lines
33 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.
|
|
|
|
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 .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", "Ende", "Ende"),
|
|
]
|
|
|
|
COPY_BUFFER = 1024 * 1024
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Eintraege und Fenster
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@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)
|
|
root = os.path.realpath(self.root)
|
|
if not (target == root or target.startswith(root + os.sep)):
|
|
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 = ""
|
|
|
|
@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()
|
|
|
|
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, "Explorer beenden?"):
|
|
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):
|
|
if not self.session or len(self.session.mounts) < 2:
|
|
message(win, "Es ist nur ein Dateisystem eingebunden.")
|
|
return
|
|
entries = self.session.mount_entries()
|
|
chosen = choose(win, "Welches Dateisystem?", entries)
|
|
if chosen:
|
|
self.panes[0].root = chosen
|
|
self.panes[0].go(chosen)
|
|
|
|
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 anderes Dateisystem des Snapshots waehlen",
|
|
"r neu einlesen",
|
|
"F10 / q beenden",
|
|
"",
|
|
"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.",
|
|
]
|
|
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?", 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 "?")
|
|
entries.append((snap.name, "%-34s %s %s"
|
|
% (truncate(snap.name, 34), when,
|
|
truncate(snap.description.replace("\n", " "), 40))))
|
|
return choose(win, "Welcher Snapshot?", 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
|
|
|
|
if args.vmid:
|
|
guest = next((g for g in guests if g.vmid == args.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
|
|
|
|
_wait_screen(stdscr, "Lese Snapshots von %s ..." % guest.label)
|
|
snapshots = list_snapshots(proxmox, guest)
|
|
if not snapshots:
|
|
message(stdscr, "%s hat keine Snapshots." % guest.label, error=True)
|
|
return 1
|
|
|
|
if args.snapshot:
|
|
name = next((s.name for s in snapshots if s.name == args.snapshot), None)
|
|
if name is None:
|
|
message(stdscr, "Snapshot %r nicht gefunden." % args.snapshot, error=True)
|
|
return 2
|
|
else:
|
|
name = _pick_snapshot(stdscr, snapshots)
|
|
if name is None:
|
|
return 0
|
|
|
|
session = Session(proxmox, guest, name)
|
|
try:
|
|
_wait_screen(stdscr, "Binde Snapshot '%s' ein - das kann etwas dauern ..." % name)
|
|
session.open()
|
|
session.prepare()
|
|
except (SnapfsError, ProxmoxError) as exc:
|
|
session.close()
|
|
_show_notes(stdscr, str(exc).splitlines() + session.notes, "Fehlgeschlagen")
|
|
return 3
|
|
|
|
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 3
|
|
start = entries[0][0]
|
|
if len(entries) > 1:
|
|
start = choose(stdscr, "Welches Dateisystem zuerst?", entries) or start
|
|
|
|
left = Pane(path=start, root=start, readonly=True,
|
|
title="%s @ %s" % (guest.label, name))
|
|
Explorer(left, local, session=session,
|
|
subtitle="%s - Snapshot %s" % (guest.label, name)).run(stdscr)
|
|
finally:
|
|
_wait_screen(stdscr, "Haenge Snapshot wieder aus ...")
|
|
session.close()
|
|
return 0
|
|
|
|
|
|
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
|
|
|
|
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())
|