lintunes/lintunes/trash.py

"""Move files to the desktop trash, per the freedesktop.org Trash spec 1.0.

Music usually lives on whatever volume the user mounted it from, not on the home
filesystem — so the trash that matters is the one at the top of the *file's own*
mount (``<topdir>/.Trash-<uid>``), not ``~/.local/share/Trash``. Getting that
wrong would mean a slow cross-device copy and, worse, a trashed file the file
manager's "Restore" can't put back, because the recorded original path would sit
on a different filesystem than the trash holding it.

Nothing here knows about the library; ``LibraryManager`` is the only caller.
"""

import logging
import os
import stat
from datetime import datetime
from pathlib import Path
from urllib.parse import quote, unquote

log = logging.getLogger(__name__)


class TrashError(OSError):
    """Raised when a file could not be trashed. The file is left where it is."""


def _home_trash() -> Path:
    base = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share")
    return Path(base) / "Trash"


def _topdir(path: Path) -> Path:
    """The mount point the path lives on."""
    current = path if path.is_dir() else path.parent
    current = current.resolve()
    while not os.path.ismount(current) and current != current.parent:
        current = current.parent
    return current


def trash_dir_for(path: Path) -> tuple[Path, Path | None]:
    """Where ``path`` should be trashed.

    Returns ``(trash_dir, topdir)``. ``topdir`` is None for the home trash — the
    spec wants an absolute ``Path=`` there and a topdir-relative one in a volume
    trash, so the caller needs to know which it got.
    """
    home_trash = _home_trash()
    top = _topdir(path)

    # Same filesystem as the home trash? Then that's the right one, whether or
    # not it already exists. Compare devices rather than paths: /home is often
    # its own mount, which would otherwise look like a "volume".
    try:
        home_dev = os.stat(_first_existing(home_trash)).st_dev
        if os.stat(path).st_dev == home_dev:
            return home_trash, None
    except OSError:  # no readable home — fall through to the volume trash
        pass

    # Spec: an admin-provided <topdir>/.Trash must be a sticky, non-symlink dir,
    # and we get our own uid-named subdir inside it. Otherwise we make our own.
    uid = os.getuid()
    admin = top / ".Trash"
    try:
        st = os.lstat(admin)
        if stat.S_ISDIR(st.st_mode) and (st.st_mode & stat.S_ISVTX):
            return admin / str(uid), top
    except OSError:
        pass
    return top / f".Trash-{uid}", top


def _first_existing(path: Path) -> Path:
    """Nearest existing ancestor (inclusive) — for stat'ing a dir we may create."""
    current = path
    while not current.exists() and current != current.parent:
        current = current.parent
    return current


def _ensure_trash(trash_dir: Path) -> tuple[Path, Path]:
    files_dir, info_dir = trash_dir / "files", trash_dir / "info"
    try:
        for directory in (trash_dir, files_dir, info_dir):
            directory.mkdir(mode=0o700, parents=True, exist_ok=True)
    except OSError as e:
        raise TrashError(f"Could not prepare the trash at {trash_dir}: {e}") from e
    return files_dir, info_dir


def _claim(info_dir: Path, name: str) -> tuple[Path, str]:
    """Reserve a name in the trash by creating its .trashinfo exclusively.

    Creating the info file is what makes the claim atomic against another app
    trashing a same-named file at the same moment; the audio file is moved in
    afterwards, under the name we just won.
    """
    stem, suffix = Path(name).stem, Path(name).suffix
    for i in range(0, 1000):
        candidate = name if i == 0 else f"{stem} {i}{suffix}"
        info_path = info_dir / f"{candidate}.trashinfo"
        try:
            fd = os.open(info_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except FileExistsError:
            continue
        except OSError as e:
            raise TrashError(f"Could not write {info_path}: {e}") from e
        os.close(fd)
        return info_path, candidate
    raise TrashError(f"Could not find a free name in {info_dir} for {name}")


def send_to_trash(path: Path) -> Path:
    """Move ``path`` into the trash for its filesystem; return its new location.

    Raises ``TrashError`` without touching the file if anything goes wrong, so a
    caller can treat a failure as "this was not deleted".
    """
    path = Path(path)
    if not path.exists():
        raise TrashError(f"{path} does not exist")

    trash_dir, top = trash_dir_for(path)
    files_dir, info_dir = _ensure_trash(trash_dir)
    info_path, name = _claim(info_dir, path.name)

    # Path= is relative to the topdir in a volume trash, absolute in the home
    # one, and percent-encoded either way.
    original = path.resolve()
    recorded = original.relative_to(top) if top is not None else original
    info_path.write_text(
        "[Trash Info]\n"
        f"Path={quote(str(recorded))}\n"
        f"DeletionDate={datetime.now().strftime('%Y-%m-%dT%H:%M:%S')}\n",
        encoding="utf-8")

    dest = files_dir / name
    try:
        os.rename(original, dest)
    except OSError as e:
        info_path.unlink(missing_ok=True)  # never leave a half-trashed pair
        raise TrashError(f"Could not move {path} to the trash: {e}") from e
    log.info("Trashed %s -> %s", original, dest)
    return dest


def restore_from_trash(trashed: Path, original: Path) -> None:
    """Move a trashed file back and drop its .trashinfo. Raises TrashError."""
    trashed, original = Path(trashed), Path(original)
    if not trashed.exists():
        raise TrashError(f"{trashed} is no longer in the trash")
    try:
        original.parent.mkdir(parents=True, exist_ok=True)
        os.rename(trashed, original)
    except OSError as e:
        raise TrashError(f"Could not restore {trashed} to {original}: {e}") from e
    info_path = trashed.parent.parent / "info" / f"{trashed.name}.trashinfo"
    info_path.unlink(missing_ok=True)


def original_path(trashed: Path) -> Path | None:
    """The path a trashed file came from, per its .trashinfo — None if unreadable.

    Only used by tests and diagnostics; restoring uses the location the library
    already knows.
    """
    info_path = Path(trashed).parent.parent / "info" / f"{Path(trashed).name}.trashinfo"
    try:
        text = info_path.read_text(encoding="utf-8")
    except OSError:
        return None
    for line in text.splitlines():
        if line.startswith("Path="):
            recorded = Path(unquote(line[len("Path="):]))
            if recorded.is_absolute():
                return recorded
            return _topdir(Path(trashed)) / recorded
    return None