Hilarious oversights

.python-version

@@ -0,0 +1 @@
+3.10.10

BackupRotator.py

@@ -6,7 +6,7 @@ Mike's Backup Rotator
 
 A simple script to help automatically rotate backup files
 
-Copyright 2019 Mike Peralta; All rights reserved
+Copyright 2022 Mike Peralta; All rights reserved
 
 Released under the GNU GENERAL PUBLIC LICENSE v3 (See LICENSE file for more)
 
@@ -14,9 +14,11 @@ Released under the GNU GENERAL PUBLIC LICENSE v3 (See LICENSE file for more)
 
 import datetime
 import os
+# import pprint
 import shutil
 import sys
 import syslog
+import time
 import yaml
 
 
@@ -29,11 +31,14 @@ class BackupRotator:
 		self.__config_paths = []
 		self.__calculated_actions = []
 	
-	def run(self):
+	def run(self, configs, dry_run: bool = False):
 		
 		self.log("Begin")
-		self.consume_arguments()
-		self.consume_configs(self.__config_paths)
+		
+		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)):
@@ -43,7 +48,7 @@ class BackupRotator:
 			
 			#
 			self.log("Rotating for config " + str(config_index + 1) + " of " + str(len(self.__configs)), config["__path"])
-			self.do_rotate(config)
+			self._do_rotate(config)
 	
 	@staticmethod
 	def current_time():
@@ -64,52 +69,26 @@ class BackupRotator:
 		
 		print(to_log)
 	
-	def consume_arguments(self):
+	def _consume_configs(self, paths: list=None):
 		
-		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)
-				self.log("Found config path argument:", one_path)
-			
-			elif arg == "--dry-run":
-				self.__dry_run = True
-				self.log("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")
+		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)
+				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))
+					self._consume_config(os.path.join(path, file_name))
 				
-	def consume_config(self, path: str):
+	def _consume_config(self, path: str):
 		
 		# Open the file
 		f = open(path)
@@ -126,60 +105,136 @@ class BackupRotator:
 		self.__configs.append(config)
 		self.log("Consumed config from path:", path)
 	
-	def do_rotate(self, config):
+	def _do_rotate(self, config):
 	
-		self.rotate_paths(config)
+		self._rotate_paths(config)
 	
-	def rotate_paths(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)
+			self._rotate_path(config, path)
 	
-	def rotate_path(self, config, path):
+	def _rotate_path(self, config, path):
 		
-		self.log("Rotating path", path)
+		assert os.path.isdir(path), "Path should be a directory: {}".format(path)
 		
-		if "maximum-items" not in config:
-			raise Exception("Please provide config key: \"maximum-items\"")
-		max_items = config["maximum-items"]
+		self.log("Rotating path: {}".format(path))
 		
-		if not os.path.isdir(path):
-			raise Exception("Path should be a directory:" + str(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):
 		
-		children = self.gather_rotation_candidates(config, path)
+		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) <= max_items:
-			self.log(
-				"Path only has " + str(len(children)) + " items,"
-				+ " but needs " + str(max_items) + " for rotation"
-				+ "; Won't rotate this path."
-			)
+		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
+			))
 			return
+		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(
-			"Need to purge " + str(purge_count) + " 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 = self.pick_item_to_purge(config, children)
+			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
+			))
 			
 			#
-			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))
+			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, child_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)
+		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
+			))
+			return
+		
+		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):
+	def _gather_rotation_candidates(config, path):
 		
 		candidates = []
 		
@@ -203,28 +258,97 @@ class BackupRotator:
 		
 		return candidates
 	
-	@staticmethod
-	def pick_item_to_purge(config, items):
+	def _pick_oldest_item(self, 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))
+			ctime = self._detect_item_date(config, item)
+			if best_ctime is None or ctime < best_ctime:
+				best_ctime = ctime
+				best_item = item
 		
-		return best_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
 	
-	def remove_file(self, config, file_path):
+	@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))
@@ -237,7 +361,7 @@ class BackupRotator:
 			self.log("Purging file:", file_path)
 			os.remove(file_path)
 	
-	def remove_directory(self, config, dir_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))
@@ -249,3 +373,16 @@ class BackupRotator:
 		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

