dotfiles/.scripts/restic.py

#!/usr/bin/env python3

import json
import subprocess
from getpass import getpass
from os import environ
from pathlib import Path

CONFIG_PATH = Path.home() / ".config/restic_bk/"


def run(command: str, **kwargs):
    return subprocess.run(command, shell=True, **kwargs)


def decorated_print(str: str):
    print(f"\n<<-<>==={str}===<>->>\n")


def write_home_visible_includes():
    home_visible_content = set(
        run("fd -d 1 '.*' ~", check=True, stdout=subprocess.PIPE, text=True)
        .stdout.strip()
        .split()
    )

    home_visible_excludes = []
    with open(CONFIG_PATH / "home_visible_excludes.txt") as f:
        for line in f:
            home_visible_excludes.append(line.strip())

    home_visible_includes = home_visible_content.difference(home_visible_excludes)

    with open(CONFIG_PATH / "home_visible_includes.txt", "w") as f:
        for line in home_visible_includes:
            f.write(line + "\n")


def mounted_backup_drive():
    mounted_drives_path = Path("/run/media") / environ["USER"]
    if not mounted_drives_path.is_dir():
        raise Exception("No drives mounted!")

    known_backup_drives = []
    with open(CONFIG_PATH / "known_backup_drives.txt") as f:
        for line in f:
            known_backup_drives.append(line.strip())

    mounted_backup_drives = []
    for path in mounted_drives_path.iterdir():
        if path.is_dir():
            drive_name = path.parts[-1]
            if drive_name in known_backup_drives:
                mounted_backup_drives.append(drive_name)

    if len(mounted_backup_drives) == 0:
        raise Exception("No known backup drives found!")

    ind = 0
    if len(mounted_backup_drives) > 1:
        decorated_print("Found multiple known backup drives")

        for i, drive in enumerate(mounted_backup_drives):
            print(f"{i}: {drive}")

        ind = int(input("Enter index: "))

    return mounted_drives_path / mounted_backup_drives[ind]


def get_parent_snapshot(env):
    snapshots_json = run(
        "restic snapshots -q --latest 1 --json",
        env=env,
        check=True,
        stdout=subprocess.PIPE,
        text=True,
    ).stdout.strip()

    return json.loads(snapshots_json)[-1]["id"]


def pre_backup():
    password = getpass("Restic repo password: ")

    write_home_visible_includes()

    repo = mounted_backup_drive() / "restic"
    init_repo = False
    if not repo.is_dir():
        decorated_print(f"No repo found at {repo} !!")

        ans = input("Initialize repo? [Y/n]: ")
        if ans.strip().lower() not in ("", "y"):
            raise Exception("No repo to work with!")
        else:
            init_repo = True

    with open(CONFIG_PATH / "pre_backup_scripts.txt") as f:
        for line in f:
            line = line.strip()
            decorated_print(f"Running {line}")
            run(f"python {line}", stderr=subprocess.DEVNULL)

    run("restic self-update", stderr=subprocess.DEVNULL)

    env = environ.copy()
    env["RESTIC_REPOSITORY"] = repo
    env["RESTIC_PASSWORD"] = password
    env["RESTIC_COMPRESSION"] = "auto"

    if init_repo:
        run("restic init", env=env, check=True)
        parent_snapshot = None
    else:
        parent_snapshot = get_parent_snapshot(env)

    decorated_print(f"Parent snapshot: {parent_snapshot}")

    return env, parent_snapshot


def backup(env, parent_snapshot):
    home_visible_includes = CONFIG_PATH / "home_visible_includes.txt"
    includes = CONFIG_PATH / "includes.txt"
    excludes = CONFIG_PATH / "excludes.txt"

    if parent_snapshot is not None:
        parent_arg = "--parent " + parent_snapshot
    else:
        parent_arg = ""

    run(
        f"restic backup -v {parent_arg} --files-from {home_visible_includes} --files-from {includes} --exclude-file {excludes}",
        env=env,
        check=True,
    )


def post_backup(env):
    run("restic check", env=env, check=True)

    run("restic snapshots --compact", env=env)


def main():
    env, parent_snapshot = pre_backup()

    backup(env, parent_snapshot)

    post_backup(env)

    decorated_print("DONE")


if __name__ == "__main__":
    main()