dotfiles/.scripts/restic.py

#!/usr/bin/python3

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.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 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

    if init_repo:
        run("restic init", env=env, check=True)

    return env


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

    run(
        f"restic backup -v --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 = pre_backup()

    backup(env)

    post_backup(env)

    decorated_print("DONE")


if __name__ == "__main__":
    main()