Add Beat Saber data backup command
This commit is contained in:
@@ -1,16 +1,42 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import fnmatch
|
||||
import shutil
|
||||
import tarfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from tempfile import NamedTemporaryFile
|
||||
from typing import Any
|
||||
from typing import Any, Callable
|
||||
|
||||
from .fsutil import sha256_file
|
||||
from .state import backups_dir
|
||||
|
||||
|
||||
DEFAULT_BACKUP_EXCLUDES = (
|
||||
"BeatLeader/Replays",
|
||||
"BeatLeader/Replays/**",
|
||||
"BeatLeader/ReplayerCache",
|
||||
"BeatLeader/ReplayerCache/**",
|
||||
"BeatLeader/LeaderboardsCache",
|
||||
"BeatLeader/LeaderboardsCache/**",
|
||||
"BeatLeader/ReplayHeadersCache",
|
||||
"ScoreSaber/Replays",
|
||||
"ScoreSaber/Replays/**",
|
||||
"BeatSaberPlus/Cache",
|
||||
"BeatSaberPlus/Cache/**",
|
||||
"BeatSaverNotifier.json",
|
||||
"Accsaber/PlayerScoreCache.json",
|
||||
"NalulunaAvatars/cache",
|
||||
"NalulunaAvatars/cache/**",
|
||||
"SongDetailsCache.proto",
|
||||
"com.unity.addressables",
|
||||
"com.unity.addressables/**",
|
||||
"*.log",
|
||||
"*.log.*",
|
||||
)
|
||||
|
||||
|
||||
def backup_userdata(instance: str, instance_path: Path, state_root: Path) -> dict[str, Any]:
|
||||
source = instance_path / "UserData"
|
||||
if not source.is_dir():
|
||||
@@ -44,3 +70,94 @@ def backup_userdata(instance: str, instance_path: Path, state_root: Path) -> dic
|
||||
handle.flush()
|
||||
archive.add(handle.name, arcname="manifest.json")
|
||||
return {"archive": str(destination), "manifest": manifest}
|
||||
|
||||
|
||||
def infer_windows_appdata_path(instance_path: Path) -> Path:
|
||||
parts = instance_path.resolve().parts
|
||||
try:
|
||||
users_index = parts.index("Users")
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}") from exc
|
||||
if users_index + 1 >= len(parts):
|
||||
raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}")
|
||||
profile = Path(*parts[: users_index + 2])
|
||||
return profile / "AppData" / "LocalLow" / "Hyperbolic Magnetism" / "Beat Saber"
|
||||
|
||||
|
||||
def sync_windows_data_repo(
|
||||
*,
|
||||
instance: str,
|
||||
instance_path: Path,
|
||||
backup_root: Path,
|
||||
appdata_path: Path | None = None,
|
||||
include_appdata: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
sources: list[tuple[str, Path, Path]] = [
|
||||
("UserData", instance_path / "UserData", backup_root / "UserData"),
|
||||
]
|
||||
if include_appdata:
|
||||
appdata = appdata_path or infer_windows_appdata_path(instance_path)
|
||||
sources.append(("AppData", appdata, backup_root / "AppData"))
|
||||
|
||||
for label, source, _ in sources:
|
||||
if not source.is_dir():
|
||||
raise FileNotFoundError(f"{label} directory not found: {source}")
|
||||
|
||||
backup_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
copied: list[dict[str, Any]] = []
|
||||
skipped: list[str] = []
|
||||
for label, source, destination in sources:
|
||||
if destination.exists():
|
||||
shutil.rmtree(destination)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copytree(
|
||||
source,
|
||||
destination,
|
||||
symlinks=True,
|
||||
ignore=_ignore_backup_paths(source, skipped),
|
||||
)
|
||||
copied.append(
|
||||
{
|
||||
"label": label,
|
||||
"source": str(source),
|
||||
"destination": str(destination),
|
||||
"fileCount": sum(1 for item in destination.rglob("*") if item.is_file()),
|
||||
}
|
||||
)
|
||||
|
||||
manifest = {
|
||||
"createdAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
||||
"instance": instance,
|
||||
"sources": copied,
|
||||
"excludePatterns": list(DEFAULT_BACKUP_EXCLUDES),
|
||||
"skipped": sorted(set(skipped)),
|
||||
}
|
||||
(backup_root / "backup-descriptor.json").write_text(
|
||||
json.dumps(manifest, indent=2, sort_keys=True) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
return {
|
||||
"backupRoot": str(backup_root),
|
||||
"copied": copied,
|
||||
"manifest": manifest,
|
||||
}
|
||||
|
||||
|
||||
def _ignore_backup_paths(source_root: Path, skipped: list[str]) -> Callable[[str, list[str]], set[str]]:
|
||||
def ignore(current_dir: str, names: list[str]) -> set[str]:
|
||||
ignored: set[str] = set()
|
||||
current_path = Path(current_dir)
|
||||
for name in names:
|
||||
relative = (current_path / name).relative_to(source_root).as_posix()
|
||||
if _is_excluded(relative):
|
||||
ignored.add(name)
|
||||
skipped.append(relative)
|
||||
return ignored
|
||||
|
||||
return ignore
|
||||
|
||||
|
||||
def _is_excluded(relative_path: str) -> bool:
|
||||
return any(fnmatch.fnmatchcase(relative_path, pattern) for pattern in DEFAULT_BACKUP_EXCLUDES)
|
||||
|
||||
Reference in New Issue
Block a user