From 3e5d58ea256e8ed3bd84c29fe6666b428df1e6e6 Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Sat, 5 Sep 2026 09:29:26 +0200 Subject: [PATCH] feat(backup-to-external): Add hardlink-aware backup tool --- backup_to_external.py | 319 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 319 insertions(+) create mode 100755 backup_to_external.py diff --git a/backup_to_external.py b/backup_to_external.py new file mode 100755 index 0000000..06145dc --- /dev/null +++ b/backup_to_external.py @@ -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...] stage a file for backup + backup_tool.py unmark [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...]") + 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...]") + 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()