feat(backup-to-external): Add hardlink-aware backup tool
This commit is contained in:
Executable
+319
@@ -0,0 +1,319 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user