Google Drive op Linux: lokaal werken, sneller synchroniseren

#!/usr/bin/python3
"""Keep a complete local Drive tree, then expose it at ~/GoogleDrive."""
import datetime
import fcntl
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import uuid

USER_HOME = Path.home()
LOCAL = USER_HOME / '.local/share/gdrive-local'
PUBLIC = USER_HOME / 'GoogleDrive'
STATE = USER_HOME / '.local/state/gdrive-sync'
BACKUPS = USER_HOME / '.local/share/gdrive-sync-backups'
READY = STATE / 'baseline-ready'
INITIALIZING = STATE / 'initializing'
CHECK_NAME = '.gdrive-sync-access'
FILTERS = USER_HOME / '.config/rclone/gdrive-sync-filters.conf'
BASE = ['/usr/bin/rclone', '--config', str(USER_HOME / '.config/rclone/rclone.conf')]
COMMON = ['--fast-list', '--create-empty-src-dirs', '--transfers', '8', '--checkers', '16',
          '--stats', '30s', '--stats-one-line', '--stats-log-level', 'NOTICE',
          '--log-level', 'INFO', '--drive-use-trash=true']
BISYNC = ['--filters-file', str(FILTERS), '--check-access', '--check-filename', CHECK_NAME, '--max-delete', '10',
          '--resilient', '--recover', '--max-lock', '5m',
          '--conflict-resolve', 'none', '--conflict-loser', 'num',
          '--compare', 'size,modtime', '--workdir', str(STATE / 'bisync')]

def run(args, **kwargs):
    return subprocess.run([str(a) for a in args], check=True, **kwargs)

def status(phase, **extra):
    path = STATE / 'status.json'
    previous = json.loads(path.read_text()) if path.exists() else {}
    previous.update(phase=phase, updated=datetime.datetime.now().astimezone().isoformat(), **extra)
    tmp = path.with_suffix('.tmp')
    tmp.write_text(json.dumps(previous, indent=2))
    tmp.replace(path)
    print('Google Drive:', phase, flush=True)

def expose_local():
    if PUBLIC.is_symlink() and PUBLIC.resolve() == LOCAL:
        return
    if PUBLIC.exists() or PUBLIC.is_symlink():
        raise RuntimeError('GoogleDrive already exists; refusing to replace it.')
    PUBLIC.symlink_to(LOCAL, target_is_directory=True)

def sync():
    if not LOCAL.is_dir() or LOCAL.is_symlink():
        raise RuntimeError('The local sync directory is missing or unexpectedly linked.')
    if shutil.disk_usage(LOCAL).free < 15 * 1024**3:
        raise RuntimeError('Less than 15 GiB free; sync paused to protect local disk space.')
    stamp = datetime.datetime.now().strftime('%Y%m%d-%H%M%S-%f')
    backup = BACKUPS / stamp
    command = BASE + ['bisync', 'gdrive:', str(LOCAL)] + COMMON + BISYNC + ['--backup-dir2', str(backup)]
    if not READY.exists():
        if PUBLIC.exists() or PUBLIC.is_symlink():
            raise RuntimeError('GoogleDrive exists without a baseline; inspect before recovery.')
        if not INITIALIZING.exists():
            if any(LOCAL.iterdir()):
                raise RuntimeError('First setup requires an empty local sync directory.')
            INITIALIZING.write_text('Initial download in progress. Do not edit the staging folder.\n')
        status('initial-download')
        # systemd joins an already-running initial download; after reboot it resumes
        # by copying only missing/changed files. This job never modifies Drive.
        run(['systemctl', '--user', 'start', 'gdrive-initial-download.service'])
        check = STATE / CHECK_NAME
        if not check.exists():
            check.write_text(str(uuid.uuid4()) + '\n')
        run(BASE + ['copyto', str(check), 'gdrive:' + CHECK_NAME, '--ignore-existing'])
        # The staging folder is not exposed yet. Refresh its remote snapshot and
        # archive any stale staged files rather than upload them during resync.
        status('initial-verification')
        run(BASE + ['sync', 'gdrive:', str(LOCAL), '--backup-dir', str(backup), '--filter-from', str(FILTERS)] + COMMON)
        with (STATE / 'initial-dry-run.log').open('w') as log:
            run(command + ['--resync', '--dry-run'], stdout=log, stderr=subprocess.STDOUT)
        status('initial-baseline')
        run(command + ['--resync'])
        READY.write_text(datetime.datetime.now().astimezone().isoformat() + '\n')
        INITIALIZING.unlink(missing_ok=True)
    else:
        if not (LOCAL / CHECK_NAME).is_file():
            raise RuntimeError('Local access-check file is missing; refusing to sync.')
        status('syncing')
        run(command)
    status('activating-local-folder')
    expose_local()
    status('up-to-date', last_success=datetime.datetime.now().astimezone().isoformat(), error=None)

