serato-doctor/serato_doctor/repair.py

"""Backup-first duplicate consolidation.

Serato's binary metadata is deliberately not rewritten.  Removed duplicate files
are replaced with symbolic links, so every existing path continues to resolve.
"""

import json
import shutil
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Optional, Tuple
from uuid import uuid4


BACKUP_FOLDER = ".serato-doctor-backups"


@dataclass(frozen=True)
class DuplicateRepairPlan:
    keeper: Path
    replaced: Tuple[Path, ...]
    metadata_files: Tuple[Path, ...]

    @property
    def changes(self) -> int:
        return len(self.replaced)


@dataclass(frozen=True)
class RepairReceipt:
    backup: Path
    keeper: Path
    replaced: Tuple[Path, ...]


def plan_duplicate_repair(
    keeper: Path, duplicates: Iterable[Path], serato_root: Path
) -> DuplicateRepairPlan:
    keeper = keeper.expanduser().resolve()
    candidates = tuple(path.expanduser().resolve() for path in duplicates)
    if keeper not in candidates:
        raise ValueError("The file to keep must belong to this duplicate group")
    if not keeper.is_file():
        raise ValueError(f"The file to keep no longer exists: {keeper}")
    replaced = tuple(path for path in candidates if path != keeper)
    if not replaced:
        raise ValueError("Choose a duplicate group containing at least two files")
    if any(not path.is_file() or path.is_symlink() for path in replaced):
        raise ValueError("A duplicate changed since the analysis; analyze again")
    metadata = tuple(
        sorted(
            path
            for path in serato_root.expanduser().resolve().rglob("*")
            if path.is_file()
            and BACKUP_FOLDER not in path.parts
            and (
                path.name == "database V2"
                or path.suffix.casefold() in {".crate", ".scrate"}
            )
        )
    )
    return DuplicateRepairPlan(keeper, replaced, metadata)


def apply_duplicate_repair(
    plan: DuplicateRepairPlan,
    serato_root: Path,
    backup_limit: Optional[int] = 10,
) -> RepairReceipt:
    """Create a complete rollback snapshot, then replace extras with symlinks."""
    if backup_limit is not None and backup_limit < 1:
        raise ValueError("Backup limit must be at least 1, or unlimited")
    backup_root = serato_root.expanduser().resolve() / BACKUP_FOLDER
    backup = backup_root / _backup_name()
    files_root = backup / "files"
    metadata_root = backup / "serato-metadata"
    backup.mkdir(parents=True)

    entries = []
    try:
        for path in plan.replaced:
            destination = files_root / _safe_backup_path(path)
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(path, destination)
            entries.append({"original": str(path), "backup": str(destination)})
        for path in plan.metadata_files:
            relative = path.relative_to(serato_root.expanduser().resolve())
            destination = metadata_root / relative
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(path, destination)
        manifest = {
            "created_at": datetime.now(timezone.utc).isoformat(),
            "keeper": str(plan.keeper),
            "replaced": entries,
            "strategy": "symlink",
        }
        (backup / "manifest.json").write_text(
            json.dumps(manifest, indent=2), encoding="utf-8"
        )
        for path in plan.replaced:
            path.unlink()
            path.symlink_to(plan.keeper)
    except Exception:
        _rollback_entries(entries)
        shutil.rmtree(backup, ignore_errors=True)
        raise

    rotate_backups(backup_root, backup_limit)
    return RepairReceipt(backup, plan.keeper, plan.replaced)


def restore_backup(backup: Path) -> Tuple[Path, ...]:
    manifest_path = backup.expanduser().resolve() / "manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    restored = []
    for entry in manifest["replaced"]:
        original = Path(entry["original"])
        saved = Path(entry["backup"])
        if original.exists() and not original.is_symlink():
            raise ValueError(f"Restore would overwrite a real file: {original}")
        if original.is_symlink():
            original.unlink()
        original.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(saved, original)
        restored.append(original)
    return tuple(restored)


def rotate_backups(backup_root: Path, limit: Optional[int]) -> None:
    if limit is None or not backup_root.is_dir():
        return
    backups = sorted(
        (path for path in backup_root.iterdir() if (path / "manifest.json").is_file()),
        key=lambda path: path.name,
        reverse=True,
    )
    for expired in backups[limit:]:
        shutil.rmtree(expired)


def _backup_name() -> str:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return f"{stamp}-{uuid4().hex[:8]}"


def _safe_backup_path(path: Path) -> Path:
    anchorless = path.as_posix().lstrip("/").replace(":", "_")
    return Path(anchorless)


def _rollback_entries(entries: Iterable[dict]) -> None:
    for entry in entries:
        original = Path(entry["original"])
        saved = Path(entry["backup"])
        if original.is_symlink():
            original.unlink()
        if not original.exists() and saved.is_file():
            original.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(saved, original)