Allow enforcement of minimum item count. Improve logging output.

.gitignore

@@ -3,4 +3,8 @@
 # Ignore PyCharm
 .idea
 
+#
+__pycache__
+
+
 

BackupRotator.py

@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+
+"""
+
+Mike's Backup Rotator
+
+A simple script to help automatically rotate backup files
+
+Copyright 2022 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 syslog
+import time
+import yaml
+
+
+class BackupRotator:
+	
+	def __init__(self):
+		
+		self.__dry_run = False
+		self.__configs = []
+		self.__config_paths = []
+		self.__calculated_actions = []
+	
+	def run(self, configs, dry_run: bool = False):
+		
+		self.log("Begin")
+		
+		self.__dry_run = dry_run
+		self.__config_paths = configs
+		
+		self._consume_configs(self.__config_paths)
+		
+		# 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"])
+			self._do_rotate(config)
+	
+	@staticmethod
+	def current_time():
+		
+		now = datetime.datetime.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)
+		
+		syslog.syslog(to_log)
+		
+		print(to_log)
+	
+	def _consume_configs(self, paths: list=None):
+		
+		assert paths is not None, "Config paths cannot be None"
+		assert len(paths) > 0, "Must provide at least one config file path"
+		
+		# Use each config path
+		for path in paths:
+			
+			# If this is a single path
+			if os.path.isfile(path):
+				self._consume_config(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.__configs.append(config)
+		self.log("Consumed config from path:", path)
+	
+	def _do_rotate(self, config):
+	
+		self._rotate_paths(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):
+		
+		assert os.path.isdir(path), "Path should be a directory: {}".format(path)
+		
+		self.log("Rotating path: {}".format(path))
+		
+		found_any_rotation_keys = False
+		if "maximum-items" in config.keys():
+			found_any_rotation_keys = True
+			self._rotate_path_for_maximum_items(config=config, path=path, max_items=config["maximum-items"])
+		if "maximum-age" in config.keys():
+			found_any_rotation_keys = True
+			self._rotate_path_for_maximum_age(config=config, path=path, max_age_days=config["maximum-age"])
+
+		assert found_any_rotation_keys is True, \
+			"Config needs one of the following keys: \"maximum-items\""
+
+	def _rotate_path_for_maximum_items(self, config, path: str, max_items: int):
+		
+		assert os.path.isdir(path), "Path should be a directory: {}".format(path)
+		
+		self.log("Rotating path for a maximum of {} items: {}".format(
+			max_items, path
+		))
+		
+		children = self._gather_rotation_candidates(config, path)
+
+		minimum_items = self._determine_minimum_items(config)
+		
+		# Do we need to rotate anything out?
+		if len(children) < minimum_items:
+			self.log("Path only has {} items, which does not meet the minimum threshold of {} items. Won't rotate this path.".format(
+				len(children), minimum_items
+			))
+		elif len(children) <= max_items:
+			self.log("Path only has {} items, but needs more than {} for rotation; Won't rotate this path.".format(
+				len(children), max_items
+			))
+			return
+		self.log("Found {} items to examine".format(len(children)))
+		
+		#
+		maximum_purge_count = len(children) - minimum_items
+		purge_count = len(children) - max_items
+		self.log("Want to purge {} items".format(purge_count))
+
+		if purge_count > maximum_purge_count:
+			self.log("Reducing purge count from {} to {} items to respect minimum items setting ({})".format(
+				purge_count, maximum_purge_count, minimum_items
+			))
+			purge_count = maximum_purge_count
+		
+		children_to_purge = []
+		for purge_index in range(purge_count):
+			
+			#
+			item_to_purge, item_ctime, item_age_seconds, item_age = self._pick_oldest_item(config, children)
+			children.remove(item_to_purge)
+			self.log("Found next item to purge: ({}) {} ({})".format(
+				purge_index + 1,
+				os.path.basename(item_to_purge),
+				item_age
+			))
+			
+			#
+			children_to_purge.append(item_to_purge)
+
+		#
+		self.log("Removing items")
+		for child_to_purge in children_to_purge:
+			child_basename = os.path.basename(child_to_purge)
+			self._remove_item(config, item_to_purge)
+	
+	def _rotate_path_for_maximum_age(self, config, path: str, max_age_days: int):
+		
+		assert os.path.isdir(path), "Path should be a directory: {}".format(path)
+		
+		self.log("Rotating path for max age of {} days: {}".format(max_age_days, path))
+		
+		children = self._gather_rotation_candidates(config, path)
+		
+		self.log("Examining {} items for deletion".format(len(children)))
+		children_to_delete = []
+		for child in children:
+			
+			age_seconds = self._detect_item_age_seconds(config, child)
+			age_days = self._detect_item_age_days(config, child)
+			age_formatted = self.seconds_to_time_string(age_seconds)
+			child_basename = os.path.basename(child)
+			
+			if age_days > max_age_days:
+				self.log("[Old enough    ] {} ({})".format(
+					child_basename, age_formatted
+				))
+				children_to_delete.append(child)
+			else:
+				self.log("[Not Old enough] {} ({})".format(
+					child_basename, age_formatted
+				))
+
+		if len(children_to_delete) > 0:
+			self.log("Removing old items ...")
+			for child_to_delete in children_to_delete:
+				basename = os.path.basename(child_to_delete)
+				self._remove_item(config, child_to_delete)
+		else:
+			self.log("No old items to remove")
+
+		
+	@staticmethod
+	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):
+					continue
+			elif config["target-type"] == "directory":
+				if not os.path.isdir(item_path):
+					continue
+			else:
+				raise Exception("Configuration key \"target-type\" must be \"file\" or \"directory\"")
+			
+			candidates.append(item_path)
+		
+		return candidates
+	
+	def _pick_oldest_item(self, config, items):
+		
+		best_item = None
+		best_ctime = None
+		for item in items:
+			
+			ctime = self._detect_item_date(config, item)
+			if best_ctime is None or ctime < best_ctime:
+				best_ctime = ctime
+				best_item = item
+		
+		age_seconds = self._detect_item_age_seconds(config, best_item)
+		age_string = self.seconds_to_time_string(age_seconds)
+
+		return best_item, best_ctime, age_seconds, age_string
+	
+	@staticmethod
+	def _detect_item_date(config, item):
+		
+		assert "date-detection" in config.keys(), "Please provide config key: \"date-detection\""
+		detection = config["date-detection"]
+
+		if detection == "file":
+			ctime = os.path.getctime(item)
+		else:
+			raise AssertionError("Invalid value for \"date-detection\"; Should be one of {file}: {}".format(
+				detection
+			))
+
+		return ctime
+
+	def _detect_item_age_seconds(self, config, item):
+		
+		now = time.time()
+		ctime = self._detect_item_date(config, item)
+		delta = now - ctime
+		
+		return delta
+
+	def _detect_item_age_days(self, config, item):
+		
+		age_seconds = self._detect_item_age_seconds(config, item)
+		age_days = int(age_seconds / 86400)
+
+		return age_days
+
+	def seconds_to_time_string(self, seconds: float):
+		
+		if isinstance(seconds, float):
+			pass
+		elif isinstance(seconds, int):
+			seconds = float * 1.0
+		else:
+			raise AssertionError("Seconds must be an int or float")
+		
+		# Map
+		map = {
+			"year": 31536000.0,
+			"month": 2592000.0,
+			"week": 604800.0,
+			"day": 86400.0,
+			"hour": 3600.0,
+			"minute": 60.0,
+			"second": 1.0
+		}
+		
+		s_parts = []
+		for unit_label in map.keys():
+			unit_seconds = map[unit_label]
+			if seconds >= unit_seconds:
+				unit_count = int(seconds / unit_seconds)
+				s_parts.append("{} {}{}".format(
+					unit_count, unit_label,
+					"" if unit_count == 1 else "s"
+				))
+				seconds -= unit_seconds * unit_count
+		
+		s = ", ".join(s_parts)
+		
+		return s
+
+	def _remove_item(self, config, path):
+		
+		if os.path.isfile(path):
+			self._remove_file(config, path)
+		elif os.path.isdir(path):
+			self._remove_directory(config, path)
+		else:
+			raise AssertionError("Don't know how to remove this item: {}".format(path))
+
+	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)
+		else:
+			self.log("Purging file:", file_path)
+			os.remove(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)
+		else:
+			self.log("Purging directory:", dir_path)
+			shutil.rmtree(dir_path)
+
+	
+	def _determine_minimum_items(self, config):
+		
+		minimum_items = 0
+		
+		if "minimum-items" in config.keys():
+			minimum_items = config["minimum-items"]
+			self.log("Won't delete anything unless a minimum of {} items were found".format(minimum_items))
+		else:
+			self.log("No value found for \"minimum-items\"; Will not enforce minimum item constraint.")
+		
+		return minimum_items

Pipfile

@@ -0,0 +1,13 @@
+[[source]]
+url = "https://pypi.python.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+pyyaml = ">=5.4"
+
+[dev-packages]
+
+[requires]
+python_version = "3"
+

Pipfile.lock

@@ -0,0 +1,60 @@
+{
+    "_meta": {
+        "hash": {
+            "sha256": "ac6a3bf65ec43902d7a8907c3b0ae70e365127a1f93ecf12080f847eadb7dc35"
+        },
+        "pipfile-spec": 6,
+        "requires": {
+            "python_version": "3"
+        },
+        "sources": [
+            {
+                "name": "pypi",
+                "url": "https://pypi.python.org/simple",
+                "verify_ssl": true
+            }
+        ]
+    },
+    "default": {
+        "pyyaml": {
+            "hashes": [
+                "sha256:0283c35a6a9fbf047493e3a0ce8d79ef5030852c51e9d911a27badfde0605293",
+                "sha256:055d937d65826939cb044fc8c9b08889e8c743fdc6a32b33e2390f66013e449b",
+                "sha256:07751360502caac1c067a8132d150cf3d61339af5691fe9e87803040dbc5db57",
+                "sha256:0b4624f379dab24d3725ffde76559cff63d9ec94e1736b556dacdfebe5ab6d4b",
+                "sha256:0ce82d761c532fe4ec3f87fc45688bdd3a4c1dc5e0b4a19814b9009a29baefd4",
+                "sha256:1e4747bc279b4f613a09eb64bba2ba602d8a6664c6ce6396a4d0cd413a50ce07",
+                "sha256:213c60cd50106436cc818accf5baa1aba61c0189ff610f64f4a3e8c6726218ba",
+                "sha256:231710d57adfd809ef5d34183b8ed1eeae3f76459c18fb4a0b373ad56bedcdd9",
+                "sha256:277a0ef2981ca40581a47093e9e2d13b3f1fbbeffae064c1d21bfceba2030287",
+                "sha256:2cd5df3de48857ed0544b34e2d40e9fac445930039f3cfe4bcc592a1f836d513",
+                "sha256:40527857252b61eacd1d9af500c3337ba8deb8fc298940291486c465c8b46ec0",
+                "sha256:473f9edb243cb1935ab5a084eb238d842fb8f404ed2193a915d1784b5a6b5fc0",
+                "sha256:48c346915c114f5fdb3ead70312bd042a953a8ce5c7106d5bfb1a5254e47da92",
+                "sha256:50602afada6d6cbfad699b0c7bb50d5ccffa7e46a3d738092afddc1f9758427f",
+                "sha256:68fb519c14306fec9720a2a5b45bc9f0c8d1b9c72adf45c37baedfcd949c35a2",
+                "sha256:77f396e6ef4c73fdc33a9157446466f1cff553d979bd00ecb64385760c6babdc",
+                "sha256:819b3830a1543db06c4d4b865e70ded25be52a2e0631ccd2f6a47a2822f2fd7c",
+                "sha256:897b80890765f037df3403d22bab41627ca8811ae55e9a722fd0392850ec4d86",
+                "sha256:98c4d36e99714e55cfbaaee6dd5badbc9a1ec339ebfc3b1f52e293aee6bb71a4",
+                "sha256:9df7ed3b3d2e0ecfe09e14741b857df43adb5a3ddadc919a2d94fbdf78fea53c",
+                "sha256:9fa600030013c4de8165339db93d182b9431076eb98eb40ee068700c9c813e34",
+                "sha256:a80a78046a72361de73f8f395f1f1e49f956c6be882eed58505a15f3e430962b",
+                "sha256:b3d267842bf12586ba6c734f89d1f5b871df0273157918b0ccefa29deb05c21c",
+                "sha256:b5b9eccad747aabaaffbc6064800670f0c297e52c12754eb1d976c57e4f74dcb",
+                "sha256:c5687b8d43cf58545ade1fe3e055f70eac7a5a1a0bf42824308d868289a95737",
+                "sha256:cba8c411ef271aa037d7357a2bc8f9ee8b58b9965831d9e51baf703280dc73d3",
+                "sha256:d15a181d1ecd0d4270dc32edb46f7cb7733c7c508857278d3d378d14d606db2d",
+                "sha256:d4db7c7aef085872ef65a8fd7d6d09a14ae91f691dec3e87ee5ee0539d516f53",
+                "sha256:d4eccecf9adf6fbcc6861a38015c2a64f38b9d94838ac1810a9023a0609e1b78",
+                "sha256:d67d839ede4ed1b28a4e8909735fc992a923cdb84e618544973d7dfc71540803",
+                "sha256:daf496c58a8c52083df09b80c860005194014c3698698d1a57cbcfa182142a3a",
+                "sha256:e61ceaab6f49fb8bdfaa0f92c4b57bcfbea54c09277b1b4f7ac376bfb7a7c174",
+                "sha256:f84fbc98b019fef2ee9a1cb3ce93e3187a6df0b2538a651bfb890254ba9f90b5"
+            ],
+            "index": "pypi",
+            "version": "==6.0"
+        }
+    },
+    "develop": {}
+}

backup-rotator

@@ -1,258 +0,0 @@
-#!/usr/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):
-		
-		self.log("Begin")
-		self.consume_arguments()
-		self.consume_configs(self.__config_paths)
-		
-		# 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"])
-			self.do_rotate(config)
-	
-	@staticmethod
-	def current_time():
-		
-		now = datetime.datetime.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)
-		
-		print(to_log)
-	
-	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.__config_paths.append(one_path)
-				print("Found config path argument:", one_path)
-			
-			elif arg == "--dry-run":
-				self.__dry_run = True
-				print("Activating global dry-run mode")
-	
-	@staticmethod
-	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):
-				self.consume_config(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.load(f)
-		
-		# Add its own path
-		config["__path"] = path
-		
-		# Consume to internal
-		self.__configs.append(config)
-		self.log("Consumed config from path:", path)
-	
-	def do_rotate(self, config):
-	
-		self.rotate_paths(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:
-			self.log(
-				"Path only has " + str(len(children)) + " items,"
-				+ " but needs " + str(max_items) + " for rotation"
-				+ "; Won't rotate this path."
-			)
-			return
-		
-		#
-		purge_count = len(children) - max_items
-		self.log(
-			"Need to purge " + str(purge_count) + " items"
-		)
-		
-		for purge_index in range(purge_count):
-			
-			#
-			item_to_purge = self.pick_item_to_purge(config, children)
-			children.remove(item_to_purge)
-			
-			#
-			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)
-			else:
-				raise Exception("Don't know how to remove this item: " + str(item_to_purge))
-	
-	@staticmethod
-	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):
-					continue
-			elif config["target-type"] == "directory":
-				if not os.path.isdir(item_path):
-					continue
-			else:
-				raise Exception("Configuration key \"target-type\" must be \"file\" or \"directory\"")
-			
-			candidates.append(item_path)
-		
-		return candidates
-	
-	@staticmethod
-	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
-			else:
-				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)
-		else:
-			self.log("Purging file:", file_path)
-			os.remove(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)
-		else:
-			self.log("Purging directory:", dir_path)
-			shutil.rmtree(dir_path)
-
-
-def main():
-	
-	rotator = BackupRotator()
-	rotator.run()
-
-
-if __name__ == "__main__":
-	main()

main.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+
+
+from BackupRotator import BackupRotator
+
+import argparse
+
+
+#
+def main():
+	
+	parser = argparse.ArgumentParser(
+		description="Mike's Backup Rotator. Helps automatically remove old backup files or folders."
+	)
+	
+	parser.add_argument(
+		"--config", "-c",
+		dest="config_files",
+		default=[],
+		action="append",
+		type=str,
+		help="Specify a configuration file. Can be called multiple times."
+	)
+	parser.add_argument(
+		"--dry-run", "-d",
+		dest="dry_run",
+		default=False,
+		action="store_true",
+		help="Only perform an analysis; Don't delete anything."
+	)
+	
+	args = parser.parse_args()
+	
+	rotator = BackupRotator()
+	rotator.run(
+		configs=args.config_files,
+		dry_run=args.dry_run
+	)
+	
+
+if __name__ == "__main__":
+	main()
+