def main():
    os.umask(0o077)
    STATE.mkdir(parents=True, exist_ok=True, mode=0o700)
    with (STATE / 'runner.lock').open('w') as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return
        try:
            sync()
        except Exception as error:
            status('error', error=str(error))
            print(str(error), file=sys.stderr, flush=True)
            raise SystemExit(1)

if __name__ == '__main__':
    main()

#!/usr/bin/python3
"""Trigger the existing, serialized bisync job from local and Drive changes."""
import configparser
import ctypes
import datetime
import errno
import json
import os
from pathlib import Path
import queue
import select
import struct
import subprocess
import threading
import time

import requests

ROOT = Path.home() / '.local/share/gdrive-local'
CONFIG = Path.home() / '.config/rclone/rclone.conf'
STATE = Path.home() / '.local/state/gdrive-sync/watcher-status.json'
UNIT = 'gdrive-sync.service'
POLL_SECONDS = 10

def timestamp():
    return datetime.datetime.now().astimezone().isoformat()

class LocalWatch:
    # Linux inotify constants. Reads and directory timestamp updates are ignored.
    ATTRIB = 0x00000004
    CLOSE_WRITE = 0x00000008
    MOVED_FROM = 0x00000040
    MOVED_TO = 0x00000080
    CREATE = 0x00000100
    DELETE = 0x00000200
    DELETE_SELF = 0x00000400
    MOVE_SELF = 0x00000800
    OVERFLOW = 0x00004000
    IGNORED = 0x00008000
    ISDIR = 0x40000000
    MASK = ATTRIB | CLOSE_WRITE | MOVED_FROM | MOVED_TO | CREATE | DELETE | DELETE_SELF | MOVE_SELF

    def __init__(self, root):
        self.root = Path(root)
        self.libc = ctypes.CDLL(None, use_errno=True)
        self.libc.inotify_init1.argtypes = [ctypes.c_int]
        self.libc.inotify_init1.restype = ctypes.c_int
        self.libc.inotify_add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32]
        self.libc.inotify_add_watch.restype = ctypes.c_int
        self.fd = -1
        self.paths = {}
        self.rebuild()

    def rebuild(self):
        if not self.root.is_dir():
            raise RuntimeError('Local Google Drive directory is missing.')
        new_fd = self.libc.inotify_init1(os.O_NONBLOCK | os.O_CLOEXEC)
        if new_fd < 0:
            raise OSError(ctypes.get_errno(), 'inotify_init1')
        new_paths = {}
        try:
            for directory, dirs, _ in os.walk(self.root, followlinks=False):
                dirs[:] = [d for d in dirs if not (Path(directory) / d).is_symlink()]
                wd = self.libc.inotify_add_watch(new_fd, os.fsencode(directory), self.MASK)
                if wd < 0:
                    err = ctypes.get_errno()
                    if err in (errno.ENOENT, errno.ENOTDIR):
                        continue
                    raise OSError(err, 'inotify_add_watch')
                new_paths[wd] = Path(directory)
        except BaseException:
            os.close(new_fd)
            raise
        old_fd = self.fd
        self.fd, self.paths = new_fd, new_paths
        if old_fd >= 0:
            os.close(old_fd)

    def read(self, timeout=1):
        if not select.select([self.fd], [], [], timeout)[0]:
            return False
        changed = rebuild = False
        while True:
            try:
                data = os.read(self.fd, 1024 * 1024)
            except BlockingIOError:
                break
            offset = 0
            while offset < len(data):
                wd, mask, _, length = struct.unpack_from('iIII', data, offset)
                offset += 16
                name = os.fsdecode(data[offset:offset + length].split(b'\0', 1)[0])
                offset += length
                if mask & self.OVERFLOW:
                    changed = rebuild = True
                    continue
                if mask & self.IGNORED:
                    continue
                if mask & self.ISDIR and mask & self.ATTRIB:
                    continue
                if mask & self.CREATE and not mask & self.ISDIR:
                    # Wait for close-write instead of starting while a new file
                    # is still being copied into the directory.
                    continue
                parent = self.paths.get(wd)
                if parent is None:
                    continue
                changed = True
                if mask & self.ISDIR or mask & (self.DELETE_SELF | self.MOVE_SELF):
                    rebuild = True
        if rebuild:
            self.rebuild()
        return changed

    def close(self):
        os.close(self.fd)
        self.fd = -1

