backup-scripts/nextcloud_backup

#!/usr/bin/env python3

import gotify
import json
import subprocess
import os
from pathlib import Path


def read_config():
    source_path = Path(__file__).resolve()
    nextcloud_secret = source_path.parent / "secrets/nextcloud.json"
    with open(nextcloud_secret, "r") as f:
        config = json.load(f)
    return config


class BackupManager:
    def __init__(self):
        self._config = read_config()
        self._gotify = gotify.Gotify(self._config["GOTIFY_TOKEN"])

    def enable_maintenance(self):
        cmd = "docker compose exec -i --user 1000:1000 app /var/www/html/occ maintenance:mode --on"
        try:
            result = subprocess.run(
                cmd, shell=True, text=True, check=True, capture_output=True
            )
        except subprocess.CalledProcessError as e:
            self._gotify.send_error(
                "❗💀❗ Enabling maintenace failed",
                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
            )
            return False
        if "Maintenance mode already enabled" in result.stdout:
            self._gotify.send_info(
                "❗ Maintenance unexpectedly enabled",
                (
                    "Maintenance mode was already enabled. "
                    "Did not expect that. Will continue."
                ),
            )
        return True

    def disable_maintenance(self):
        cmd = "docker compose exec -i --user 1000:1000 app /var/www/html/occ maintenance:mode --off"
        try:
            result = subprocess.run(
                cmd, shell=True, text=True, check=True, capture_output=True
            )
        except subprocess.CalledProcessError as e:
            self._gotify.send_error(
                "❗💀❗ Disabling maintenace failed",
                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
            )
            return False
        if "Maintenance mode already disabled" in result.stdout:
            self._gotify.send_info(
                "❗ Maintenance mode unexpectedly already disabled",
                (
                    "Maintenance mode was already disabled. "
                    "Did not expect that. Will continue."
                ),
            )
        return True

    def dump_database(self):
        password = self._config["MYSQL_PASSWORD"]
        user = self._config["MYSQL_USER"]
        db = self._config["MYSQL_DB"]

        cmd = f"docker compose exec -i --user 1000:1000 db mariadb-dump --single-transaction --default-character-set=utf8mb4 -h localhost -u {user} --password={password} {db} > db/nextcloud.sql"
        try:
            result = subprocess.run(
                cmd,
                shell=True,
                check=True,
                text=True,
                capture_output=True,
            )
        except subprocess.CalledProcessError as e:
            self._gotify.send_error(
                "❗💀❗ Dumping Database failed",
                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
            )
            return False
        text = "\n".join([result.stdout, result.stderr])
        self._gotify.send_info("✅ Database dumped", f"Result:\n{text}")
        return True

    def borg_backup(self):
        backup_dirs = " ".join(self._config["BACKUP_DIRS"])
        exclude_dirs = " ".join(self._config["EXCLUDE_DIRS"])
        repo_subdir = self._config["REPO_SUBDIR"]
        time_format = self._config["TIME_FORMAT"]
        backup_user = self._config["BACKUP_USER"]
        hostname = os.uname().nodename
        borg_env = os.environ.copy()
        borg_env["BORG_RSH"] = "ssh -i /home/lennartalff/.ssh/borg.ed25519"
        borg_env["BORG_PASSPHRASE"] = self._config["BORG_PASSPHRASE"]
        repo = f"ssh://{backup_user}@{backup_user}.your-storagebox.de:23/./backups/{hostname}/{repo_subdir}::{{{time_format}}}"
        cmd = f"borg create -v --stats {repo} {backup_dirs} --exclude {exclude_dirs}"
        try:
            result = subprocess.run(
                cmd,
                shell=True,
                check=True,
                text=True,
                capture_output=True,
                env=borg_env,
            )
        except subprocess.CalledProcessError as e:
            self._gotify.send_error(
                title="❗💀❗ Backup failed!",
                text=f"stdout: \n{e.stdout}\nsterr: \n{e.stderr}",
            )
            return False
        text = "\n".join([result.stdout, result.stderr])
        self._gotify.send_info("✅ Backup completed", f"Result:\n{text}\n")
        return True


def main():
    config = read_config()
    os.chdir(config["BACKUP_DOCKER_DIR"])
    backup_manager = BackupManager()
    if not backup_manager.enable_maintenance():
        backup_manager.disable_maintenance()
        exit(1)
    if not backup_manager.dump_database():
        backup_manager.disable_maintenance()
        exit(1)
    if not backup_manager.borg_backup():
        backup_manager.disable_maintenance()
        exit(1)
    if not backup_manager.disable_maintenance():
        exit(1)
    exit(0)


if __name__ == "__main__":
    main()