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, ...]


@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:
    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."""
    return apply_duplicate_repair_batch((plan,), serato_root, backup_limit)


def apply_duplicate_repair_batch(
    plans: Iterable[DuplicateRepairPlan],
    serato_root: Path,
    backup_limit: Optional[int] = 10,
) -> RepairReceipt:
    """Apply several approved duplicate choices as one atomic backup."""
    plans = tuple(plans)
    if not plans:
        raise ValueError("Choose at least one duplicate group")
    if backup_limit is not None and backup_limit < 1:
        raise ValueError("Backup limit must be at least 1, or unlimited")
    replaced_paths = tuple(
        path for plan in plans for path in plan.replaced
    )
    if len(set(replaced_paths)) != len(replaced_paths):
        raise ValueError("The same file appears in more than one repair choice")
    keepers = {plan.keeper for plan in plans}
    if keepers.intersection(replaced_paths):
        raise ValueError("A selected keeper cannot be removed by another choice")
    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 plan in plans:
            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),
                        "keeper": str(plan.keeper),
                    }
                )
        for path in plans[0].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(plans[0].keeper),
            "keepers": [str(plan.keeper) for plan in plans],
            "choice_count": len(plans),
            "replaced": entries,
            "strategy": "symlink",
        }
        (backup / "manifest.json").write_text(
            json.dumps(manifest, indent=2), encoding="utf-8"
        )
        for entry in entries:
            path = Path(entry["original"])
            path.unlink()
            path.symlink_to(Path(entry["keeper"]))
    except Exception:
        _rollback_entries(entries)
        shutil.rmtree(backup, ignore_errors=True)
        raise

    rotate_backups(backup_root, backup_limit)
    return RepairReceipt(backup, plans[0].keeper, replaced_paths)


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)
    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
    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)