plugin-helper/src/plugin_helper/userdata.py

from __future__ import annotations

import json
import fnmatch
import shutil
import tarfile
from datetime import datetime, timezone
from pathlib import Path
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():
        raise FileNotFoundError(f"UserData directory not found: {source}")

    created_at = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    destination = backups_dir(state_root, instance) / f"userdata-{created_at}.tar.gz"
    files: list[dict[str, Any]] = []
    total_size = 0
    for path in sorted(item for item in source.rglob("*") if item.is_file()):
        rel = path.relative_to(instance_path).as_posix()
        size = path.stat().st_size
        total_size += size
        files.append({"path": rel, "size": size, "sha256": sha256_file(path)})

    manifest = {
        "createdAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        "instance": instance,
        "source": str(source),
        "fileCount": len(files),
        "totalSize": total_size,
        "files": files,
    }

    destination.parent.mkdir(parents=True, exist_ok=True)
    manifest_path = destination.parent / f".{destination.name}.manifest.json"
    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    with tarfile.open(destination, "w:gz") as archive:
        archive.add(source, arcname="UserData")
        archive.add(manifest_path, arcname="manifest.json")
    manifest_path.unlink(missing_ok=True)
    return {"archive": str(destination), "manifest": manifest}


def infer_windows_appdata_path(instance_path: Path) -> Path:
    parts = instance_path.resolve().parts
    try:
        bsmanager_index = parts.index("BSManager")
    except ValueError as exc:
        raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}") from exc
    if bsmanager_index < 2 or parts[bsmanager_index - 2] != "Users":
        raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}")
    profile = Path(*parts[:bsmanager_index])
    return profile / "AppData" / "LocalLow" / "Hyperbolic Magnetism" / "Beat Saber"


def infer_proton_appdata_path() -> Path:
    return (
        Path.home()
        / ".local/share/BSManager/SharedContent/compatdata/pfx/drive_c/users/steamuser/AppData/LocalLow/Hyperbolic Magnetism/Beat Saber"
    )


def infer_appdata_path(instance_path: Path) -> Path:
    if "Users" in instance_path.resolve().parts:
        return infer_windows_appdata_path(instance_path)
    return infer_proton_appdata_path()


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_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 restore_windows_data_repo(
    *,
    instance: str,
    instance_path: Path,
    backup_root: Path,
    appdata_path: Path | None = None,
    include_appdata: bool = True,
) -> dict[str, Any]:
    descriptor_path = backup_root / "backup-descriptor.json"
    if descriptor_path.is_file():
        descriptor = json.loads(descriptor_path.read_text(encoding="utf-8"))
        descriptor_instance = descriptor.get("instance")
        if descriptor_instance and descriptor_instance != instance:
            raise ValueError(
                f"backup descriptor instance {descriptor_instance!r} does not match requested instance {instance!r}"
            )

    created_at = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    restores: list[tuple[str, Path, Path]] = [
        ("UserData", backup_root / "UserData", instance_path / "UserData"),
    ]
    if include_appdata:
        appdata = appdata_path or infer_appdata_path(instance_path)
        restores.append(("AppData", backup_root / "AppData", appdata))

    for label, source, _ in restores:
        if not source.is_dir():
            raise FileNotFoundError(f"{label} backup not found: {source}")

    restored: list[dict[str, Any]] = []
    snapshots: list[dict[str, Any]] = []
    for label, source, destination in restores:
        snapshot: Path | None = None
        if destination.exists():
            snapshot = destination.parent / f"{destination.name}.pre-restore-{created_at}"
            destination.rename(snapshot)
            snapshots.append({"label": label, "path": str(snapshot)})

        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, destination, symlinks=True)
        restored.append(
            {
                "label": label,
                "source": str(source),
                "destination": str(destination),
                "fileCount": sum(1 for item in destination.rglob("*") if item.is_file()),
                "snapshot": str(snapshot) if snapshot else None,
            }
        )

    return {
        "backupRoot": str(backup_root),
        "restored": restored,
        "snapshots": snapshots,
    }


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)