Simple script to rotate out backup files and folders and stuffff.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

248 lines
6.1 KiB

#!/usr/bin/env python3
Mike's Backup Rotator
A simple script to help automatically rotate backup files
Copyright 2019 Mike Peralta; All rights reserved
Released under the GNU GENERAL PUBLIC LICENSE v3 (See LICENSE file for more)
import datetime
import os
import shutil
import sys
import yaml
class BackupRotator:
def __init__(self):
self.__dry_run = False
self.__configs = []
self.__config_paths = []
self.__calculated_actions = []
def run(self):
# Rotate once per config
for config_index in range(len(self.__configs)):
config = self.__configs[config_index]
self.log("Rotating for config " + str(config_index + 1) + " of " + str(len(self.__configs)), config["__path"])
def current_time():
now =
now_s = now.strftime("%b-%d-%Y %I:%M%p")
return str(now_s)
def log(self, s, o=None):
now = self.current_time()
to_log = "[" + now + "][Backup Rotator] " + str(s)
if o is not None:
to_log += " " + str(o)
def consume_arguments(self):
self.__config_paths = []
for i in range(1, len(sys.argv)):
arg = sys.argv[i]
if arg == "--config":
i, one_path = self.consume_argument_companion(i)
self.log("Found config path argument:", one_path)
elif arg == "--dry-run":
self.__dry_run = True
self.log("Activating global dry-run mode")
def consume_argument_companion(arg_index):
companion_index = arg_index + 1
if companion_index >= len(sys.argv):
raise Exception("Expected argument after", sys.argv[arg_index])
return companion_index, sys.argv[companion_index]
def consume_configs(self, paths: list=None):
if paths is None:
raise Exception("Auto-finding of config file not implemented")
# Use each config path
for path in paths:
# If this is a single path
if os.path.isfile(path):
# If this is a directory
elif os.path.isdir(path):
# Iterate over each file inside
for file_name in os.listdir(path):
self.consume_config(os.path.join(path, file_name))
def consume_config(self, path: str):
# Open the file
f = open(path)
if not f:
raise Exception("Unable to open config file: " + path)
# Parse
config = yaml.safe_load(f)
# Add its own path
config["__path"] = path
# Consume to internal
self.log("Consumed config from path:", path)
def do_rotate(self, config):
def rotate_paths(self, config):
self.log("Begin rotating " + str(len(config["paths"])) + " paths")
for path in config["paths"]:
self.rotate_path(config, path)
def rotate_path(self, config, path):
self.log("Rotating path", path)
if "maximum-items" not in config:
raise Exception("Please provide config key: \"maximum-items\"")
max_items = config["maximum-items"]
if not os.path.isdir(path):
raise Exception("Path should be a directory:" + str(path))
children = self.gather_rotation_candidates(config, path)
# Do we need to rotate anything out?
if len(children) <= max_items:
"Path only has " + str(len(children)) + " items,"
+ " but needs " + str(max_items) + " for rotation"
+ "; Won't rotate this path."
purge_count = len(children) - max_items
"Need to purge " + str(purge_count) + " items"
for purge_index in range(purge_count):
item_to_purge = self.pick_item_to_purge(config, children)
if os.path.isfile(item_to_purge):
self.remove_file(config, item_to_purge)
elif os.path.isdir(item_to_purge):
self.remove_directory(config, item_to_purge)
raise Exception("Don't know how to remove this item: " + str(item_to_purge))
def gather_rotation_candidates(config, path):
candidates = []
if "target-type" not in config.keys():
raise Exception("Please provide the configuration key: target-type")
for item_name in os.listdir(path):
item_path = os.path.join(path, item_name)
if config["target-type"] == "file":
if not os.path.isfile(item_path):
elif config["target-type"] == "directory":
if not os.path.isdir(item_path):
raise Exception("Configuration key \"target-type\" must be \"file\" or \"directory\"")
return candidates
def pick_item_to_purge(config, items):
if "date-detection" not in config.keys():
raise Exception("Please provide config key: \"date-detection\"")
detection = config["date-detection"]
best_item = None
best_ctime = None
for item in items:
if detection == "file":
ctime = os.path.getctime(item)
if best_ctime is None or ctime < best_ctime:
best_ctime = ctime
best_item = item
raise Exception("Invalid value for \"date-detection\": " + str(detection))
return best_item
def remove_file(self, config, file_path):
if not os.path.isfile(file_path):
raise Exception("Tried to remove a file, but this path isn't a file: " + str(file_path))
if self.__dry_run:
self.log("Won't purge file during global-level dry run: ", file_path)
elif "dry-run" in config.keys() and config["dry-run"] is True:
self.log("Won't purge file during config-level dry run: ", file_path)
self.log("Purging file:", file_path)
def remove_directory(self, config, dir_path):
if not os.path.isdir(dir_path):
raise Exception("Tried to remove a directory, but this path isn't a directory: " + str(dir_path))
if self.__dry_run:
self.log("Won't purge directory during global-level dry run: ", dir_path)
elif "dry-run" in config.keys() and config["dry-run"] is True:
self.log("Won't purge directory during config-level dry run: ", dir_path)
self.log("Purging directory:", dir_path)