@@ -4,10 +4,11 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-pyyaml = "*"
+pyyaml = ">=5.4"
 
 [dev-packages]
 
 [requires]
-python_version = "3"
+python_version = "3.10.10"
+
 

Pipfile.lock

@@ -1,11 +1,11 @@
 {
     "_meta": {
         "hash": {
-            "sha256": "25d0d6f494ae55bc9fa72711e1e8edb7e3d6b7ce72e6254d2ec85bd3b0e637bd"
+            "sha256": "cceb18d3baeb19edef3ba31b743720003102c4c3d9cddd6b595c664692a37384"
         },
         "pipfile-spec": 6,
         "requires": {
-            "python_version": "3"
+            "python_version": "3.10.5"
         },
         "sources": [
             {
@@ -18,22 +18,42 @@
     "default": {
         "pyyaml": {
             "hashes": [
-                "sha256:06a0d7ba600ce0b2d2fe2e78453a470b5a6e000a985dd4a4e54e436cc36b0e97",
-                "sha256:240097ff019d7c70a4922b6869d8a86407758333f02203e0fc6ff79c5dcede76",
-                "sha256:4f4b913ca1a7319b33cfb1369e91e50354d6f07a135f3b901aca02aa95940bd2",
-                "sha256:6034f55dab5fea9e53f436aa68fa3ace2634918e8b5994d82f3621c04ff5ed2e",
-                "sha256:69f00dca373f240f842b2931fb2c7e14ddbacd1397d57157a9b005a6a9942648",
-                "sha256:73f099454b799e05e5ab51423c7bcf361c58d3206fa7b0d555426b1f4d9a3eaf",
-                "sha256:74809a57b329d6cc0fdccee6318f44b9b8649961fa73144a98735b0aaf029f1f",
-                "sha256:7739fc0fa8205b3ee8808aea45e968bc90082c10aef6ea95e855e10abf4a37b2",
-                "sha256:95f71d2af0ff4227885f7a6605c37fd53d3a106fcab511b8860ecca9fcf400ee",
-                "sha256:ad9c67312c84def58f3c04504727ca879cb0013b2517c85a9a253f0cb6380c0a",
-                "sha256:b8eac752c5e14d3eca0e6dd9199cd627518cb5ec06add0de9d32baeee6fe645d",
-                "sha256:cc8955cfbfc7a115fa81d85284ee61147059a753344bc51098f3ccd69b0d7e0c",
-                "sha256:d13155f591e6fcc1ec3b30685d50bf0711574e2c0dfffd7644babf8b5102ca1a"
+                "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": "==5.3.1"
+            "version": "==6.0"
         }
     },
     "develop": {}

README.md

@@ -78,11 +78,25 @@ Specifies the method used when attempting to determine how old a backup file/dir
 
 Currently, only *file* is supported
 
+### minimum-items < INTEGER >
+
+Specifies the minimum number of backup files/dirs that must be present before rotating can happen. Should be an integer.
+
+This option doesn't specify how much to rotate on its own, but when rotation is possible. It should probably be used with maximum-age or something other than maximum-items.
+
+For example, when the *minimum-items* value is set to 5, and *target-type* is *file*, the program will not rotate any files until there are at least 5 in the target directory.
+
 ### maximum-items < INTEGER >
 
 Specifies the maximum number of backup files/dirs that are allowed in a path before rotating will happen. Should be an integer.
 
-For example, when the *maximum-items* value is set to 5, and *target-type* is *file*, the program will not rotate any files until there are at least 5 in the target directory.
+For example, when the *maximum-items* value is set to 500, and *target-type* is *file*, the program will not rotate any files until there are at least 500 in the target directory.
+
+### maximum-age < INTEGER >
+
+Specifies the maximum age (in days) of backup files/dirs that are allowed in a path before rotating will happen. Should be an integer.
+
+For example, when the *maximum-age* value is set to 30, and *target-type* is *file*, the program will not rotate any files that are newer than 30 days.
 
 ### paths < Array of Paths >
 

main.py

@@ -3,13 +3,41 @@
 
 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()
-
+	rotator.run(
+		configs=args.config_files,
+		dry_run=args.dry_run
+	)
+	
 
 if __name__ == "__main__":
 	main()
+