Hitonabi
8554e7b29c
Behebt 3 von 4 Backup-Luecken (Schicht 1, lokal): - backup.sh sichert jetzt den ECHTEN Zustand als ein Tarball mc2-state-<ts>.tar.gz: mem0 (Chroma+history.db), ~/.hermes (config.yaml, .env, plugins/), llama-swap config. Vorher wurde nur die alte/leere mc2-memory.db gesichert. chmod 600 (enthaelt .env). - restore.sh: --list / --dry-run / [--yes] <datei|latest>; macht VOR dem Zurueckspielen ein Sicherheits-Backup, stoppt/startet Dienste, Health-Check. Live round-trip verifiziert. - mc2-backup.timer/.service: taegliches Backup ~03:30 (vorher gab es KEINE Automatik). - backend/services/backup.py delegiert an backup.sh (eine Quelle der Wahrheit); UI-Button + /api/system/backups zeigen die Tarballs. - docs/BACKUP.md: Backup/Restore-Anleitung. Offen (Schicht 2): Off-Box-Spiegel (Box ist Bare Metal -> PBS-Client oder rsync in LXC). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
"""
|
|
Voll-Zustands-Backup (mem0 + Hermes-Configs/Secrets + llama-swap config).
|
|
Delegiert an deploy/backup.sh (eine Quelle der Wahrheit, identisch zum systemd-Timer);
|
|
Restore läuft bewusst nur per CLI (deploy/restore.sh) — siehe docs/BACKUP.md.
|
|
"""
|
|
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
from config import MODELS_DIR
|
|
|
|
BACKUP_DIR = Path(MODELS_DIR) / "mc2-backups"
|
|
SRC_ROOT = Path(__file__).resolve().parents[2]
|
|
BACKUP_SH = SRC_ROOT / "deploy" / "backup.sh"
|
|
|
|
|
|
def _ts(p: Path) -> str:
|
|
"""Zeitstempel aus 'mc2-state-<ts>.tar.gz' (Path.stem ließe '.tar' stehen)."""
|
|
return p.name[len("mc2-state-"):-len(".tar.gz")]
|
|
|
|
|
|
def _latest() -> Path | None:
|
|
if not BACKUP_DIR.exists():
|
|
return None
|
|
snaps = sorted(BACKUP_DIR.glob("mc2-state-*.tar.gz"), reverse=True)
|
|
return snaps[0] if snaps else None
|
|
|
|
|
|
def _components(tarball: Path) -> list[str]:
|
|
"""Top-Level-Einträge im Tarball (zur Anzeige im UI)."""
|
|
try:
|
|
with tarfile.open(tarball, "r:gz") as t:
|
|
top = {m.name.split("/")[1] for m in t.getmembers()
|
|
if m.name.startswith("./") and "/" in m.name[2:]}
|
|
top |= {m.name[2:] for m in t.getmembers()
|
|
if m.name.startswith("./") and "/" not in m.name[2:] and m.isfile()}
|
|
return sorted(x for x in top if x)
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def backup_now() -> dict:
|
|
"""Erstellt einen Voll-Zustands-Snapshot via deploy/backup.sh."""
|
|
try:
|
|
r = subprocess.run(["/bin/bash", str(BACKUP_SH)], capture_output=True, text=True, timeout=180)
|
|
if r.returncode != 0:
|
|
return {"ok": False, "snapshot": "", "files": [], "error": (r.stderr or r.stdout).strip()[-300:]}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"ok": False, "snapshot": "", "files": [], "error": str(exc)}
|
|
|
|
latest = _latest()
|
|
if not latest:
|
|
return {"ok": False, "snapshot": "", "files": [], "error": "Kein Backup erzeugt"}
|
|
return {
|
|
"ok": True,
|
|
"snapshot": _ts(latest),
|
|
"files": _components(latest),
|
|
"size_mb": round(latest.stat().st_size / 1_000_000, 2),
|
|
}
|
|
|
|
|
|
def list_backups() -> list[dict]:
|
|
if not BACKUP_DIR.exists():
|
|
return []
|
|
out = []
|
|
for p in sorted(BACKUP_DIR.glob("mc2-state-*.tar.gz"), reverse=True):
|
|
out.append({
|
|
"snapshot": _ts(p),
|
|
"file": p.name,
|
|
"size_mb": round(p.stat().st_size / 1_000_000, 2),
|
|
})
|
|
return out
|