playground/backup_to_external.py

#!/usr/bin/env python3
"""Mark files for backup, then copy them to an external drive and remove
all their hardlinked siblings once the copy is verified.

Usage:
    backup_tool.py mark <file> [file...]     stage a file for backup
    backup_tool.py unmark <file> [file...]   drop a file from staging
    backup_tool.py list                      show what's staged
    backup_tool.py run [--yes]               copy staged files to the drive,
                                              then delete originals + siblings

Config (override via environment):
    BACKUP_STAGING_DIR   default: ~/.backup-staging

Destination routing is controlled by the RULES table below — edit it to
change which drive/subdirectory a file backs up to based on its source path
and type.
"""

import argparse
import hashlib
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path

STAGING_DIR = Path(
    os.environ.get("BACKUP_STAGING_DIR", str(Path.home() / ".backup-staging"))
)

VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".m4v", ".webm"}

# Roots under which an external drive's mount point may appear. Native
# drives (HFS+/APFS) mount under /Volumes; Mounty mounts NTFS drives
# read-write under ~/.mounty instead.
MOUNT_ROOTS = [Path("/Volumes"), Path.home() / ".mounty"]


def is_video(path: Path) -> bool:
    return path.suffix.lower() in VIDEO_EXTS


@dataclass
class Rule:
    prefix: Path  # source files under this path are matched
    volume: str  # drive name as it appears under a mount root, e.g. "FIVER"
    subdir: str  # path under the volume root to copy into, e.g. "_ingress"
    match_file: Callable[[Path], bool] | None = field(
        default=None
    )  # None matches any file

    def candidate_mount_points(self) -> list[Path]:
        """Where this volume might be mounted: a regular /Volumes mount, or
        a Mounty read-write remount under ~/.mounty."""
        return [root / self.volume for root in MOUNT_ROOTS]

    def mounted_at(self) -> Path | None:
        """Whichever candidate mount point is actually mounted right now, or None."""
        for mount_point in self.candidate_mount_points():
            if is_mounted(mount_point):
                return mount_point
        return None

    def matches(self, path: Path) -> bool:
        if not path.is_relative_to(self.prefix):
            return False
        if self.match_file is not None and not self.match_file(path):
            return False
        return True

    def destination_for(self, path: Path, mount_point: Path) -> Path:
        rel = path.relative_to(self.prefix)
        return mount_point / self.subdir / rel


# Order matters: first matching rule wins.
RULES = [
    Rule(
        prefix=Path.home() / "Downloads/temp",
        volume="FIVER",
        subdir="_ingress",
        match_file=is_video,
    ),
    Rule(
        prefix=Path.home() / "Movies",
        volume="FAST",
        subdir="_movies",
    ),
]


def rule_for(path: Path) -> Rule | None:
    for rule in RULES:
        if rule.matches(path):
            return rule
    return None


def die(msg: str) -> None:
    print(f"error: {msg}", file=sys.stderr)
    sys.exit(1)


def staged_path_for(abs_path: Path) -> Path:
    return STAGING_DIR / abs_path.relative_to(abs_path.anchor)


def original_path_for(staged_path: Path) -> Path:
    return Path("/") / staged_path.relative_to(STAGING_DIR)


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def mountpoint_of(path: Path) -> str:
    out = subprocess.run(
        ["df", str(path)], capture_output=True, text=True, check=True
    ).stdout
    return out.strip().splitlines()[-1].split()[-1]


def mount_line(mount_point: Path) -> str | None:
    out = subprocess.run(["mount"], capture_output=True, text=True, check=True).stdout
    for line in out.splitlines():
        if f" on {mount_point} " in line:
            return line
    return None


def is_mounted(mount_point: Path) -> bool:
    return mount_line(mount_point) is not None


def is_read_only(mount_point: Path) -> bool:
    """True if mounted read-only — the state NTFS drives auto-mount into
    before Mounty remounts them read-write via ntfs-3g."""
    line = mount_line(mount_point)
    return line is not None and "read-only" in line


def find_siblings(path: Path) -> list[Path]:
    """All paths sharing this file's inode, scoped to its own filesystem."""
    inode = path.stat().st_ino
    root = mountpoint_of(path)
    result = subprocess.run(
        ["find", root, "-xdev", "-inum", str(inode)],
        capture_output=True,
        text=True,
    )
    return [Path(p) for p in result.stdout.splitlines() if p]


