from __future__ import annotations import json import shutil import zipfile from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any @dataclass(slots=True) class BackupRecordData: id: str label: str archive_path: Path manifest: dict[str, object] created_at: datetime class BackupService: def __init__(self, data_dir: Path, backup_dir: Path) -> None: self.data_dir = data_dir self.backup_dir = backup_dir self.backup_dir.mkdir(parents=True, exist_ok=True) def create_backup(self, label: str | None = None) -> BackupRecordData: created_at = datetime.now(UTC) backup_id = created_at.strftime("%Y%m%d%H%M%S") safe_label = label or f"backup-{backup_id}" archive_path = self.backup_dir / f"{safe_label}.zip" manifest: dict[str, Any] = { "id": backup_id, "label": safe_label, "created_at": created_at.isoformat(), "format_version": 1, "included_paths": [], } with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: for file_path in sorted(self.data_dir.rglob("*")): if not file_path.is_file(): continue if self.backup_dir in file_path.parents: continue relative = file_path.relative_to(self.data_dir) archive.write(file_path, arcname=f"data/{relative.as_posix()}") manifest["included_paths"].append(f"data/{relative.as_posix()}") archive.writestr("manifest.json", json.dumps(manifest, indent=2, ensure_ascii=True)) return BackupRecordData( id=backup_id, label=safe_label, archive_path=archive_path, manifest=manifest, created_at=created_at, ) def list_backups(self) -> list[BackupRecordData]: backups: list[BackupRecordData] = [] for archive_path in sorted(self.backup_dir.glob("*.zip"), reverse=True): try: with zipfile.ZipFile(archive_path) as archive: manifest = json.loads(archive.read("manifest.json").decode("utf-8")) except (KeyError, zipfile.BadZipFile, json.JSONDecodeError): continue backups.append( BackupRecordData( id=str(manifest["id"]), label=str(manifest["label"]), archive_path=archive_path, manifest=manifest, created_at=datetime.fromisoformat(str(manifest["created_at"])), ) ) return backups def restore_backup(self, backup_id: str) -> BackupRecordData: target = next((item for item in self.list_backups() if item.id == backup_id), None) if target is None: raise FileNotFoundError(f"Backup {backup_id} ikke fundet") self.create_backup(label=f"pre-restore-{backup_id}") temp_dir = self.backup_dir / f"restore-{backup_id}" if temp_dir.exists(): shutil.rmtree(temp_dir) temp_dir.mkdir(parents=True, exist_ok=True) try: with zipfile.ZipFile(target.archive_path) as archive: archive.extractall(temp_dir) restored_data_dir = temp_dir / "data" self._clear_runtime_files() if restored_data_dir.exists(): for file_path in sorted(restored_data_dir.rglob("*")): if not file_path.is_file(): continue relative = file_path.relative_to(restored_data_dir) destination = self.data_dir / relative destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(file_path, destination) finally: shutil.rmtree(temp_dir, ignore_errors=True) return target def _clear_runtime_files(self) -> None: for file_path in sorted(self.data_dir.rglob("*"), reverse=True): if self.backup_dir == file_path or self.backup_dir in file_path.parents: continue if file_path.is_file(): file_path.unlink() elif file_path.is_dir(): file_path.rmdir() def delete_backup(self, backup_id: str) -> None: target = next((item for item in self.list_backups() if item.id == backup_id), None) if target is None: raise FileNotFoundError(f"Backup {backup_id} ikke fundet") target.archive_path.unlink(missing_ok=True)