Add backup recovery center
This commit is contained in:
@@ -34,6 +34,20 @@ class RepairReceipt:
|
||||
replaced: Tuple[Path, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BackupSummary:
|
||||
path: Path
|
||||
created_at: str
|
||||
keeper: Path
|
||||
replaced: Tuple[Path, ...]
|
||||
size_bytes: int
|
||||
restored_at: Optional[str]
|
||||
|
||||
@property
|
||||
def status(self) -> str:
|
||||
return "restored" if self.restored_at else "ready"
|
||||
|
||||
|
||||
def plan_duplicate_repair(
|
||||
keeper: Path, duplicates: Iterable[Path], serato_root: Path
|
||||
) -> DuplicateRepairPlan:
|
||||
@@ -124,9 +138,47 @@ def restore_backup(backup: Path) -> Tuple[Path, ...]:
|
||||
original.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(saved, original)
|
||||
restored.append(original)
|
||||
manifest["restored_at"] = datetime.now(timezone.utc).isoformat()
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
|
||||
return tuple(restored)
|
||||
|
||||
|
||||
def list_backups(serato_root: Path) -> Tuple[BackupSummary, ...]:
|
||||
backup_root = serato_root.expanduser().resolve() / BACKUP_FOLDER
|
||||
if not backup_root.is_dir():
|
||||
return ()
|
||||
summaries = []
|
||||
for backup in backup_root.iterdir():
|
||||
manifest_path = backup / "manifest.json"
|
||||
if not manifest_path.is_file():
|
||||
continue
|
||||
try:
|
||||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||||
replaced = tuple(
|
||||
Path(entry["original"]) for entry in manifest["replaced"]
|
||||
)
|
||||
size = sum(
|
||||
path.stat().st_size
|
||||
for path in backup.rglob("*")
|
||||
if path.is_file()
|
||||
)
|
||||
summaries.append(
|
||||
BackupSummary(
|
||||
path=backup.resolve(),
|
||||
created_at=manifest["created_at"],
|
||||
keeper=Path(manifest["keeper"]),
|
||||
replaced=replaced,
|
||||
size_bytes=size,
|
||||
restored_at=manifest.get("restored_at"),
|
||||
)
|
||||
)
|
||||
except (KeyError, OSError, TypeError, json.JSONDecodeError):
|
||||
continue
|
||||
return tuple(
|
||||
sorted(summaries, key=lambda item: item.created_at, reverse=True)
|
||||
)
|
||||
|
||||
|
||||
def rotate_backups(backup_root: Path, limit: Optional[int]) -> None:
|
||||
if limit is None or not backup_root.is_dir():
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user