class DriveChanges:
    def __init__(self):
        self.session = requests.Session()

    def get(self, path, params):
        for attempt in range(2):
            config = configparser.ConfigParser(interpolation=None)
            config.read(CONFIG)
            token = json.loads(config['gdrive']['token'])['access_token']
            response = self.session.get('https://www.googleapis.com/drive/v3/' + path,
                params=params, headers={'Authorization': 'Bearer ' + token, 'Cache-Control': 'no-cache'}, timeout=(5, 15))
            if response.status_code == 401 and attempt == 0:
                # Let rclone manage OAuth and its own config atomically. No copies
                # of refresh tokens or application credentials are stored here.
                subprocess.run(['/usr/bin/rclone', 'about', 'gdrive:', '--config', str(CONFIG), '--json'],
                    check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=45)
                continue
            response.raise_for_status()
            return response.json()
        raise RuntimeError('Google Drive authentication failed.')

    def start_token(self):
        return self.get('changes/startPageToken', {'fields': 'startPageToken'})['startPageToken']

    def recent(self):
        # A second discovery path covers uploads not yet exposed by changes.list.
        # Phone photos may retain an old modifiedTime, so include createdTime.
        cutoff = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(minutes=15)).strftime('%Y-%m-%dT%H:%M:%SZ')
        params = {'q': f"trashed = false and (createdTime > '{cutoff}' or modifiedTime > '{cutoff}')",
                  'spaces': 'drive', 'pageSize': 1000,
                  'fields': 'nextPageToken,files(id,version,createdTime,modifiedTime,size,md5Checksum)'}
        found = {}
        while True:
            page = self.get('files', params)
            for item in page.get('files', []):
                found[item['id']] = tuple(item.get(key) for key in
                    ('version', 'createdTime', 'modifiedTime', 'size', 'md5Checksum'))
            if not page.get('nextPageToken'):
                return found
            params['pageToken'] = page['nextPageToken']

    def page(self, token):
        return self.get('changes', {
            'pageToken': token, 'pageSize': 1000, 'spaces': 'drive',
            'restrictToMyDrive': 'true', 'includeRemoved': 'true',
            'fields': 'nextPageToken,newStartPageToken,changes(fileId,removed)',
        })

def cloud_worker(events):
    drive = DriveChanges()
    token = None
    backoff = POLL_SECONDS
    recent_seen = {}
    while True:
        try:
            if token is None:
                token = drive.start_token()
                # A full sync after taking the token covers changes during downtime.
                events.put(('cloud-baseline', timestamp()))
            recent_now = drive.recent()
            if any(recent_seen.get(ident) != signature for ident, signature in recent_now.items()):
                events.put(('cloud-recent-upload', timestamp()))
            recent_seen = recent_now
            events.put(('recent-check', timestamp()))
            while True:
                page = drive.page(token)
                if page.get('changes'):
                    events.put(('cloud-change', timestamp()))
                next_token = page.get('nextPageToken')
                token = next_token or page['newStartPageToken']
                if not next_token:
                    break
            events.put(('cloud-check', timestamp()))
            backoff = POLL_SECONDS
        except Exception as error:
            # No response bodies, filenames, page tokens or credentials in logs.
            code = getattr(getattr(error, 'response', None), 'status_code', None)
            events.put(('cloud-error', f'{type(error).__name__}' + (f' HTTP {code}' if code else '')))
            if code == 410:
                token = None
            backoff = min(60, backoff * 2)
        time.sleep(backoff)

class Triggers:
    def __init__(self):
        self.first_local = None
        self.local_due = None
        self.cloud_due = None
        self.reasons = set()

    def local(self, now):
        if self.first_local is None:
            self.first_local = now
        self.local_due = min(now + 2, self.first_local + 10)
        self.reasons.add('local-change')

    def cloud(self, now, reason):
        self.cloud_due = now
        self.reasons.add(reason)

    def due(self, now):
        return any(t is not None and t <= now for t in (self.local_due, self.cloud_due))

    def clear(self):
        self.first_local = self.local_due = self.cloud_due = None
        self.reasons.clear()

def sync_state():
    result = subprocess.run(['systemctl', '--user', 'show', UNIT, '-p', 'ActiveState', '--value'],
                            check=True, capture_output=True, text=True, timeout=5)
    return result.stdout.strip()

