|
@@ -12,29 +12,29 @@ import re
|
|
|
|
|
|
|
|
|
|
DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR = {
|
|
DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR = {
|
|
- 'B': 1,
|
|
+ "B": 1,
|
|
- 'kB': 10**3,
|
|
+ "kB": 10 ** 3,
|
|
- 'KB': 10**3,
|
|
+ "KB": 10 ** 3,
|
|
- 'MB': 10**6,
|
|
+ "MB": 10 ** 6,
|
|
- 'GB': 10**9,
|
|
+ "GB": 10 ** 9,
|
|
- 'TB': 10**12,
|
|
+ "TB": 10 ** 12,
|
|
- 'KiB': 2**10,
|
|
+ "KiB": 2 ** 10,
|
|
- 'MiB': 2**20,
|
|
+ "MiB": 2 ** 20,
|
|
- 'GiB': 2**30,
|
|
+ "GiB": 2 ** 30,
|
|
- 'TiB': 2**40,
|
|
+ "TiB": 2 ** 40,
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
def data_size_to_bytes(size_with_unit: str) -> int:
|
|
def data_size_to_bytes(size_with_unit: str) -> int:
|
|
- match = re.match(r'^([\d\.]+)\s*([A-Za-z]+)?$', size_with_unit)
|
|
+ match = re.match(r"^([\d\.]+)\s*([A-Za-z]+)?$", size_with_unit)
|
|
if not match:
|
|
if not match:
|
|
- raise ValueError('Unable to parse data size {!r}'.format(size_with_unit))
|
|
+ raise ValueError("Unable to parse data size {!r}".format(size_with_unit))
|
|
unit_symbol = match.group(2)
|
|
unit_symbol = match.group(2)
|
|
if unit_symbol:
|
|
if unit_symbol:
|
|
try:
|
|
try:
|
|
byte_conversion_factor = DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR[unit_symbol]
|
|
byte_conversion_factor = DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR[unit_symbol]
|
|
except KeyError:
|
|
except KeyError:
|
|
- raise ValueError('Unknown data size unit symbol {!r}'.format(unit_symbol))
|
|
+ raise ValueError("Unknown data size unit symbol {!r}".format(unit_symbol))
|
|
else:
|
|
else:
|
|
byte_conversion_factor = 1
|
|
byte_conversion_factor = 1
|
|
byte_size = float(match.group(1)) * byte_conversion_factor
|
|
byte_size = float(match.group(1)) * byte_conversion_factor
|
|
@@ -43,23 +43,31 @@ def data_size_to_bytes(size_with_unit: str) -> int:
|
|
|
|
|
|
def main():
|
|
def main():
|
|
argparser = argparse.ArgumentParser(description=__doc__)
|
|
argparser = argparse.ArgumentParser(description=__doc__)
|
|
- argparser.add_argument('-d', '--debug', action='store_true')
|
|
+ argparser.add_argument("-d", "--debug", action="store_true")
|
|
- argparser.add_argument('--free-bytes', type=data_size_to_bytes, required=True,
|
|
+ argparser.add_argument(
|
|
- help='examples: 1024, 1024B, 4KiB, 4KB, 2TB')
|
|
+ "--free-bytes",
|
|
- argparser.add_argument('root_dir_path', metavar='ROOT_DIR')
|
|
+ type=data_size_to_bytes,
|
|
|
|
+ required=True,
|
|
|
|
+ help="examples: 1024, 1024B, 4KiB, 4KB, 2TB",
|
|
|
|
+ )
|
|
|
|
+ argparser.add_argument("root_dir_path", metavar="ROOT_DIR")
|
|
args = argparser.parse_args()
|
|
args = argparser.parse_args()
|
|
- logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO,
|
|
+ logging.basicConfig(
|
|
- format='%(asctime)s:%(levelname)s:%(message)s',
|
|
+ level=logging.DEBUG if args.debug else logging.INFO,
|
|
- datefmt='%Y-%m-%dT%H:%M:%S%z')
|
|
+ format="%(asctime)s:%(levelname)s:%(message)s",
|
|
- logging.debug('Required free bytes: %d', args.free_bytes)
|
|
+ datefmt="%Y-%m-%dT%H:%M:%S%z",
|
|
|
|
+ )
|
|
|
|
+ logging.debug("Required free bytes: %d", args.free_bytes)
|
|
disk_usage = shutil.disk_usage(args.root_dir_path)
|
|
disk_usage = shutil.disk_usage(args.root_dir_path)
|
|
logging.debug(disk_usage)
|
|
logging.debug(disk_usage)
|
|
if disk_usage.free >= args.free_bytes:
|
|
if disk_usage.free >= args.free_bytes:
|
|
- logging.debug('Requirement already fulfilled')
|
|
+ logging.debug("Requirement already fulfilled")
|
|
return
|
|
return
|
|
- file_paths = [os.path.join(dirpath, filename)
|
|
+ file_paths = [
|
|
- for dirpath, _, filenames in os.walk(args.root_dir_path)
|
|
+ os.path.join(dirpath, filename)
|
|
- for filename in filenames]
|
|
+ for dirpath, _, filenames in os.walk(args.root_dir_path)
|
|
|
|
+ for filename in filenames
|
|
|
|
+ ]
|
|
file_mtime_paths = [(os.stat(p).st_mtime, p) for p in file_paths]
|
|
file_mtime_paths = [(os.stat(p).st_mtime, p) for p in file_paths]
|
|
file_mtime_paths.sort()
|
|
file_mtime_paths.sort()
|
|
removed_files_counter = 0
|
|
removed_files_counter = 0
|
|
@@ -68,15 +76,18 @@ def main():
|
|
if shutil.disk_usage(args.root_dir_path).free >= args.free_bytes:
|
|
if shutil.disk_usage(args.root_dir_path).free >= args.free_bytes:
|
|
break
|
|
break
|
|
os.remove(file_path)
|
|
os.remove(file_path)
|
|
- logging.debug('Removed file %s', file_path)
|
|
+ logging.debug("Removed file %s", file_path)
|
|
removed_files_counter += 1
|
|
removed_files_counter += 1
|
|
last_mtime = file_mtime
|
|
last_mtime = file_mtime
|
|
if removed_files_counter == 0:
|
|
if removed_files_counter == 0:
|
|
- logging.warning('No files to remove')
|
|
+ logging.warning("No files to remove")
|
|
else:
|
|
else:
|
|
- logging.info('Removed %d file(s) with modification date <= %sZ', removed_files_counter,
|
|
+ logging.info(
|
|
- datetime.datetime.utcfromtimestamp(last_mtime).isoformat('T'))
|
|
+ "Removed %d file(s) with modification date <= %sZ",
|
|
|
|
+ removed_files_counter,
|
|
|
|
+ datetime.datetime.utcfromtimestamp(last_mtime).isoformat("T"),
|
|
|
|
+ )
|
|
|
|
|
|
|
|
|
|
-if __name__ == '__main__':
|
|
+if __name__ == "__main__":
|
|
main()
|
|
main()
|