def cmd_mark(args: argparse.Namespace) -> None:
    if not args.files:
        die("usage: mark <file> [file...]")
    for f in args.files:
        p = Path(f)
        if not p.is_file():
            print(f"skip (not a regular file): {f}", file=sys.stderr)
            continue
        abs_path = p.resolve()
        if rule_for(abs_path) is None:
            die(f"no backup rule matches '{abs_path}' — add one to RULES")
        dest = staged_path_for(abs_path)
        dest.parent.mkdir(parents=True, exist_ok=True)
        if dest.exists():
            print(f"already marked: {abs_path}")
            continue
        try:
            os.link(abs_path, dest)
        except OSError as e:
            die(
                f"could not hardlink '{abs_path}' into staging "
                f"(staging dir must be on the same filesystem as the file): {e}"
            )
        print(f"marked: {abs_path}")


def cmd_unmark(args: argparse.Namespace) -> None:
    if not args.files:
        die("usage: unmark <file> [file...]")
    for f in args.files:
        abs_path = Path(f).resolve()
        staged = staged_path_for(abs_path)
        if staged.exists():
            staged.unlink()
            print(f"unmarked: {abs_path}")
        else:
            print(f"not marked: {abs_path}", file=sys.stderr)


def cmd_list(_args: argparse.Namespace) -> None:
    if not STAGING_DIR.is_dir():
        print("(nothing staged)")
        return
    staged = sorted(p for p in STAGING_DIR.rglob("*") if p.is_file())
    if not staged:
        print("(nothing staged)")
        return
    for p in staged:
        print("/" + str(p.relative_to(STAGING_DIR)))


def cmd_run(args: argparse.Namespace) -> None:
    if not STAGING_DIR.is_dir():
        print("(nothing staged)")
        return
    staged = sorted(p for p in STAGING_DIR.rglob("*") if p.is_file())
    if not staged:
        print("(nothing staged)")
        return

    # Resolve each staged file's destination; skip (with a message) anything
    # whose target rule can no longer be matched or whose volume isn't mounted.
    plan = []
    for src in staged:
        original = original_path_for(src)
        rule = rule_for(original)
        if rule is None:
            print(f"skipping (no rule matches): {original}", file=sys.stderr)
            continue
        mount_point = rule.mounted_at()
        if mount_point is None:
            print(f"skipping (drive '{rule.volume}' not mounted): {original}")
            continue
        if is_read_only(mount_point):
            print(
                f"skipping (drive '{rule.volume}' is mounted read-only — open Mounty "
                f"and choose 'Remount' to enable write access): {original}"
            )
            continue
        dest = rule.destination_for(original, mount_point)
        plan.append((src, dest))

    if not plan:
        print("nothing to do — connect the relevant drive(s) and run again.")
        return

    print("The following staged files will be copied to their destinations,")
    print(
        "then ALL hardlinked copies of each (including the originals) will be deleted:"
    )
    for _src, dest in plan:
        print(f"  -> {dest}")
    print()

    if not args.yes:
        reply = input("Proceed? [y/N] ").strip().lower()
        if reply not in ("y", "yes"):
            print("aborted.")
            sys.exit(1)

    for src, dest in plan:
        dest.parent.mkdir(parents=True, exist_ok=True)

        print(f"copying: {original_path_for(src)} -> {dest}")
        shutil.copy2(src, dest)

        if sha256(src) != sha256(dest):
            print(
                f"  verification FAILED, leaving originals in place: {original_path_for(src)}",
                file=sys.stderr,
            )
            continue

        siblings = find_siblings(src)
        print(f"  verified. removing {len(siblings)} hardlinked path(s):")
        for sib in siblings:
            print(f"    {sib}")
        for sib in siblings:
            try:
                sib.unlink()
            except OSError as e:
                print(f"    failed to remove {sib}: {e}", file=sys.stderr)

    # Clean up now-empty staging directories.
    for dirpath, dirnames, filenames in os.walk(STAGING_DIR, topdown=False):
        d = Path(dirpath)
        if d != STAGING_DIR and not any(d.iterdir()):
            d.rmdir()

    print("done.")


def main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    sub = parser.add_subparsers(dest="command", required=True)

    p_mark = sub.add_parser("mark")
    p_mark.add_argument("files", nargs="+")
    p_mark.set_defaults(func=cmd_mark)

    p_unmark = sub.add_parser("unmark")
    p_unmark.add_argument("files", nargs="+")
    p_unmark.set_defaults(func=cmd_unmark)

    p_list = sub.add_parser("list")
    p_list.set_defaults(func=cmd_list)

    p_run = sub.add_parser("run")
    p_run.add_argument("--yes", action="store_true", help="skip confirmation prompt")
    p_run.set_defaults(func=cmd_run)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()