def main():
    os.umask(0o077)
    STATE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    watch = LocalWatch(ROOT)
    events = queue.SimpleQueue()
    threading.Thread(target=cloud_worker, args=(events,), daemon=True).start()
    trigger = Triggers()
    trigger.cloud(time.monotonic(), 'watcher-start')
    details = {'started': timestamp(), 'cloud_error': None, 'watched_directories': len(watch.paths)}
    last_save = 0
    next_start = 0
    print(f'Watching {len(watch.paths)} directories; Drive change check every {POLL_SECONDS}s.', flush=True)
    try:
        while True:
            if watch.read(timeout=1):
                trigger.local(time.monotonic())
                details['last_local_change'] = timestamp()
            while not events.empty():
                kind, value = events.get()
                if kind in ('cloud-change', 'cloud-baseline', 'cloud-recent-upload'):
                    trigger.cloud(time.monotonic(), kind)
                    details['last_cloud_change'] = value
                elif kind == 'recent-check':
                    details['last_recent_upload_check'] = value
                elif kind == 'cloud-check':
                    details['last_cloud_check'] = value
                    details['cloud_error'] = None
                elif kind == 'cloud-error':
                    if details.get('cloud_error') != value:
                        print('Drive change check temporarily unavailable:', value, flush=True)
                    details['cloud_error'] = value
            now = time.monotonic()
            if trigger.due(now) and now >= next_start:
                # All entrypoints use the same systemd oneshot and runner lock.
                # Changes received during a run stay pending for a following run.
                if sync_state() in ('inactive', 'failed'):
                    reasons = ', '.join(sorted(trigger.reasons))
                    subprocess.run(['systemctl', '--user', 'start', '--no-block', UNIT], check=True, timeout=5)
                    print('Requested synchronization:', reasons, flush=True)
                    details['last_sync_requested'] = timestamp()
                    details['last_sync_reason'] = reasons
                    trigger.clear()
                    next_start = now + 5
            if now - last_save >= 5:
                details['updated'] = timestamp()
                details['watched_directories'] = len(watch.paths)
                details['pending_reasons'] = sorted(trigger.reasons)
                tmp = STATE.with_suffix('.tmp')
                tmp.write_text(json.dumps(details, indent=2))
                tmp.replace(STATE)
                last_save = now
    finally:
        watch.close()

if __name__ == '__main__':
    main()

#!/usr/bin/env bash
set -euo pipefail
umask 077

if [[ "$EUID" -eq 0 ]]; then
  printf '%s\n' 'Voer dit uit als je eigen gebruiker, zonder sudo.' >&2
  exit 1
fi
source_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
script_dir="$HOME/.local/bin"
unit_dir="$HOME/.config/systemd/user"
config_dir="$HOME/.config/rclone"
local_dir="$HOME/.local/share/gdrive-local"
state_dir="$HOME/.local/state/gdrive-sync"
files=(gdrive-sync gdrive-watch gdrive-sync.service gdrive-sync.timer
       gdrive-watch.service gdrive-initial-download.service gdrive-sync-filters.conf)
targets=("$script_dir/gdrive-sync" "$script_dir/gdrive-watch"
         "$unit_dir/gdrive-sync.service" "$unit_dir/gdrive-sync.timer"
         "$unit_dir/gdrive-watch.service" "$unit_dir/gdrive-initial-download.service"
         "$config_dir/gdrive-sync-filters.conf")

# Controleer alles voordat er bestanden worden geplaatst.
for path in "$HOME/GoogleDrive" "$local_dir" "$state_dir" "${targets[@]}"; do
  if [[ -e "$path" || -L "$path" ]]; then
    printf 'Bestaat al; installatie gestopt: %s\n' "$path" >&2
    exit 1
  fi
done
for file in "${files[@]}"; do
  [[ -f "$source_dir/$file" ]] || { printf 'Ontbreekt: %s\n' "$file" >&2; exit 1; }
done
[[ -x /usr/bin/rclone ]] || { printf '%s\n' 'Installeer rclone in /usr/bin/rclone.' >&2; exit 1; }
/usr/bin/python3 -c 'import requests'
mkdir -p "$script_dir" "$unit_dir" "$config_dir" "$local_dir" "$state_dir"
for index in "${!files[@]}"; do
  mode=600
  if (( index < 2 )); then mode=700; fi
  install -m "$mode" "$source_dir/${files[$index]}" "${targets[$index]}"
done
printf '%s\n' 'Bestanden geplaatst. Volg nu stap 4 van de handleiding.'