Dominic Gundel
8e88efc094
Migrationen 003–009, Kundaufträge mit Lieferschein, Abräumen mit Urteil, ntfy/Backup/Zeitfenster, Live-Stream und Mobile-Mehr-Menü.
93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
"""Datenbank- und Datenordner-Backup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import zipfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
def backup_dir(cfg):
|
|
base = Path(cfg.get("data_dir") or "data")
|
|
path = base / "backup"
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def _safe_name(name):
|
|
name = re.sub(r"[^\w.\-]+", "_", name or "")
|
|
return name[:180] or "backup.zip"
|
|
|
|
|
|
def create_backup(cfg):
|
|
"""ZIP mit pg_dump (stdout) und data/ ohne toolpath/ und backup/."""
|
|
data_dir = Path(cfg.get("data_dir") or "data")
|
|
out_dir = backup_dir(cfg)
|
|
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
filename = _safe_name("grotti-%s.zip" % stamp)
|
|
dest = out_dir / filename
|
|
|
|
url = cfg.get("database_url") or ""
|
|
parsed = urlparse(url)
|
|
db_name = (parsed.path or "/grotti").lstrip("/") or "grotti"
|
|
env = {}
|
|
if parsed.password:
|
|
env["PGPASSWORD"] = parsed.password
|
|
|
|
cmd = ["pg_dump", "--no-owner", "--no-acl", db_name]
|
|
if parsed.hostname:
|
|
cmd[1:1] = ["-h", parsed.hostname]
|
|
if parsed.port:
|
|
cmd[1:1] = ["-p", str(parsed.port)]
|
|
if parsed.username:
|
|
cmd[1:1] = ["-U", parsed.username]
|
|
|
|
proc = subprocess.run(cmd, capture_output=True, env={**os.environ, **env})
|
|
if proc.returncode != 0:
|
|
err = (proc.stderr or b"").decode("utf-8", "replace").strip()
|
|
raise RuntimeError(err or "pg_dump fehlgeschlagen.")
|
|
|
|
skip_dirs = {"toolpath", "backup"}
|
|
with zipfile.ZipFile(dest, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("database.sql", proc.stdout)
|
|
if data_dir.is_dir():
|
|
for path in sorted(data_dir.rglob("*")):
|
|
if not path.is_file():
|
|
continue
|
|
rel = path.relative_to(data_dir)
|
|
if rel.parts and rel.parts[0] in skip_dirs:
|
|
continue
|
|
zf.write(path, "data/" + rel.as_posix())
|
|
return {
|
|
"name": filename,
|
|
"path": str(dest),
|
|
"size": dest.stat().st_size,
|
|
}
|
|
|
|
|
|
def list_backups(cfg):
|
|
out_dir = backup_dir(cfg)
|
|
items = []
|
|
for path in sorted(out_dir.glob("*.zip"), reverse=True):
|
|
try:
|
|
size = path.stat().st_size
|
|
except OSError:
|
|
size = 0
|
|
items.append({"name": path.name, "size": size})
|
|
return items
|
|
|
|
|
|
def backup_path(cfg, name):
|
|
out_dir = backup_dir(cfg)
|
|
safe = _safe_name(name)
|
|
path = (out_dir / safe).resolve()
|
|
if not str(path).startswith(str(out_dir.resolve())):
|
|
return None
|
|
if not path.is_file():
|
|
return None
|
|
return path
|