pybackup/pybackup.py

#!/usr/bin/python3

import sys
import os
import json
import subprocess
from datetime import datetime

# CONFIGURATION_PATH = "/etc/pybackup.conf"  # for "production"
CONFIGURATION_PATH = "pybackup.conf"  # for testing
BACKUP_NAME_PATTERN = "%m%d%Y_%H-%M-%S.pybackup.tar"  # DON'T CHANGE THIS IF YOU DON'T KNOW WHAT YOU ARE DOING!


def log(message: str, log_level: int) -> bool:
    assignments = {0: "DEBUG", 1: "VERBOSE", 2: "INFO", 3: "ERROR", 4: "FATAL"}
    if log_level not in assignments:
        return False  # show that something went wrong
    print('\033[91m', end="") if log_level >= 3 else print("", end="")  # ink the output red if error/fatal is logged
    print(f"[{assignments[log_level]}] {message}\033[0m")
    if log_level == 4:
        print("Exiting...")
        exit(1)
    return True


def make_backup(path_from: str, path_to: str):
    backup_name = datetime.now().strftime(BACKUP_NAME_PATTERN)
    tar_backup = subprocess.run(["tar", "-cf", f"{os.path.abspath(path_to)}/{backup_name}", path_from])
    if tar_backup.returncode != 0:
        log(f"Something went wrong handling the backup of {path_from}", 3)


# just to make sure this program isn't used as a module
if not __name__ == "__main__":
    print(f"[{sys.argv[0]}]: THIS BACKUP PROGRAM IS NOT A PYTHON MODULE!")
    exit(0)

conf_f = open(CONFIGURATION_PATH, "r")
try:  # load the config if possible
    conf = json.load(conf_f)
except json.decoder.JSONDecodeError:
    log("The configuration file is corrupted.", 4)
conf_f.close()

# test if the locations key exists in the config file
try:
    conf["locations"]
except KeyError:
    log("The configuration file has no locations inside. Maybe the root 'locations' key is wrongly spelled?", 4)

# test for each location if the data was correctly written into the config
counter = 0
for _location in conf["locations"]:
    try:
        _location["name"]
    except KeyError:
        log(f"The {counter + 1}th element of locations in the config has no name. Skipping this backup path...", 3)
        conf["locations"].pop(counter)
        continue

    try:
        _location["path"]
        _location["backup_path"]
        _location["frequency"]
        _location["versions"]
    except KeyError:
        log(f"The {counter + 1}th element with the name \"{_location['name']}\" of locations in the config is "
            f"corrupted. Skipping this backup path...", 3)
        conf["locations"].pop(counter)
        continue

    if type(_location["versions"]) != int:
        log(f"The value of \"versions\" must be an integer, which isn't given in {_location['name']}. Skipping it.", 3)
        conf["locations"].pop(counter)
        continue

    if type(_location["frequency"]) != int:
        log(f"The value of \"frequency\" must be an integer, which isn't given in {_location['name']}. Skipping it.", 3)
        conf["locations"].pop(counter)
        continue

    if not os.path.exists(_location["path"]) or not os.path.exists(_location["backup_path"]):
        log(f"The {counter + 1}th element with the name \"{_location['name']}\" of locations in the config has a "
            f"wrong path and/or backup path set. Maybe you spelled it wrong? Skipping this backup...", 3)
        conf["locations"].pop(counter)
        continue

    if not os.path.isdir(_location["backup_path"]):
        log(f"The backup configuration \"{_location['backup_path']}\" has no folder set as a backup path. "
            f"Skipping it...", 3)
        conf["locations"].pop(counter)
        continue

    if not os.access(_location["backup_path"], os.R_OK | os.X_OK | os.W_OK) or not os.access(_location["path"], os.R_OK):
        log(f"Backup configuration \"{_location['name']}\": no access to the path and/or backup path. "
            f"Skipping it...", 3)
        conf["locations"].pop(counter)
        continue

    if _location["frequency"] == 0:
        log(f"Frequency can't be zero (0), but it seems {_location['name']} has this frequency! Skipping it...", 3)
        conf["locations"].pop(counter)
        continue

    for entry in os.listdir(_location["backup_path"]):
        try:
            datetime.strptime(entry, BACKUP_NAME_PATTERN)
        except ValueError:
            log(f"The backup folder {_location['backup_path']} contains files different to the backup files which is"
                f" not supported. Skipping this backup...", 3)
            conf["locations"].pop(counter)
            break

    counter += 1


# the REAL backup procedure now!
for location in conf["locations"]:
    path = location["path"]
    backup_path = location["backup_path"]
    frequency = location["frequency"]
    versions = location["versions"]

    # backup if there's no backup yet
    if (backup_folder := os.listdir(backup_path)) == []:
        make_backup(path, backup_path)
    else:
        # backup if the last backup is 'frequency' days old
        backup = True
        now = datetime.now()
        for entry in backup_folder:
            timedelta = now - datetime.strptime(entry, BACKUP_NAME_PATTERN)
            if timedelta.total_seconds() <= 60*60*24*frequency:  # if the backup was made before less than frequency
                # days
                backup = False
                break
        if backup:
            make_backup(path, backup_path)

    # last, if there are more than version backups in the folder, remove the oldest
    if (backup_folder_len := len(os.listdir(backup_path))) > versions:
        files_to_be_deleted = sorted(backup_folder)[0:backup_folder_len-versions]
        for file in files_to_be_deleted:
            os.remove(f"{backup_path}/{file}")