backup-scripts/nextcloud_backup

128 lines
4.5 KiB
Python
Executable file

#!/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_subprocess_error(
title="Enabling maintenance failed.", error=e
)
return False
if "Maintenance mode already enabled" in result.stdout:
title = "Maintenance unexpectedly enabled"
text = (
"Maintenance mode was already enabled. "
"Did not expect that. Will continue."
)
self._gotify.send_warning(title, text)
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_subprocess_error("Disabling maintenance failed", e)
return False
if "Maintenance mode already disabled" in result.stdout:
title = "Maintenance mode unexpectedly already disabled"
text = (
"Maintenance mode was already disabled. "
"Did not expect that. Will continue."
)
self._gotify.send_warning(title, text)
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_subprocess_error("Dumping database failed", e)
return False
self._gotify.send_backup_successful(result)
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"] = self._config["BORG_RSH"]
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_subprocess_error("Backup failed", e)
return False
self._gotify.send_backup_successful(result)
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()