#!/usr/bin/env python3 """StayPuller for Linux. Keeps your Storage Pod folder, "~/Desktop/Storage Pod", in step with your Storage Pod: the private GitHub repo "Storage-Pod", your Dropbox, or both, alongside the StayPuller Mac and Android apps. Needs only python3 (3.8+) and git (2.31+). staypuller install sign in + set up the background service (run this first) staypuller login sign in with your StayPuller account (staypuller.com) staypuller logout sign this computer out of StayPuller staypuller status show sync status and recent log staypuller sync run one sync pass now staypuller drive add keep your Storage Pod on a USB drive too (plug it in first); "drive list" shows them staypuller share FILE --to someone@example.com share a file from your Storage Pod (or any file) by email staypuller shared list what you share and what is shared with you; "unshare ID" stops one staypuller set-token use a GitHub token instead of a StayPuller account staypuller run run the sync loop in the foreground staypuller uninstall remove the service (keeps your files) """ import argparse import base64 import concurrent.futures import datetime import fnmatch import getpass import hashlib import json import os import re import shutil import socket import stat import subprocess import sys import time import threading import unicodedata import urllib.error import urllib.parse import urllib.request import uuid from pathlib import Path VERSION = "1.9" GITHUB_BASE = os.environ.get("STORAGE_POD_GITHUB_BASE", "https://github.com").rstrip("/") # tests point this at a local folder REPO_NAME = "Storage-Pod" # GitHub repo names can't contain spaces FOLDER_NAME = "Storage Pod" BRANCH = "main" MAX_FILE_SIZE = 95 * 1024 * 1024 # GitHub rejects files over 100 MB POLL_INTERVAL = 30 # seconds between checks for other machines' changes DEBOUNCE = 2 # seconds to wait after a local change before syncing SCAN_INTERVAL = 2 # seconds between local change scans HOME = Path.home() CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", HOME / ".config")) / "storage-pod" DATA_DIR = Path(os.environ.get("XDG_DATA_HOME", HOME / ".local/share")) / "storage-pod" STATE_DIR = Path(os.environ.get("XDG_STATE_HOME", HOME / ".local/state")) / "storage-pod" TOKEN_FILE = CONFIG_DIR / "token" ACCOUNT_FILE = CONFIG_DIR / "account.json" # StayPuller sign-in: server address and this device's token LOG_FILE = STATE_DIR / "storage-pod.log" STATUS_FILE = STATE_DIR / "status.json" GIT_DIR = DATA_DIR / "sync.git" BIN_PATH = HOME / ".local/bin/staypuller" LEGACY_BIN_PATH = HOME / ".local/bin/storage-pod" # the old command name, kept as a link UNIT_PATH = Path(os.environ.get("XDG_CONFIG_HOME", HOME / ".config")) / "systemd/user/storage-pod.service" EXCLUDE_PATTERNS = [ ".DS_Store", "._*", ".localized", "Icon\r\r", ".Trashes", ".fseventsd", ".Spotlight-V100", ".TemporaryItems", "*.icloud", "~$*", ".~lock.*#", ".directory", ".Trash-*", "*.swp", "*~", ".goutputstream-*", ".staypuller-tmp-*", ] def desktop_dir(): try: out = subprocess.run(["xdg-user-dir", "DESKTOP"], capture_output=True, text=True, timeout=5).stdout.strip() if out and Path(out) != HOME: return Path(out) except (OSError, subprocess.SubprocessError): pass return HOME / "Desktop" FOLDER = Path(os.environ["STORAGE_POD_FOLDER"]) if os.environ.get("STORAGE_POD_FOLDER") else desktop_dir() / FOLDER_NAME DEVICE_NAME = socket.gethostname().split(".")[0] def log(message): line = f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S} {message}" print(line, file=sys.stderr, flush=True) try: STATE_DIR.mkdir(parents=True, exist_ok=True) with open(LOG_FILE, "a") as f: f.write(line + "\n") except OSError: pass class GitError(Exception): pass class GitHubError(Exception): pass # ---------------------------------------------------------------- git class Git: """Runs git against a repo whose git dir lives outside the synced folder.""" def __init__(self, git_dir, work_tree): self.git_dir = Path(git_dir) self.work_tree = Path(work_tree) self.token = None def run(self, *args): env = dict(os.environ, GIT_DIR=str(self.git_dir), GIT_WORK_TREE=str(self.work_tree), GIT_TERMINAL_PROMPT="0", GIT_EDITOR="true") if self.token: # Passed via environment so the token never appears in process arguments. basic = base64.b64encode(f"x-access-token:{self.token}".encode()).decode() env.update(GIT_CONFIG_COUNT="1", GIT_CONFIG_KEY_0="http.https://github.com/.extraheader", GIT_CONFIG_VALUE_0=f"Authorization: Basic {basic}") return subprocess.run(["git", *args], cwd=self.work_tree, env=env, stdin=subprocess.DEVNULL, capture_output=True) def check(self, *args): r = self.run(*args) if r.returncode != 0: msg = r.stderr.decode(errors="replace").strip() or f"exit {r.returncode}" raise GitError(f"git {args[0]} failed: {msg}") return r def ok(self, *args): return self.run(*args).returncode == 0 def text(self, *args): return self.check(*args).stdout.decode(errors="replace").strip() def fields(self, *args): """NUL-separated output (from -z flags) as a list of strings.""" return [f.decode(errors="surrogateescape") for f in self.check(*args).stdout.split(b"\0") if f] def check_git_version(): try: out = subprocess.run(["git", "--version"], capture_output=True, text=True).stdout except OSError: sys.exit("git is not installed. Install it with your package manager (e.g. sudo apt install git).") parts = out.split()[2].split(".") if len(out.split()) > 2 else ["0", "0"] if (int(parts[0]), int(parts[1])) < (2, 31): sys.exit(f"StayPuller needs git 2.31 or newer (found {out.strip()}).") # ---------------------------------------------------------------- sync class Syncer: """One two-way sync pass: commit local changes, fetch, merge without losing data, push.""" def __init__(self, folder, git_dir, remote_url, device_name): self.folder = Path(folder) self.git = Git(git_dir, folder) self.remote_url = remote_url self.device_name = device_name self.skipped_files = [] self.events = [] def prepare(self): folder_was_missing = False if not self.folder.exists(): self.folder.mkdir(parents=True) folder_was_missing = True log(f"Created folder {self.folder}") g = self.git if not (g.git_dir / "HEAD").exists(): g.git_dir.mkdir(parents=True, exist_ok=True) g.check("init", "-q", "-b", BRANCH) email_host = self.device_name.replace(" ", "-") for k, v in [("core.bare", "false"), ("core.autocrlf", "false"), ("core.quotepath", "false"), ("merge.renames", "true"), ("user.name", self.device_name), ("user.email", f"storage-pod@{email_host}.local")]: g.check("config", k, v) g.check("remote", "add", "origin", self.remote_url) log(f"Initialized sync repo at {g.git_dir}") elif g.run("remote", "get-url", "origin").stdout.decode().strip() != self.remote_url: g.check("remote", "set-url", "origin", self.remote_url) info = g.git_dir / "info" info.mkdir(parents=True, exist_ok=True) (info / "exclude").write_text("\n".join(EXCLUDE_PATTERNS) + "\n") # If the whole folder vanished, restore it rather than deleting everything everywhere. if folder_was_missing and g.ok("rev-parse", "--verify", "-q", "HEAD"): g.check("reset", "-q", "--hard", "HEAD") log("Folder was missing; restored contents from last sync") def emit(self, kind, path, size=0): if len(self.events) < MAX_EVENTS: self.events.append((kind, path, int(size or 0))) def drain(self): out, self.events = self.events, [] return out def totals(self): """Files and bytes in the Storage Pod's latest commit, or None.""" try: out = self.git.fields("ls-tree", "-r", "-l", "-z", "HEAD") except GitError: return None sizes = [int(e.split("\t", 1)[0].split()[3]) for e in out if e.split("\t", 1)[0].split()[3].isdigit()] return len(sizes), sum(sizes) def _emit_diff(self, old, new, kind_for_add): """Events for what changed between two commits (what came from the other side, for example).""" try: out = self.git.fields("diff", "--name-status", "-z", old, new) except GitError: return i = 0 while i < len(out): code = out[i][:1] if code in ("R", "C"): path, i = out[i + 2], i + 3 else: path, i = out[i + 1], i + 2 if code == "D": self.emit("deleted", path) else: size = self.git.run("cat-file", "-s", f"{new}:{path}").stdout.decode().strip() self.emit(kind_for_add, path, int(size) if size.isdigit() else 0) def sync(self): self.events = [] self.prepare() self.commit_local_changes() g = self.git remote_ref = f"refs/remotes/origin/{BRANCH}" before_fetch = g.text("rev-parse", remote_ref) if g.ok("rev-parse", "--verify", "-q", remote_ref) else None g.check("fetch", "-q", "--prune", "origin") if before_fetch and g.ok("rev-parse", "--verify", "-q", remote_ref): after_fetch = g.text("rev-parse", remote_ref) if after_fetch != before_fetch: self._emit_diff(before_fetch, after_fetch, "downloaded") # what the other side changed elif g.ok("rev-parse", "--verify", "-q", remote_ref): for name in g.fields("ls-tree", "-r", "--name-only", "-z", remote_ref)[:MAX_EVENTS]: self.emit("downloaded", name) for attempt in range(1, 5): has_remote = g.ok("rev-parse", "--verify", "-q", remote_ref) has_local = g.ok("rev-parse", "--verify", "-q", "HEAD") if has_remote and not has_local: # Fresh machine with an empty folder: take whatever is on GitHub. g.check("reset", "-q", "--hard", remote_ref) log("Downloaded existing Storage Pod contents") return if has_remote and not g.ok("merge-base", "--is-ancestor", remote_ref, "HEAD"): self.merge(remote_ref) if not g.ok("rev-parse", "--verify", "-q", "HEAD"): return # nothing anywhere yet ahead = int(g.text("rev-list", "--count", f"{remote_ref}..HEAD")) if has_remote else 1 if ahead == 0: return push = g.run("push", "-q", "origin", f"HEAD:refs/heads/{BRANCH}") if push.returncode == 0: log(f"Pushed {ahead} commit(s)") return # Another machine pushed first; fetch its changes and try again. log(f"Push rejected (attempt {attempt}): {push.stderr.decode(errors='replace').strip()}") g.check("fetch", "-q", "--prune", "origin") raise GitHubError("Could not push after several attempts; will retry.") def commit_local_changes(self): skipped = self.find_large_files() if skipped and skipped != self.skipped_files: log(f"Skipping files over 95 MB: {', '.join(skipped)}") self.skipped_files = skipped g = self.git g.check("add", "-A", "--", ".", *[f":(exclude,literal){p}" for p in skipped]) if g.ok("diff", "--cached", "--quiet"): return changes = g.fields("diff", "--cached", "--name-status", "-z") counts = {"added": 0, "changed": 0, "removed": 0} i = 0 while i < len(changes): code = changes[i][:1] counts["added" if code == "A" else "removed" if code == "D" else "changed"] += 1 i += 3 if code in ("R", "C") else 2 summary = ", ".join(f"{n} {k}" for k, n in counts.items() if n) i = 0 while i < len(changes): # what this computer changed, on its way to the other side code = changes[i][:1] path = changes[i + 2] if code in ("R", "C") else changes[i + 1] if code == "D": self.emit("deleted", path) else: try: self.emit("uploaded", path, os.lstat(self.folder / path).st_size) except OSError: self.emit("uploaded", path) i += 3 if code in ("R", "C") else 2 g.check("commit", "-q", "--no-verify", "-m", f"{self.device_name}: {summary}") log(f"Committed local changes: {summary}") def find_large_files(self): large = [] for dirpath, _, filenames in os.walk(self.folder): for name in filenames: p = os.path.join(dirpath, name) try: st = os.lstat(p) except OSError: continue if st.st_size > MAX_FILE_SIZE and not os.path.islink(p): large.append(os.path.relpath(p, self.folder)) return sorted(large) def merge(self, remote_ref): """Merge remote changes. Edits on both sides keep both copies; edit-vs-delete keeps the file.""" g = self.git r = g.run("merge", "-q", "--no-edit", "--allow-unrelated-histories", "-m", f"{self.device_name}: merge", remote_ref) if r.returncode == 0: log("Merged remote changes") return conflicted = g.fields("diff", "--name-only", "--diff-filter=U", "-z") if not conflicted: g.run("merge", "--abort") raise GitError(f"git merge failed: {r.stderr.decode(errors='replace').strip()}") other = g.run("log", "-1", "--format=%an", remote_ref).stdout.decode().strip() or "another machine" stamp = f"{datetime.datetime.now():%Y-%m-%d %H%M}" for path in conflicted: stages = set() for entry in g.fields("ls-files", "-u", "-z", "--", path): meta = entry.split("\t", 1)[0].split(" ") # " \t" stages.add(int(meta[-1])) ours, theirs = 2 in stages, 3 in stages if ours and theirs: copy = self.conflict_copy_path(path, other, stamp) dest = self.folder / copy dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(g.check("show", f":3:{path}").stdout) g.check("checkout", "--ours", "--", path) g.check("add", "--", path, copy) log(f"Conflict on {path}: kept both, other version saved as {copy}") elif ours or theirs: g.check("checkout", "--ours" if ours else "--theirs", "--", path) g.check("add", "--", path) log(f"Conflict on {path}: edited on one machine, deleted on another; kept the file") else: g.check("rm", "-q", "--cached", "--ignore-unmatch", "--", path) g.check("commit", "-q", "--no-verify", "--no-edit") log(f"Merged remote changes with {len(conflicted)} conflict(s)") def conflict_copy_path(self, path, device, stamp): base, ext = os.path.splitext(path) if "/" in ext: # dot was in a directory name, not the file base, ext = path, "" device = device.replace("/", "-") candidate = f"{base} (conflict from {device} {stamp}){ext}" n = 2 while (self.folder / candidate).exists(): candidate = f"{base} (conflict from {device} {stamp} {n}){ext}" n += 1 return candidate # ---------------------------------------------------------------- GitHub def github(method, path, token, body=None): req = urllib.request.Request(f"https://api.github.com{path}", method=method) req.add_header("Authorization", f"Bearer {token}") req.add_header("Accept", "application/vnd.github+json") req.add_header("X-GitHub-Api-Version", "2022-11-28") data = None if body is not None: data = json.dumps(body).encode() req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, data=data, timeout=30) as resp: return resp.status, json.loads(resp.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {} def github_user(token): status, js = github("GET", "/user", token) if status == 401: raise GitHubError("GitHub rejected the token (401). Run: staypuller set-token") if status != 200: raise GitHubError(f"Could not read GitHub user (HTTP {status}).") return js["login"] def ensure_repo(owner, token): status, _ = github("GET", f"/repos/{owner}/{REPO_NAME}", token) if status == 200: return if status != 404: raise GitHubError(f"Could not check repo (HTTP {status}).") status, js = github("POST", "/user/repos", token, { "name": REPO_NAME, "private": True, "description": "Synced by the StayPuller app from ~/Desktop/Storage Pod", }) if status != 201: raise GitHubError(f"Could not create repo {REPO_NAME}: {js.get('message', status)}. " "The token needs access to create repositories, or create the repo on another machine first.") log(f"Created GitHub repo {owner}/{REPO_NAME}") # ---------------------------------------------------------------- token def read_token(): try: return TOKEN_FILE.read_text().strip() or None except OSError: return None def prompt_for_token(): print("\nStayPuller needs a GitHub token that can read and write your Storage-Pod repo.") print("Create a fine-grained token at https://github.com/settings/personal-access-tokens/new") print(" Repository access: Only select repositories > Storage-Pod") print(" Permissions: Contents = Read and write\n") while True: token = getpass.getpass("Paste token (input hidden): ").strip() if not token: sys.exit("No token entered.") try: login = github_user(token) ensure_repo(login, token) except (GitHubError, OSError) as e: print(f" {e}") continue CONFIG_DIR.mkdir(parents=True, exist_ok=True) fd = os.open(TOKEN_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w") as f: f.write(token + "\n") print(f" Connected as {login}. Token saved to {TOKEN_FILE} (readable only by you).") return # ---------------------------------------------------------------- StayPuller account SIGNED_OUT = "You're signed out of StayPuller. Run: staypuller login" def say(line): print(line, flush=True) # sign-in prompts must appear at once, even when output is piped def staypuller_url(): return os.environ.get("STAYPULLER_URL", "https://staypuller.com").rstrip("/") def staypuller_api(method, path, body=None, token=None, server=None): req = urllib.request.Request((server or staypuller_url()) + path, method=method) req.add_header("Accept", "application/json") if token: req.add_header("Authorization", f"Bearer {token}") data = None if body is not None: data = json.dumps(body).encode() req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, data=data, timeout=30) as resp: return resp.status, json.loads(resp.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {} def read_account(): try: acct = json.loads(ACCOUNT_FILE.read_text()) return acct if acct.get("device_token") and acct.get("server") else None except (OSError, ValueError): return None def write_account(acct): CONFIG_DIR.mkdir(parents=True, exist_ok=True) tmp = ACCOUNT_FILE.with_suffix(".tmp") fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w") as f: json.dump(acct, f) os.replace(tmp, ACCOUNT_FILE) def open_browser(url): if os.environ.get("STORAGE_POD_NO_BROWSER") or not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")): return opener = shutil.which("xdg-open") if opener: subprocess.Popen([opener, url], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def device_login(sleep=time.sleep, out=say): """Sign this computer in with a code approved on staypuller.com (works over SSH: open the link anywhere).""" server = staypuller_url() for _ in range(3): # a fresh code if one expires while waiting status, code = staypuller_api("POST", "/api/v1/device/code", {"name": DEVICE_NAME, "platform": "linux"}, server=server) if status != 200: raise GitHubError(f"StayPuller couldn't start the sign-in (HTTP {status}).") out(f"\nTo sign in, open {code['verification_uri']} on any device and enter this code:\n\n {code['user_code']}\n") out(f"Or open this link: {code['verification_uri_complete']}") open_browser(code["verification_uri_complete"]) interval, waited = code.get("interval", 5), 0 while waited < code.get("expires_in", 600): sleep(interval) waited += interval status, r = staypuller_api("POST", "/api/v1/device/token", {"device_code": code["device_code"]}, server=server) if status == 200 and r.get("access_token"): write_account({"server": server, "device_token": r["access_token"], "device_id": r.get("device_id", "")}) out("Signed in.") return read_account() err = r.get("error") if err == "authorization_pending": continue if err == "slow_down": interval += 5 continue if err == "access_denied": raise GitHubError("The sign-in was denied on staypuller.com.") if err == "expired_token": break raise GitHubError(f"StayPuller sign-in failed: {err or 'HTTP ' + str(status)}") out("That code expired. Getting a new one...") raise GitHubError("Sign-in timed out. Run: staypuller login") def wait_for_github(acct, sleep=time.sleep, out=say, limit=600): """After sign-in, wait for GitHub to be connected on the website (a one-time step).""" waited, told = 0, False while True: status, cfg = staypuller_api("GET", "/api/v1/config", token=acct["device_token"], server=acct["server"]) if status == 401: raise GitHubError(SIGNED_OUT) gh = (cfg.get("connections") or {}).get("github") or {} if status == 200 and gh.get("connected"): out(f"GitHub is connected: {gh.get('repo')}") return gh.get("repo") if not told: out(f"\nOne more step, once per account: connect GitHub at {acct['server']}/connections") out("Waiting for you to finish (Ctrl-C to skip; run 'staypuller sync' afterwards)...") told = True if waited >= limit: out("GitHub isn't connected yet. Connect it on the website, then run: staypuller sync") return None sleep(5) waited += 5 def wait_for_store(acct, sleep=time.sleep, out=say, limit=600): """After sign-in, wait for a store (GitHub or Dropbox) to be connected on the website: a one-time step per account.""" waited, told = 0, False while True: status, cfg = staypuller_api("GET", "/api/v1/config", token=acct["device_token"], server=acct["server"]) if status == 401: raise GitHubError(SIGNED_OUT) pod = cfg.get("pod") or {} gh = (cfg.get("connections") or {}).get("github") or {} if status == 200 and (pod.get("connected") or gh.get("connected")): names = [x["provider"].capitalize() for x in pod.get("stores", [])] or ["GitHub" if gh.get("connected") else pod.get("provider", "").capitalize()] out("Your Storage Pod is connected: " + " and ".join(names)) return names if not told: out(f"\nOne more step, once per account: connect GitHub or Dropbox at {acct['server']}/connections") out("Waiting for you to finish (Ctrl-C to skip; run 'staypuller sync' afterwards)...") told = True if waited >= limit: out("Your Storage Pod isn't connected yet. Connect it on the website, then run: staypuller sync") return None sleep(5) waited += 5 class StayPullerAccess: """Fetches one-hour GitHub tokens for the Storage-Pod repo and reuses them until they are nearly expired.""" def __init__(self, acct, now=time.time): self.acct, self.now = acct, now self.token = self.repo = None self.expires = 0.0 def get(self): if self.token and self.expires - self.now() > 300: return self.token, self.repo status, r = staypuller_api("POST", "/api/v1/github/token", token=self.acct["device_token"], server=self.acct["server"]) if status == 200 and r.get("token"): self.token, self.repo = r["token"], r["repo"] self.expires = datetime.datetime.strptime(r["expires_at"], "%Y-%m-%dT%H:%M:%SZ").replace( tzinfo=datetime.timezone.utc).timestamp() return self.token, self.repo self.token = None err = r.get("error") if status == 401: raise GitHubError(SIGNED_OUT) if status == 409 and err == "github_not_connected": raise GitHubError(f"GitHub isn't connected to your StayPuller account yet. Connect it at {self.acct['server']}/connections") if status == 403 and err == "storagepod_disabled": raise GitHubError(f"Syncing is turned off for your account. Turn it on at {self.acct['server']}/dashboard") raise GitHubError(f"StayPuller couldn't provide GitHub access (HTTP {status}).") # ---------------------------------------------------------------- Dropbox MAX_EVENTS = 200 # what one report to the Activity page may carry WORKERS = 6 # files moved at the same time with Dropbox BATCH = 16 # how many of the most important files are moved before looking for newer ones PROGRESS_SECONDS = float(os.environ.get("STAYPULLER_PROGRESS_SECONDS", "5")) # how often a long sync tells the website how it is going DROPBOX_MAX_FILE_SIZE = 100 * 1024 * 1024 # what the StayPuller server allows for a Dropbox Storage Pod HASH_BLOCK = 4 * 1024 * 1024 FRESH_SECONDS = 1.0 # a file written this recently may still be changing: wait for the next pass TEMP_PREFIX = ".staypuller-tmp-" SAFETY_FILES = 3 # if this many or more files vanish from one side at once, restore them instead of deleting them on the other class DropboxError(Exception): pass class DropboxAuthError(DropboxError): """Dropbox rejected the access token, even after getting a new one.""" class DropboxReset(DropboxError): """Dropbox no longer remembers our place in the list of changes; list everything again.""" def dropbox_content_hash_of(blocks): """Dropbox's content hash: SHA-256 of each 4 MiB block, then SHA-256 of those digests joined.""" outer = hashlib.sha256() for block in blocks: outer.update(hashlib.sha256(block).digest()) return outer.hexdigest() def dropbox_content_hash(path): def blocks(): with open(path, "rb") as f: while True: chunk = f.read(HASH_BLOCK) if not chunk: return yield chunk return dropbox_content_hash_of(blocks()) def parse_time(text): """A Dropbox timestamp ("2026-10-14T20:00:00Z") as seconds since 1970, or 0.""" try: return datetime.datetime.strptime(text, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc).timestamp() except (TypeError, ValueError): return 0.0 def path_key(path): """Dropbox ignores letter case, and a Mac may store accents in either form, so compare paths this way.""" return unicodedata.normalize("NFC", path).lower() class DropboxAPI: """The few Dropbox calls the sync needs. `creds` gives get() -> {token, api_base, content_base, account} and invalidate().""" def __init__(self, creds, sleep=time.sleep): self.creds, self.sleep = creds, sleep def account(self): return self.creds.get()["account"] def _send(self, kind, endpoint, headers=None, body=None, open_body=None, length=None, to_file=None): """One call with a fresh-token retry on 401, patience on 429 and 5xx. Returns (headers, parsed JSON or None).""" refreshed = False last = "no answer" for attempt in range(4): c = self.creds.get() base = c["content_base"] if kind == "content" else c["api_base"] req = urllib.request.Request(f"{base}/2/{endpoint}", method="POST") req.add_header("Authorization", f"Bearer {c['token']}") for k, v in (headers or {}).items(): req.add_header(k, v) data = None if open_body is not None: data = open_body() req.data = data # before the header: setting the data drops any Content-Length that was already there req.add_header("Content-Length", str(length)) elif body is not None: data = json.dumps(body).encode() req.data = data req.add_header("Content-Type", "application/json") try: resp = urllib.request.urlopen(req, timeout=120) try: if to_file is not None: with open(to_file, "wb") as out: shutil.copyfileobj(resp, out) return dict(resp.headers), None raw = resp.read() return dict(resp.headers), (json.loads(raw) if raw else None) finally: resp.close() except urllib.error.HTTPError as e: raw = e.read() try: err = json.loads(raw) except ValueError: err = {} summary = err.get("error_summary", "") if isinstance(err, dict) else "" if e.code == 401: if refreshed: raise DropboxAuthError("Dropbox didn't accept the access StayPuller was given.") refreshed = True self.creds.invalidate() continue if e.code == 429 or e.code >= 500: last = f"HTTP {e.code}" self.sleep(min(float(e.headers.get("Retry-After") or 2 ** attempt), 30)) continue if summary.startswith("reset/"): raise DropboxReset(summary) if "insufficient_space" in summary: raise DropboxError("Your Dropbox is full, so StayPuller can't add more files to your Storage Pod.") raise DropboxError(summary or f"HTTP {e.code}") except (urllib.error.URLError, OSError) as e: last = str(e) self.sleep(2 ** attempt) finally: if hasattr(data, "close") and open_body is not None: data.close() raise DropboxError(f"Couldn't reach Dropbox ({last}).") @staticmethod def _entries(page): out = [] for e in page.get("entries", []): tag = e.get(".tag") if tag == "file": out.append({"deleted": False, "key": path_key(e["path_display"].lstrip("/")), "path": e["path_display"].lstrip("/"), "hash": e.get("content_hash", ""), "rev": e.get("rev", ""), "size": e.get("size", 0), "modified": parse_time(e.get("server_modified"))}) elif tag == "deleted": out.append({"deleted": True, "key": path_key((e.get("path_display") or e["path_lower"]).lstrip("/"))}) # folders carry nothing we need: a file's path says where it lives return out def list_all(self): """Every file in the Storage Pod, and the cursor to ask for later changes from.""" _, page = self._send("api", "files/list_folder", body={"path": "", "recursive": True, "limit": 2000}) entries = self._entries(page) while page.get("has_more"): _, page = self._send("api", "files/list_folder/continue", body={"cursor": page["cursor"]}) entries += self._entries(page) return entries, page["cursor"] def list_changes(self, cursor): entries = [] while True: _, page = self._send("api", "files/list_folder/continue", body={"cursor": cursor}) entries += self._entries(page) cursor = page["cursor"] if not page.get("has_more"): return entries, cursor def download(self, path, dest): self._send("content", "files/download", headers={"Dropbox-API-Arg": json.dumps({"path": "/" + path})}, to_file=dest) def upload(self, path, local_path, size, update_rev=None): arg = {"path": "/" + path, "mode": {".tag": "update", "update": update_rev} if update_rev else "add", "autorename": False, "mute": True} _, meta = self._send("content", "files/upload", headers={"Dropbox-API-Arg": json.dumps(arg), "Content-Type": "application/octet-stream"}, open_body=lambda: open(local_path, "rb"), length=size) return {"path": meta.get("path_display", "/" + path).lstrip("/"), "hash": meta.get("content_hash", ""), "rev": meta.get("rev", ""), "size": meta.get("size", size), "modified": time.time()} def delete(self, path, rev=None): body = {"path": "/" + path} if rev: body["parent_rev"] = rev self._send("api", "files/delete_v2", body=body) class PodAccess: """Short-lived credentials for one store of the Storage Pod, fetched from StayPuller and reused until nearly expired.""" def __init__(self, acct, provider, now=time.time): self.acct, self.provider, self.now = acct, provider, now self.cred = None self.expires = 0.0 self.lock = threading.Lock() # several files are moved at once, and they share this credential def invalidate(self): self.cred = None def get(self): with self.lock: return self._get() def _get(self): if self.cred and self.expires - self.now() > 300: return self.cred status, r = staypuller_api("POST", f"/api/v1/pod/token?provider={self.provider}", token=self.acct["device_token"], server=self.acct["server"]) if status == 200 and r.get("token"): self.expires = datetime.datetime.strptime(r["expires_at"], "%Y-%m-%dT%H:%M:%SZ").replace( tzinfo=datetime.timezone.utc).timestamp() self.cred = {"token": r["token"], "api_base": r.get("api_base", "https://api.dropboxapi.com").rstrip("/"), "content_base": r.get("content_base", "https://content.dropboxapi.com").rstrip("/"), "account": (r.get("account") or "").lower()} return self.cred self.cred = None err = r.get("error") if status == 401: raise GitHubError(SIGNED_OUT) if status == 403 and err == "storagepod_disabled": raise GitHubError(f"Syncing is turned off for your account. Turn it on at {self.acct['server']}/dashboard") if status == 404 or (status == 409 and err == "pod_not_connected"): raise GitHubError(f"Dropbox isn't connected to your StayPuller account. Connect it at {self.acct['server']}/connections") if status == 409 and err == "pod_needs_reconnect": raise GitHubError(f"Dropbox needs to be connected again. Reconnect it at {self.acct['server']}/connections") raise GitHubError(f"StayPuller couldn't provide Dropbox access (HTTP {status}).") def conflict_copy_name(folder, path, device, stamp): base, ext = os.path.splitext(path) if "/" in ext: # the dot was in a directory name, not the file base, ext = path, "" device = device.replace("/", "-") candidate = f"{base} (conflict from {device} {stamp}){ext}" n = 2 while (Path(folder) / candidate).exists(): candidate = f"{base} (conflict from {device} {stamp} {n}){ext}" n += 1 return candidate def excluded(name): return name.startswith(TEMP_PREFIX) or any(fnmatch.fnmatchcase(name, pat) for pat in EXCLUDE_PATTERNS) class DropboxSyncer: """Keeps the Storage Pod folder and the Storage Pod's copy in Dropbox in step, in both directions. For every file it compares three things: what is in the folder now, what is in Dropbox now, and what the two agreed on the last time they matched (the base, remembered in a small state file). Whoever changed it since the base wins; if both changed, both are kept; a file edited on one side and deleted on the other is kept. """ def __init__(self, folder, state_file, api, now=time.time, name="Dropbox", max_size=DROPBOX_MAX_FILE_SIZE, workers=WORKERS): self.folder, self.state_file, self.api, self.now = Path(folder), Path(state_file), api, now self.workers = workers # a USB drive takes one at a time self.name, self.max_size = name, max_size # what to call the other side in messages; the biggest file it takes self.skipped_files = [] self.changed_local = False self.errors = [] self.events = [] # what moved in the last sync, for the Activity page: (kind, path, bytes) self.state = {"account": "", "cursor": None, "remote": {}, "base": {}} self._warned = set() self._hashes = {} self._events_lock = threading.Lock() self.on_progress = None # called with how many files are still to go, after each handful def totals(self): """How many files and bytes the other side holds, as last seen.""" remote = self.state.get("remote", {}) return len(remote), sum(r.get("size", 0) for r in remote.values()) def emit(self, kind, path, size=0): with self._events_lock: if len(self.events) < MAX_EVENTS: self.events.append((kind, path, int(size or 0))) def drain(self): """What moved since the last time, handed over (a long sync reports as it goes, so this stays small).""" with self._events_lock: out, self.events = self.events, [] return out # ---- state def load_state(self): try: st = json.loads(self.state_file.read_text()) if isinstance(st, dict) and isinstance(st.get("base"), dict): self.state = {"account": st.get("account", ""), "cursor": st.get("cursor"), "remote": st.get("remote", {}), "base": st["base"]} except (OSError, ValueError): pass def save_state(self): self.state_file.parent.mkdir(parents=True, exist_ok=True) tmp = self.state_file.with_suffix(".tmp") tmp.write_text(json.dumps(self.state)) os.replace(tmp, self.state_file) # ---- the three views def refresh_remote(self): st = self.state if st["cursor"]: try: entries, st["cursor"] = self.api.list_changes(st["cursor"]) except DropboxReset: st["cursor"] = None if not st["cursor"]: entries, st["cursor"] = self.api.list_all() st["remote"] = {} for e in entries: if e["deleted"]: prefix = e["key"] + "/" # a deleted folder takes everything inside it along for k in [k for k in st["remote"] if k == e["key"] or k.startswith(prefix)]: del st["remote"][k] else: st["remote"][e["key"]] = {"path": e["path"], "hash": e["hash"], "rev": e["rev"], "size": e["size"], "modified": e.get("modified", 0)} def scan_local(self): local, busy, skipped, seen = {}, set(), [], {} for dirpath, dirnames, filenames in os.walk(self.folder): dirnames[:] = sorted(d for d in dirnames if not excluded(d)) for name in sorted(filenames): if excluded(name): continue full = os.path.join(dirpath, name) try: st = os.lstat(full) except OSError: continue if not stat.S_ISREG(st.st_mode): continue # links and special files are not synced rel = os.path.relpath(full, self.folder).replace(os.sep, "/") key = path_key(rel) if key in seen: if rel not in self._warned: self._warned.add(rel) log(f"Not syncing {rel}: its name differs from {seen[key]} only by letter case, which {self.name} treats as the same file") continue seen[key] = rel if self.max_size is not None and st.st_size > self.max_size: skipped.append(rel) continue b = self.state["base"].get(key) known = b and b.get("size") == st.st_size and b.get("mtime_ns") == st.st_mtime_ns # as we last left it if self.now() - st.st_mtime < FRESH_SECONDS and not known: busy.add(key) # may still be changing; leave it for the next pass local[key] = {"path": rel, "size": st.st_size, "mtime_ns": st.st_mtime_ns} if skipped and skipped != self.skipped_files: log(f"Skipping files over {self.max_size >> 20} MB: {', '.join(skipped)}" if self.max_size >> 20 else f"Skipping files that are too big: {', '.join(skipped)}") self.skipped_files = skipped return local, busy def local_hash(self, key, lf): b = self.state["base"].get(key) if b and b.get("size") == lf["size"] and b.get("mtime_ns") == lf["mtime_ns"]: return b["hash"] # unchanged since we last looked: no need to read it again memo = self._hashes.get(key) if memo and memo[0] == (lf["size"], lf["mtime_ns"]): return memo[1] # read earlier in this sync, while the file was waiting its turn h = dropbox_content_hash(self.folder / lf["path"]) self._hashes[key] = ((lf["size"], lf["mtime_ns"]), h) return h def safe_path(self, rel): target = (self.folder / rel).resolve() if self.folder.resolve() not in target.parents: raise DropboxError(f"Refusing a path outside the folder: {rel}") return target # ---- one pass def sync(self): """Runs passes until nothing is left to do (a conflict needs a second one to upload the copy).""" self.changed_local = False self.errors = [] self.load_state() folder_missing = not self.folder.exists() if folder_missing: self.folder.mkdir(parents=True) log(f"Created folder {self.folder}") account = self.api.account() if self.state["account"] != account: if self.state["account"]: log(f"{self.name} changed ({self.state['account']} -> {account}); starting over without deleting anything") self.state = {"account": account, "cursor": None, "remote": {}, "base": {}} for round_ in range(3): again = self._pass(folder_missing and round_ == 0) self.save_state() if not again: break if self.errors: raise DropboxError(f"{len(self.errors)} file(s) could not be synced; will retry. First: {self.errors[0]}") def _pass(self, folder_missing): """Moves what needs moving, the newest first, a few at a time. After each handful it looks again, so a file that arrives while a long sync is under way (a screenshot from the phone) goes next instead of waiting at the end.""" failed, tries = set(), {} again = False restore = None with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, self.workers)) as pool: while True: self.refresh_remote() local, busy = self.scan_local() base, remote = self.state["base"], self.state["remote"] if restore is None: # Safety: if everything vanished from one side at once (an unmounted drive, a wiped folder, a different # Dropbox account), that is not a request to delete everything on the other side. Put it back instead. # Decided once, at the start, so it holds for the whole sync even as the files come back. restore = (folder_missing or (not local and len(base) >= SAFETY_FILES and not busy), not remote and len(base) >= SAFETY_FILES) actions = [] for key in sorted(set(local) | set(base) | set(remote)): if key in busy or key in failed: continue try: action = self._reconcile(key, local.get(key), base.get(key), remote.get(key), *restore) except (DropboxError, OSError) as e: self._fail(key, local, remote, base, e) failed.add(key) continue if action: actions.append(action) if not actions: return again actions.sort(key=lambda a: (a[0], a[1])) # removals first (they are quick), then the newest batch = actions[:BATCH] for _, _, key, _ in batch: tries[key] = tries.get(key, 0) + 1 if tries[key] > 3: # it keeps coming back after being handled: something is wrong, leave it for next time self._fail(key, local, remote, base, DropboxError("it did not settle")) failed.add(key) batch = [a for a in batch if a[2] not in failed] again |= self._run(pool, batch, failed, local, remote, base) if self.on_progress: try: self.on_progress(len(actions) - len(batch)) except Exception: pass # telling the website how it is going must never stop the sync def _fail(self, key, local, remote, base, error): entry = local.get(key) or remote.get(key) or base.get(key) or {"path": key} self.errors.append(f"{entry['path']}: {error}") def _run(self, pool, batch, failed, local, remote, base): again = False results = [] for group, _, key, fn in batch: if group == 0: # removals one at a time, before anything is written next to them results.append((key, self._try(fn))) jobs = [(key, pool.submit(self._try, fn)) for group, _, key, fn in batch if group != 0] results += [(key, f.result()) for key, f in jobs] for key, (ok, value) in results: if ok: again |= bool(value) else: self._fail(key, local, remote, base, value) failed.add(key) return again @staticmethod def _try(fn): try: return True, fn() except (DropboxError, OSError) as e: return False, e def _reconcile(self, key, lf, b, r, restore_local, restore_remote): """Decides what to do about one file. What needs no transfer is done at once; what does is returned as (group, order, key, function): group 0 removals, group 1 transfers, ordered newest first.""" base = self.state["base"] lh = self.local_hash(key, lf) if lf else None rh = r["hash"] if r else None bh = b["hash"] if b else None if lh == rh: # in step (or gone from both) if lf: base[key] = {"path": lf["path"], "hash": lh, "rev": r["rev"], "size": lf["size"], "mtime_ns": lf["mtime_ns"]} else: base.pop(key, None) return None local_changed, remote_changed = lh != bh, rh != bh newest_local = -(lf["mtime_ns"] / 1e9) if lf else 0 newest_remote = -(r.get("modified", 0)) if r else 0 def transfer_down(fn): return (1, newest_remote, key, fn) def transfer_up(fn): return (1, newest_local, key, fn) if restore_local and lf is None and r: return transfer_down(lambda: self._download(key, r)) # the folder was emptied: bring the file back rather than delete it if restore_remote and r is None and lf: return transfer_up(lambda: self._upload(key, lf, lh, None)) if local_changed and not remote_changed: if lf is None: return (0, 0, key, lambda: self._remove_remote(key, r)) return transfer_up(lambda: self._upload(key, lf, lh, r)) if remote_changed and not local_changed: if r is None: return (0, 0, key, lambda: self._remove_local(key, lf)) return transfer_down(lambda: self._download(key, r)) # both sides changed, and differently if lf is None: log(f"{r['path']} was deleted here but edited in {self.name}; keeping the file") return transfer_down(lambda: self._download(key, r)) if r is None: log(f"{lf['path']} was edited here but deleted in {self.name}; keeping the file") return transfer_up(lambda: self._upload(key, lf, lh, None)) return transfer_down(lambda: self._conflict(key, lf, r, rh)) def _remove_remote(self, key, r): self.api.delete(r["path"], r["rev"]) self.emit("deleted", r["path"]) log(f"Deleted {r['path']} from {self.name}") self.state["remote"].pop(key, None) self.state["base"].pop(key, None) return False def _remove_local(self, key, lf): self._delete_local(lf["path"]) self.state["base"].pop(key, None) self.emit("deleted", lf["path"]) log(f"Deleted {lf['path']} (removed from {self.name})") return False def _conflict(self, key, lf, r, rh): stamp = f"{datetime.datetime.fromtimestamp(self.now()):%Y-%m-%d %H%M}" copy = conflict_copy_name(self.folder, lf["path"], self.name, stamp) self._download(key, r, dest_rel=copy, record=False) # The folder's version stays at the path; it goes up on the next pass and the Dropbox version survives as the copy. self.state["base"][key] = {"path": lf["path"], "hash": rh, "rev": r["rev"], "size": r["size"], "mtime_ns": 0} self.emit("conflict", lf["path"], r["size"]) log(f"Conflict on {lf['path']}: kept both, the {self.name} version is saved as {copy}") return True def _upload(self, key, lf, lh, r): meta = self.api.upload(r["path"] if r else lf["path"], self.folder / lf["path"], lf["size"], update_rev=r["rev"] if r else None) self.state["remote"][key] = meta if meta.get("hash") and lh and meta["hash"] != lh: # What arrived is not what was sent (a failing drive, say). Don't call it in step: leave the two to be compared, and # keep both versions, rather than let the damaged copy overwrite the original. raise DropboxError(f"{lf['path']} didn't arrive intact; will retry") self.state["base"][key] = {"path": lf["path"], "hash": lh, "rev": meta["rev"], "size": lf["size"], "mtime_ns": lf["mtime_ns"]} self.emit("uploaded", lf["path"], lf["size"]) log(f"Uploaded {lf['path']} to {self.name}") return False def _download(self, key, r, dest_rel=None, record=True): rel = dest_rel or r["path"] dest = self.safe_path(rel) dest.parent.mkdir(parents=True, exist_ok=True) tmp = dest.parent / f"{TEMP_PREFIX}{uuid.uuid4().hex[:12]}" # unique: several files are written at once try: self.api.download(r["path"], tmp) got = dropbox_content_hash(tmp) if r["hash"] and got != r["hash"]: raise DropboxError(f"{r['path']} changed while it was downloading; will retry") os.replace(tmp, dest) finally: if tmp.exists(): tmp.unlink() self.changed_local = True if not record: # a conflict copy is ours to upload now, so it must not look as if someone is still writing it past = self.now() - 2 * FRESH_SECONDS os.utime(dest, (past, past)) if record: st = os.lstat(dest) self.state["base"][key] = {"path": rel, "hash": r["hash"], "rev": r["rev"], "size": st.st_size, "mtime_ns": st.st_mtime_ns} self.emit("downloaded", rel, st.st_size) log(f"Downloaded {rel} from {self.name}") return False def _delete_local(self, rel): target = self.safe_path(rel) target.unlink() self.changed_local = True parent = target.parent while parent != self.folder.resolve() and parent.is_dir() and not any(parent.iterdir()): parent.rmdir() # leave no empty folders behind parent = parent.parent # ---------------------------------------------------------------- USB drives DRIVE_MARKER = ".staypuller-pod" # at the top of the drive: who it is DRIVE_FOLDER = "Storage Pod" # where the files go, so the rest of the drive is left alone def read_marker(root): """The drive's identity, from the marker file on the drive itself, or None.""" try: m = json.loads((Path(root) / DRIVE_MARKER).read_text()) return m if isinstance(m, dict) and isinstance(m.get("drive_id"), str) else None except (OSError, ValueError): return None def volume_parents(): """The folders where this computer shows plugged-in drives. STORAGE_POD_VOLUMES (a test hook) replaces them.""" if os.environ.get("STORAGE_POD_VOLUMES"): return [Path(p) for p in os.environ["STORAGE_POD_VOLUMES"].split(os.pathsep) if p] user = getpass.getuser() return [Path(p) for p in (f"/media/{user}", f"/run/media/{user}", "/media", "/mnt", "/Volumes")] def mounted_volumes(): out = [] for parent in volume_parents(): try: out += sorted(p for p in parent.iterdir() if p.is_dir() and not p.name.startswith(".")) except OSError: continue return out def find_drive(drive_id): """Where a registered drive is plugged in right now, or None.""" for v in mounted_volumes(): m = read_marker(v) if m and m["drive_id"] == drive_id: return v return None class DriveStore: """A registered USB drive that is plugged in, seen through the same calls the Dropbox sync makes of Dropbox. The files live in the "Storage Pod" folder on the drive. A full list is taken each time (a drive has no change feed), so what each file hashed to is remembered by size and modification time, and a pass reads only what changed. """ def __init__(self, root, drive_id, cache_file): self.root, self.drive_id, self.cache_file = Path(root), drive_id, Path(cache_file) self.base = self.root / DRIVE_FOLDER try: self.cache = json.loads(self.cache_file.read_text()) except (OSError, ValueError): self.cache = {} def account(self): return self.drive_id def _save_cache(self): try: self.cache_file.parent.mkdir(parents=True, exist_ok=True) tmp = self.cache_file.with_suffix(".tmp") tmp.write_text(json.dumps(self.cache)) os.replace(tmp, self.cache_file) except OSError: pass def _hash(self, full, rel, st): c = self.cache.get(rel) if c and c[0] == st.st_size and c[1] == st.st_mtime_ns: return c[2] h = dropbox_content_hash(full) self.cache[rel] = [st.st_size, st.st_mtime_ns, h] return h def list_all(self): if not self.base.is_dir(): raise DropboxError(f"The drive has no \"{DRIVE_FOLDER}\" folder; was it formatted? Remove it and add it again.") entries, seen = [], set() for dirpath, dirnames, filenames in os.walk(self.base): dirnames[:] = sorted(d for d in dirnames if not excluded(d)) for name in sorted(filenames): if excluded(name): continue full = os.path.join(dirpath, name) try: st = os.lstat(full) except OSError: continue if not stat.S_ISREG(st.st_mode): continue rel = os.path.relpath(full, self.base).replace(os.sep, "/") seen.add(rel) entries.append({"deleted": False, "key": path_key(rel), "path": rel, "hash": self._hash(full, rel, st), "rev": "", "size": st.st_size}) self.cache = {k: v for k, v in self.cache.items() if k in seen} self._save_cache() return entries, "scan" def list_changes(self, cursor): raise DropboxReset("a drive is scanned in full each time") def _target(self, path): target = (self.base / path).resolve() if self.base.resolve() not in target.parents: raise DropboxError(f"Refusing a path outside the drive's folder: {path}") return target def download(self, path, dest): shutil.copyfile(self._target(path), dest) def upload(self, path, local_path, size, update_rev=None): target = self._target(path) try: free = shutil.disk_usage(self.root).free except OSError: free = None if free is not None and free < size + (1 << 20): raise DropboxError("The drive is full, so StayPuller can't add more files to it.") target.parent.mkdir(parents=True, exist_ok=True) tmp = target.parent / f"{TEMP_PREFIX}{uuid.uuid4().hex[:12]}" try: shutil.copyfile(local_path, tmp) os.utime(tmp, ns=(os.stat(local_path).st_atime_ns, os.stat(local_path).st_mtime_ns)) os.replace(tmp, target) except OSError as e: raise DropboxError(f"Couldn't copy it to the drive: {e.strerror or e}") finally: if tmp.exists(): tmp.unlink() st = os.lstat(target) rel = os.path.relpath(target, self.base.resolve()).replace(os.sep, "/") h = dropbox_content_hash(target) # reading the copy back also proves it arrived intact self.cache[rel] = [st.st_size, st.st_mtime_ns, h] self._save_cache() return {"path": rel, "hash": h, "rev": "", "size": st.st_size} def delete(self, path, rev=None): target = self._target(path) target.unlink() self.cache.pop(path, None) parent = target.parent while parent != self.base.resolve() and parent.is_dir() and not any(parent.iterdir()): parent.rmdir() parent = parent.parent def register_drive(root, label, acct, out=say, confirm=None): """Makes a plugged-in drive one of this account's Storage Pod stores. Returns the drive id.""" root = Path(root) marker = read_marker(root) drive_id = marker["drive_id"] if marker else str(uuid.uuid4()) base = root / DRIVE_FOLDER existing = sum(1 for _ in base.rglob("*") if _.is_file()) if base.is_dir() else 0 if existing and confirm is not None and not confirm(existing): raise GitHubError("Not added.") if not marker: (root / DRIVE_MARKER).write_text(json.dumps({"v": 1, "drive_id": drive_id, "created": f"{datetime.datetime.utcnow():%Y-%m-%dT%H:%M:%SZ}"})) base.mkdir(exist_ok=True) status, r = staypuller_api("POST", "/api/v1/drives", {"drive_id": drive_id, "label": label}, token=acct["device_token"], server=acct["server"]) if status == 401: raise GitHubError(SIGNED_OUT) if status == 409 and r.get("error") == "too_many_drives": raise GitHubError("You already have the most drives StayPuller can keep. Remove one on the Connections page first.") if status != 200: raise GitHubError(f"StayPuller couldn't add the drive (HTTP {status}).") out(f"Added \"{label}\". Your Storage Pod will be kept on it whenever it is plugged in.") return drive_id def cmd_drive(args): acct = read_account() if not acct: sys.exit("Sign in first: staypuller login") if args.action == "list": status, cfg = staypuller_api("GET", "/api/v1/config", token=acct["device_token"], server=acct["server"]) if status != 200: sys.exit(f"StayPuller couldn't be reached (HTTP {status}).") drives = cfg.get("drives") or [] if not drives: print("No USB drives are added. Plug one in and run: staypuller drive add") for d in drives: where = find_drive(d["drive_id"]) print(f"{d['label']}: " + (f"plugged in at {where}" if where else "not plugged in")) return # add registered = set() status, cfg = staypuller_api("GET", "/api/v1/config", token=acct["device_token"], server=acct["server"]) if status == 200: registered = {d["drive_id"] for d in cfg.get("drives") or []} path = Path(args.path) if args.path else None if path is None: found = [v for v in mounted_volumes() if (read_marker(v) or {}).get("drive_id") not in registered] if not found: sys.exit("No unregistered drive is plugged in. Plug in a USB drive and try again, or give its path: staypuller drive add /path/to/drive") print("Which drive?") for i, v in enumerate(found, 1): print(f" {i}) {v}") choice = input(f"Choose 1-{len(found)}: ").strip() if sys.stdin.isatty() else "1" try: path = found[int(choice) - 1] except (ValueError, IndexError): sys.exit("Not a choice.") if not path.is_dir(): sys.exit(f"{path} is not a folder.") label = args.label or path.name or "USB drive" def confirm(n): if args.yes: return True print(f"This drive already has a \"{DRIVE_FOLDER}\" folder with {n} file(s). They will be merged with your Storage Pod.") return sys.stdin.isatty() and input("Add it anyway? [y/N] ").strip().lower() == "y" try: register_drive(path, label, acct, confirm=confirm) except (GitHubError, OSError) as e: sys.exit(str(e)) # ---------------------------------------------------------------- engine # ---- sharing a file with someone by email # # A file stays in your Storage Pod; the person you share it with sees it in their Shared Pod on staypuller.com (and gets an # email with a link). A file put in the "Shared" folder asks who to share it with, with a desktop notification. SHARED_FOLDER = "Shared" OFFERED_FILE = STATE_DIR / "shared-offered.json" SHARE_SETTLE = 3 # seconds a new file must have been still before it is offered class ShareError(Exception): pass def share_api(method, path, body=None): acct = read_account() if not acct: raise ShareError("Sign in first: staypuller login") status, out = staypuller_api(method, path, body, token=acct["device_token"], server=acct["server"]) if status == 401: raise ShareError(SIGNED_OUT) return status, out def split_emails(values): out = [] for chunk in values: for e in chunk.replace(";", ",").replace(" ", ",").split(","): e = e.strip().lower() if e and e not in out: out.append(e) return out def pod_relative(path, folder=None): """Where a file is inside the Storage Pod ('Taxes/2025.pdf'), or None if it is somewhere else.""" try: return Path(os.path.abspath(path)).relative_to(os.path.abspath(folder or FOLDER)).as_posix() except ValueError: return None def share_files(paths, emails, note="", sleep=time.sleep, sync=None, out=say): """Shares files with email addresses. A file outside the Storage Pod folder is copied into its Shared folder first.""" rels = [] for p in paths: p = Path(p).expanduser() if not p.is_file(): raise ShareError(f"{p} is not a file. (Folders can't be shared yet; share the files inside.)") rel = pod_relative(p) if rel is None: dest = FOLDER / SHARED_FOLDER / p.name dest.parent.mkdir(parents=True, exist_ok=True) n = 1 while dest.exists(): n += 1 dest = dest.with_name(f"{p.stem} ({n}){p.suffix}") shutil.copy2(p, dest) old = min(p.stat().st_mtime, time.time() - 60) # not "just written", so the next sync picks the copy up at once os.utime(dest, (old, old)) rel = dest.relative_to(FOLDER).as_posix() out(f"Copied {p.name} into your Storage Pod ({SHARED_FOLDER}/).") rels.append(rel) if sync: sync() body = {"paths": rels, "emails": split_emails(emails), "note": note} for attempt in range(5): status, res = share_api("POST", "/api/v1/shares", body) # A file just added reaches the store on the next sync; ask again for a little while. if status == 404 and res.get("error") == "not_found" and attempt < 4: sleep(15) continue break if status != 200: raise ShareError(res.get("message") or f"StayPuller couldn't share that (HTTP {status}).") return res def cmd_share(args): try: res = share_files(args.files, args.to or [], args.note or "", sync=lambda: Engine().sync_once()) except ShareError as e: sys.exit(str(e)) r = res.get("result", {}) out = r.get("shared") or [] say(f"Shared {len(args.files)} file(s) with {', '.join(out)}. They were emailed a link that works for 3 days.") if r.get("email_failed"): say(f"We could not email: {', '.join(r['email_failed'])}. They can ask for a link at staypuller.com/shared/link.") def cmd_shared(args): try: status, res = share_api("GET", "/api/v1/shares") except ShareError as e: sys.exit(str(e)) if status != 200: sys.exit(f"StayPuller couldn't list shares (HTTP {status}).") by, with_me = res.get("by_me") or [], res.get("with_me") or [] say("Shared by me:" if by else "You aren't sharing any files.") for x in by: opened = f", opened {x['last_opened_at'][:10]}" if x.get("last_opened_at") else ", not opened yet" say(f" {x['id']} {x['name']} -> {x['recipient']}{opened}") say("Shared with me:" if with_me else "Nothing is shared with you.") for x in with_me: say(f" {x['name']} from {x['owner']} (download at staypuller.com/shared)") def cmd_unshare(args): try: status, res = share_api("DELETE", "/api/v1/shares/" + args.id) except ShareError as e: sys.exit(str(e)) if status != 200 or not res.get("stopped"): sys.exit("No share with that id. Run: staypuller shared") say("Sharing stopped. The file is gone from their Shared Pod.") def cmd_find(args): try: status, res = share_api("GET", "/api/v1/files/search?q=" + urllib.parse.quote(" ".join(args.words))) except ShareError as e: sys.exit(str(e)) if status != 200: sys.exit(res.get("message") or f"StayPuller couldn't search (HTTP {status}).") for f in res.get("files") or []: say(f"{f['path']} ({f['size']} bytes)") if not res.get("files"): say("No file by that name in your Storage Pod.") def notify(title, body): """A desktop notification, if this computer has notify-send.""" if shutil.which("notify-send"): try: subprocess.run(["notify-send", "--app-name=StayPuller", title, body], timeout=10) except (OSError, subprocess.SubprocessError): pass # ---- Tidbits: send the clipboard (or text, a file, standard input) to the Tidbits board in the Storage Pod, through the StayPuller account. # Meant to be bound to a key: `staypuller tidbit --hotkey` does it on GNOME; on other desktops bind "staypuller tidbit --clipboard" in the # keyboard settings. The clipboard is read only when this command runs, never in the background. TIDBIT_MAX_FILE = 5 * 1024 * 1024 TIDBIT_MAX_TEXT = 20 * 1024 # clipboard targets that password managers set to say "this is a secret" SECRET_TARGETS = ("x-kde-passwordmanagerhint", "org.nspasteboard.concealedtype") def read_clipboard(): """(text, secret): what is on the clipboard (Wayland or X11), and whether a password manager marked it as secret. Raises ShareError if it can't be read.""" if shutil.which("wl-paste") and os.environ.get("WAYLAND_DISPLAY"): types = subprocess.run(["wl-paste", "--list-types"], capture_output=True, text=True, timeout=10).stdout.lower() r = subprocess.run(["wl-paste", "--no-newline"], capture_output=True, text=True, timeout=10) elif shutil.which("xclip"): types = subprocess.run(["xclip", "-selection", "clipboard", "-t", "TARGETS", "-o"], capture_output=True, text=True, timeout=10).stdout.lower() r = subprocess.run(["xclip", "-selection", "clipboard", "-o"], capture_output=True, text=True, timeout=10) elif shutil.which("xsel"): types = "" r = subprocess.run(["xsel", "--clipboard", "--output"], capture_output=True, text=True, timeout=10) else: raise ShareError("To read the clipboard, install wl-clipboard (Wayland) or xclip (X11), or give the text: staypuller tidbit \"some text\"") return r.stdout, any(t in types for t in SECRET_TARGETS) def multipart_body(name, data): boundary = "StayPuller" + os.urandom(8).hex() head = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name.replace(chr(34), "_")}"\r\n' "Content-Type: application/octet-stream\r\n\r\n").encode() return boundary, head + data + f"\r\n--{boundary}--\r\n".encode() def send_tidbit(text=None, file=None): """Adds a tidbit. Returns the sentence to say; raises ShareError with one when it did not work.""" acct = read_account() if not acct: raise ShareError("Sign in first: staypuller login") url = acct["server"].rstrip("/") + "/api/v1/tidbits" headers = {"Authorization": "Bearer " + acct["device_token"], "Accept": "application/json"} if file is not None: p = Path(file).expanduser() if not p.is_file(): raise ShareError(f"{p} is not a file.") if p.stat().st_size > TIDBIT_MAX_FILE: raise ShareError("That file is too big for a tidbit (over 5 MB).") boundary, body = multipart_body(p.name, p.read_bytes()) headers["Content-Type"] = "multipart/form-data; boundary=" + boundary else: text = (text or "").strip() if not text: raise ShareError("There is nothing to add.") if len(text.encode()) > TIDBIT_MAX_TEXT: raise ShareError("That is too long for a tidbit (over 20 KB). Save it as a file instead.") body = json.dumps({"text": text}).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=body, method="POST", headers=headers) try: with urllib.request.urlopen(req, timeout=30): return "Added to your Tidbits." except urllib.error.HTTPError as e: try: msg = json.loads(e.read() or b"{}").get("message") except ValueError: msg = None if e.code == 401: raise ShareError(SIGNED_OUT) raise ShareError(msg or f"Tidbits couldn't be reached right now (HTTP {e.code}).") except (urllib.error.URLError, OSError): raise ShareError("Couldn't reach StayPuller. Check your connection.") def tidbit_hotkey(): """Binds Control+Alt+T to `staypuller tidbit --clipboard` on GNOME. Other desktops: say how.""" command = f"{BIN_PATH} tidbit --clipboard" if BIN_PATH.exists() else f"{sys.executable} {os.path.abspath(__file__)} tidbit --clipboard" if not shutil.which("gsettings"): say(f'Bind a key to this command in your desktop\'s keyboard settings:\n {command}') return base = "org.gnome.settings-daemon.plugins.media-keys" try: listed = subprocess.run(["gsettings", "get", base, "custom-keybindings"], capture_output=True, text=True, timeout=10, check=True).stdout paths = re.findall(r"'([^']+)'", listed) path = "/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/staypuller-tidbit/" if path not in paths: paths.append(path) schema = f"{base}.custom-keybinding:{path}" for key, value in (("name", "Send clipboard to StayPuller Tidbits"), ("command", command), ("binding", "t")): subprocess.run(["gsettings", "set", schema, key, value], check=True, timeout=10) subprocess.run(["gsettings", "set", base, "custom-keybindings", "[" + ", ".join(f"'{p}'" for p in paths) + "]"], check=True, timeout=10) except (OSError, subprocess.SubprocessError): say(f'This desktop did not take the shortcut. Bind a key to this command in its keyboard settings:\n {command}') return say("Control+Alt+T now sends your clipboard to Tidbits. (Change it in Settings, Keyboard, Custom Shortcuts.)") def cmd_tidbit(args): if args.hotkey: return tidbit_hotkey() quiet = args.quiet try: if args.file: msg = send_tidbit(file=args.file) else: words = args.text or [] if args.clipboard or not words: if words: sys.exit("Give text or --clipboard, not both.") text, secret = read_clipboard() if args.clipboard else (sys.stdin.read(), False) if secret: raise ShareError("That looks like a password or a one-time code, so it was not added to Tidbits.") elif words == ["-"]: text = sys.stdin.read() else: text = " ".join(words) msg = send_tidbit(text=text) except ShareError as e: if quiet: notify("Not added to Tidbits", str(e)) sys.exit(str(e)) if quiet: notify("Tidbits", msg) say(msg) def ask_who_to_share_with(name): """Asks 'who do you want to share this with?': a notification to click, then a small box to type in. Returns addresses or None.""" hook = os.environ.get("STAYPULLER_SHARE_PROMPT") # a command that prints the answer (tests, and people who want their own dialog) if hook: out = subprocess.run([hook, name], capture_output=True, text=True, timeout=300).stdout return split_emails([out]) or None if shutil.which("notify-send"): try: r = subprocess.run(["notify-send", "--app-name=StayPuller", "--action=share=Share\u2026", "--expire-time=0", "New file in Shared", f"Share \u201c{name}\u201d with someone?"], capture_output=True, text=True, timeout=3600) if r.returncode == 0 and "share" not in r.stdout: return None # the notification was dismissed except (OSError, subprocess.SubprocessError): pass # no actions on this desktop: go straight to the box prompt = f"Email address(es) to share \u201c{name}\u201d with (separate with commas):" for cmd in (["zenity", "--entry", "--title=Share with\u2026", f"--text={prompt}"], ["kdialog", "--title", "Share with\u2026", "--inputbox", prompt]): if shutil.which(cmd[0]): try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) except (OSError, subprocess.SubprocessError): return None return split_emails([r.stdout]) if r.returncode == 0 else None log("A new file is in the Shared folder; share it with: staypuller share --to someone@example.com") return None def shareable_name(name): return not (name.startswith(".") or name.endswith((".part", ".crdownload", ".tmp", "~")) or excluded(name)) class ShareOffers: """Watches the Shared folder: a file that arrives there starts the question 'who should get this?'.""" def __init__(self, folder=None, state=None, ask=None, share=None, now=time.time, threaded=True): self.folder = Path(folder or FOLDER) / SHARED_FOLDER self.state = Path(state or OFFERED_FILE) self.ask = ask or ask_who_to_share_with self.share = share or (lambda rels, emails: share_files([self.folder.parent / r for r in rels], emails)) self.now, self.threaded = now, threaded def signatures(self): out = {} if not self.folder.is_dir(): return out for p in self.folder.rglob("*"): if p.is_file() and shareable_name(p.name) and not any(part.startswith(".") for part in p.relative_to(self.folder).parts[:-1]): st = p.stat() if self.now() - st.st_mtime >= SHARE_SETTLE: out[p.relative_to(self.folder.parent).as_posix()] = f"{st.st_size}:{int(st.st_mtime)}" return out def load(self): try: return json.loads(self.state.read_text()) except (OSError, ValueError): return None def save(self, offered): self.state.parent.mkdir(parents=True, exist_ok=True) tmp = self.state.with_suffix(".tmp") tmp.write_text(json.dumps(offered)) os.replace(tmp, self.state) def check(self): """Offers files that are new since the last look. The first look only notes what is already there.""" now_files = self.signatures() offered = self.load() if offered is None: self.save(now_files) return [] new = [rel for rel, sig in now_files.items() if offered.get(rel) != sig] if new: offered.update({rel: now_files[rel] for rel in new}) # asked once: a dismissed question is not asked again self.save(offered) for rel in new: if self.threaded: threading.Thread(target=self.handle, args=(rel,), daemon=True).start() else: self.handle(rel) return new def handle(self, rel): try: emails = self.ask(Path(rel).name) if not emails: return res = self.share([rel], emails) r = res.get("result", {}) who = ", ".join(r.get("shared") or emails) log(f"Shared {rel} with {who}") notify("Shared", f"\u201c{Path(rel).name}\u201d is shared with {who}. They were emailed a link.") except ShareError as e: log(f"Couldn't share {rel}: {e}") notify("StayPuller couldn't share that", str(e)) except Exception as e: # never let a prompt stop syncing log(f"Sharing {rel}: {e}") class Engine: """Keeps the folder in step with every store the Storage Pod is kept on (GitHub, Dropbox). The folder is the meeting point: a change that arrives from one store lands in the folder and is then sent on to the others on the next pass, so the stores end up the same without talking to each other. """ def __init__(self): self.syncer = None # GitHub self.access = None self.dropbox = None # Dropbox self.dropbox_acct = None self.drive_syncers = {} # drive id -> the sync with that USB drive self.last_results = [] self.unsent, self.last_progress = [], 0.0 self.offers = ShareOffers() def config(self, acct): status, cfg = staypuller_api("GET", "/api/v1/config", token=acct["device_token"], server=acct["server"]) if status == 401: raise GitHubError(SIGNED_OUT) if status != 200: raise GitHubError(f"StayPuller couldn't be reached (HTTP {status}).") return cfg @staticmethod def providers(cfg): """Which stores this account keeps its Storage Pod on, primary first.""" pod = cfg.get("pod") or {} if pod.get("stores"): return [x["provider"] for x in pod["stores"]] if pod.get("connected"): # an older server that knows only one store return [pod["provider"]] return ["github"] if ((cfg.get("connections") or {}).get("github") or {}).get("connected") else [] def stores(self, acct): return self.providers(self.config(acct)) def sync_github_with_account(self, acct): if self.access is None or self.access.acct != acct: self.access = StayPullerAccess(acct) token, repo = self.access.get() remote = f"{GITHUB_BASE}/{repo}.git" if self.syncer is None or self.syncer.remote_url != remote: self.syncer = Syncer(FOLDER, GIT_DIR, remote, DEVICE_NAME) self.syncer.git.token = token self.syncer.sync() def sync_github_with_token(self): token = read_token() if not token: raise GitHubError("Not signed in. Run: staypuller login") if self.syncer is None: login = github_user(token) ensure_repo(login, token) self.syncer = Syncer(FOLDER, GIT_DIR, f"https://github.com/{login}/{REPO_NAME}.git", DEVICE_NAME) self.syncer.git.token = token self.syncer.sync() def sync_dropbox(self, acct): if self.dropbox is None or self.dropbox_acct != acct: self.dropbox = DropboxSyncer(FOLDER, STATE_DIR / "dropbox.json", DropboxAPI(PodAccess(acct, "dropbox"))) self.dropbox_acct = acct self.dropbox.on_progress = lambda left: self.progress(acct, "Dropbox", left) self.dropbox.sync() def sync_drive(self, acct, drive, where): key = drive["drive_id"] syncer = self.drive_syncers.get(key) if syncer is None or syncer.api.root != where or syncer.name != drive["label"]: store = DriveStore(where, key, STATE_DIR / f"drive-{key}-hashes.json") syncer = DropboxSyncer(FOLDER, STATE_DIR / f"drive-{key}.json", store, name=drive["label"], max_size=None, workers=1) syncer.on_progress = lambda left, label=drive["label"]: self.progress(acct, label, left) self.drive_syncers[key] = syncer syncer.sync() staypuller_api("POST", "/api/v1/drives/seen", {"drive_id": key}, token=acct["device_token"], server=acct["server"]) # best effort def sync_once(self): acct = read_account() waiting = [] # registered drives that are not plugged in: nothing to do, and nothing to delete if acct: cfg = self.config(acct) providers = self.providers(cfg) drives = cfg.get("drives") or [] if not providers and not drives: raise GitHubError(f"Your Storage Pod isn't connected yet. Connect GitHub or Dropbox at {acct['server']}/connections") runs = [] for p in providers: if p == "github": runs.append(("GitHub", lambda: self.sync_github_with_account(acct))) elif p == "dropbox": runs.append(("Dropbox", lambda: self.sync_dropbox(acct))) for d in drives: where = find_drive(d["drive_id"]) if where is None: waiting.append(d) else: runs.append((d["label"], lambda d=d, where=where: self.sync_drive(acct, d, where))) else: runs = [("GitHub", self.sync_github_with_token)] results = {} names = [n for n, _ in runs] self.unsent = [] # what moved and has not been reported yet, for the Activity page self.last_progress = 0.0 self.report(acct, [], [{"store": n, "state": "syncing"} for n in names] + [{"store": d["label"], "state": "waiting", "detail": "not plugged in"} for d in waiting]) self.acct_for_progress = acct for round_ in range(3 if len(runs) > 1 else 1): if round_: time.sleep(FRESH_SECONDS + 0.2) # files another store just wrote must not look as if they are still being written changed = False for name, run in runs: before = fingerprint(FOLDER) try: run() results[name] = "" except Exception as e: if isinstance(e, (GitHubError, GitError)) and name == "GitHub": self.syncer = None # re-check account/repo next time results[name] = str(e) if len(runs) > 1: log(f"{name}: {e}") self.collect(name) if fingerprint(FOLDER) != before: changed = True if not changed: break skipped = list(self.syncer.skipped_files if self.syncer else []) + list(self.dropbox.skipped_files if self.dropbox else []) stores = [{"name": n, "state": "error" if m else "synced", "message": m} for n, m in results.items()] stores += [{"name": d["label"], "state": "waiting", "message": "not plugged in"} for d in waiting] errors = [f"{n}: {m}" if len(results) > 1 else m for n, m in results.items() if m] status = [] for n, m in results.items(): entry = {"store": n, "state": "error" if m else "synced", "detail": m[:250]} totals = getattr(self.syncer_named(n), "totals", lambda: None)() if totals: entry["files"], entry["bytes"] = totals status.append(entry) status += [{"store": d["label"], "state": "waiting", "detail": "not plugged in"} for d in waiting] self.flush(acct, status) if errors: write_status("error", message="; ".join(errors), skipped=skipped, stores=stores) raise GitHubError("; ".join(errors)) write_status("synced", skipped=skipped, stores=stores) def syncer_named(self, name): if name == "GitHub": return self.syncer if name == "Dropbox": return self.dropbox for d in self.drive_syncers.values(): if d.name == name: return d return None def collect(self, name): syncer = self.syncer_named(name) drain = getattr(syncer, "drain", None) events = drain() if drain else [e for e in getattr(syncer, "events", [])] if not drain and syncer is not None: syncer.events = [] self.unsent += [(name, k, p, b) for k, p, b in events] def flush(self, acct, status): """Sends everything not yet reported (a long list goes in several reports), the status with the last of it.""" while True: chunk, self.unsent = self.unsent[:MAX_EVENTS], self.unsent[MAX_EVENTS:] body = [{"store": n, "kind": k, "path": p, "bytes": b} for n, k, p, b in chunk] self.report(acct, body, [] if self.unsent else status) if not self.unsent: return def progress(self, acct, name, left): """During a long sync: say every few seconds what has moved and how much is left, so the website follows along.""" now = time.monotonic() if now - self.last_progress < PROGRESS_SECONDS: return self.last_progress = now self.collect(name) status = [{"store": name, "state": "syncing", "detail": f"{left:,} file{'s' if left != 1 else ''} to go" if left else ""}] chunk, self.unsent = self.unsent[:MAX_EVENTS], self.unsent[MAX_EVENTS:] self.report(acct, [{"store": n, "kind": k, "path": p, "bytes": b} for n, k, p, b in chunk], status) def report(self, acct, events, status): """Tells StayPuller what this pass did, for the Activity page. Best effort: never gets in the way of syncing.""" if not acct or not (events or status): return try: staypuller_api("POST", "/api/v1/sync/report", {"events": events, "status": status}, token=acct["device_token"], server=acct["server"]) except Exception: pass def run_forever(self): log(f"StayPuller running: {FOLDER} <-> your Storage Pod") last_print = None dirty_since = None next_poll = 0.0 while True: now = time.monotonic() fp = fingerprint(FOLDER) if last_print is not None and fp != last_print and dirty_since is None: dirty_since = now last_print = fp if (dirty_since is not None and now - dirty_since >= DEBOUNCE) or now >= next_poll: try: self.sync_once() except Exception as e: log(f"Sync error: {e}") try: self.offers.check() # a new file in Shared asks who to share it with except Exception as e: log(f"Shared folder: {e}") dirty_since = None next_poll = time.monotonic() + POLL_INTERVAL last_print = fingerprint(FOLDER) # absorb changes the sync itself made time.sleep(SCAN_INTERVAL) def fingerprint(folder): """Cheap snapshot of the folder tree used to notice local changes.""" entries = [] for dirpath, dirnames, filenames in os.walk(folder): for name in filenames + dirnames: try: st = os.lstat(os.path.join(dirpath, name)) entries.append((dirpath, name, st.st_size, st.st_mtime_ns)) except OSError: pass return hash(tuple(sorted(entries))) if entries or os.path.isdir(folder) else None def write_status(state, message="", skipped=(), stores=()): try: STATE_DIR.mkdir(parents=True, exist_ok=True) STATUS_FILE.write_text(json.dumps({ "state": state, "message": message, "skipped": list(skipped), "stores": list(stores), "time": f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S}", })) except OSError: pass # ---------------------------------------------------------------- commands UNIT = """[Unit] Description=StayPuller keeps your Storage Pod folder in step After=network-online.target Wants=network-online.target [Service] ExecStart={bin} run Restart=always RestartSec=15 [Install] WantedBy=default.target """ def have_systemd(): return shutil.which("systemctl") and subprocess.run( ["systemctl", "--user", "show-environment"], capture_output=True).returncode == 0 def keep_old_command_working(): """Older installs ran "storage-pod", and their background service points at it: make that name a link to the new one.""" if LEGACY_BIN_PATH.is_symlink() or LEGACY_BIN_PATH.exists(): LEGACY_BIN_PATH.unlink() LEGACY_BIN_PATH.parent.mkdir(parents=True, exist_ok=True) LEGACY_BIN_PATH.symlink_to(BIN_PATH) def cmd_install(_): check_git_version() me = Path(__file__).resolve() BIN_PATH.parent.mkdir(parents=True, exist_ok=True) if me != BIN_PATH.resolve(): shutil.copy2(me, BIN_PATH) BIN_PATH.chmod(0o755) keep_old_command_working() print(f"Installed command: {BIN_PATH} (also available as {LEGACY_BIN_PATH.name})") if not read_account() and not read_token(): choose_sign_in() print(f"Syncing folder: {FOLDER}") try: Engine().sync_once() print("First sync complete.") except Exception as e: print(f"First sync failed: {e}") if have_systemd(): UNIT_PATH.parent.mkdir(parents=True, exist_ok=True) UNIT_PATH.write_text(UNIT.format(bin=BIN_PATH)) subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) subprocess.run(["systemctl", "--user", "enable", "--now", "storage-pod.service"], check=True) subprocess.run(["systemctl", "--user", "restart", "storage-pod.service"], check=True) print("Background sync service is running and will start automatically when you log in.") print(f"To keep syncing even when logged out: loginctl enable-linger {getpass.getuser()}") else: print("systemd isn't available. Add this command to your desktop's startup applications:") print(f" {BIN_PATH} run") if str(BIN_PATH.parent) not in os.environ.get("PATH", "").split(os.pathsep): print(f"Note: {BIN_PATH.parent} is not on your PATH; use the full path to run staypuller.") def cmd_uninstall(_): if have_systemd() and UNIT_PATH.exists(): subprocess.run(["systemctl", "--user", "disable", "--now", "storage-pod.service"]) UNIT_PATH.unlink() subprocess.run(["systemctl", "--user", "daemon-reload"]) print("Stopped and removed the background service.") print(f"Your files in {FOLDER} were left in place.") print(f"To remove everything else: rm -rf '{BIN_PATH}' '{LEGACY_BIN_PATH}' '{CONFIG_DIR}' '{DATA_DIR}' '{STATE_DIR}'") def choose_sign_in(): print("\nHow do you want to connect StayPuller?") print(" 1) Sign in with your StayPuller account (recommended; connect GitHub once on the website)") print(" 2) Paste a GitHub token") choice = input("Choose 1 or 2 [1]: ").strip() if sys.stdin.isatty() else "1" if choice == "2": prompt_for_token() else: sign_in() def sign_in(): try: acct = device_login() try: wait_for_store(acct) except KeyboardInterrupt: print("\nSkipped. Connect GitHub or Dropbox on the website, then run: staypuller sync") except GitHubError as e: sys.exit(str(e)) def cmd_login(_): sign_in() if have_systemd() and UNIT_PATH.exists(): subprocess.run(["systemctl", "--user", "restart", "storage-pod.service"]) def cmd_logout(_): acct = read_account() if not acct: print("Not signed in to StayPuller.") return try: staypuller_api("POST", "/api/v1/device/revoke", token=acct["device_token"], server=acct["server"]) except OSError: print("Couldn't reach StayPuller to sign this device out there; remove it on the Devices page.") ACCOUNT_FILE.unlink() print("Signed out. Your files in the Storage Pod folder were left in place.") if have_systemd() and UNIT_PATH.exists(): subprocess.run(["systemctl", "--user", "restart", "storage-pod.service"]) def cmd_set_token(_): prompt_for_token() ACCOUNT_FILE.unlink(missing_ok=True) # the token replaces a StayPuller sign-in if have_systemd() and UNIT_PATH.exists(): subprocess.run(["systemctl", "--user", "restart", "storage-pod.service"]) def cmd_sync(_): check_git_version() try: Engine().sync_once() print("Up to date.") except Exception as e: sys.exit(f"Sync failed: {e}") def cmd_run(_): check_git_version() try: Engine().run_forever() except KeyboardInterrupt: pass def cmd_status(_): print(f"Folder: {FOLDER}") acct = read_account() if acct: print(f"Account: signed in to StayPuller ({acct['server']})") elif read_token(): print("Account: using a GitHub token (staypuller login switches to a StayPuller account)") else: print("Account: not signed in (run: staypuller login)") if have_systemd() and UNIT_PATH.exists(): active = subprocess.run(["systemctl", "--user", "is-active", "storage-pod.service"], capture_output=True, text=True).stdout.strip() print(f"Service: {active}") else: print("Service: not installed (run: staypuller install)") try: s = json.loads(STATUS_FILE.read_text()) if s["state"] == "synced": print(f"Status: up to date (last synced {s['time']})") else: print(f"Status: error at {s['time']}: {s['message']}") for st in s.get("stores", []): if len(s["stores"]) > 1 or st["state"] != "synced": print(f" {st['name']}: " + {"synced": "up to date", "waiting": "not plugged in"}.get(st["state"], f"error: {st['message']}")) if s.get("skipped"): print("Not synced (too big): " + ", ".join(s["skipped"])) except (OSError, ValueError, KeyError): print("Status: no sync yet") try: lines = LOG_FILE.read_text().splitlines()[-10:] print("\nRecent activity:\n " + "\n ".join(lines)) except OSError: pass def main(): # Test hook: staypuller --sync-once if len(sys.argv) == 6 and sys.argv[1] == "--sync-once": try: Syncer(*sys.argv[2:]).sync() except Exception as e: sys.exit(f"error: {e}") return p = argparse.ArgumentParser(prog="staypuller", description="Keep ~/Desktop/Storage Pod in step with your Storage Pod (GitHub, Dropbox, or both).") p.add_argument("--version", action="version", version=f"staypuller {VERSION}") sub = p.add_subparsers(dest="cmd", required=True) for name, fn, help_ in [ ("install", cmd_install, "sign in and set up the background service"), ("login", cmd_login, "sign in with your StayPuller account"), ("logout", cmd_logout, "sign this computer out of StayPuller"), ("status", cmd_status, "show sync status and recent activity"), ("sync", cmd_sync, "run one sync pass now"), ("set-token", cmd_set_token, "use a GitHub token instead of a StayPuller account (GitHub only)"), ("run", cmd_run, "run the sync loop in the foreground"), ("uninstall", cmd_uninstall, "remove the background service (keeps your files)"), ]: sub.add_parser(name, help=help_).set_defaults(func=fn) share = sub.add_parser("share", help="share files from your Storage Pod with someone by email") share.add_argument("files", nargs="+", help="files to share (a file outside the Storage Pod folder is copied into its Shared folder)") share.add_argument("--to", action="append", required=True, help="email address to share with (repeat for several)") share.add_argument("--note", help="a note to include in the email") share.set_defaults(func=cmd_share) tb = sub.add_parser("tidbit", help="add text, the clipboard or a file to your Tidbits board") tb.add_argument("text", nargs="*", help='the text to add ("-" reads standard input)') tb.add_argument("--clipboard", action="store_true", help="add what is on the clipboard (bind this to a key)") tb.add_argument("--file", help="add a file or picture (up to 5 MB)") tb.add_argument("--hotkey", action="store_true", help="bind Control+Alt+T to --clipboard (GNOME)") tb.add_argument("--quiet", action="store_true", help="say the result in a desktop notification (for a key binding)") tb.set_defaults(func=cmd_tidbit) sub.add_parser("shared", help="list what you share and what is shared with you").set_defaults(func=cmd_shared) un = sub.add_parser("unshare", help="stop one share (the id is shown by 'staypuller shared')") un.add_argument("id") un.set_defaults(func=cmd_unshare) find = sub.add_parser("find", help="search your Storage Pod by file name") find.add_argument("words", nargs="+") find.set_defaults(func=cmd_find) check = sub.add_parser("shared-check") # test hook: look at the Shared folder once, asking STAYPULLER_SHARE_PROMPT who to share with check.set_defaults(func=lambda _: [log(f"offered {r}") for r in ShareOffers(threaded=False).check()]) drive = sub.add_parser("drive", help="keep your Storage Pod on a USB drive too") drive.add_argument("action", choices=["add", "list"]) drive.add_argument("path", nargs="?", help="where the drive is mounted (add: chosen from a list if left out)") drive.add_argument("--label", help="a name for the drive (default: its folder name)") drive.add_argument("--yes", action="store_true", help="don't ask before merging with files already on the drive") drive.set_defaults(func=cmd_drive) args = p.parse_args() args.func(args) if __name__ == "__main__": main()