__init__.py 2.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. """
  2. Delete file with the oldest modification date
  3. until a minimum of --free-bytes are available on the respective disk.
  4. """
  5. import argparse
  6. import datetime
  7. import logging
  8. import os
  9. import shutil
  10. import re
  11. # https://en.wikipedia.org/wiki/Template:Quantities_of_bytes
  12. _DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR = {
  13. "B": 1,
  14. "kB": 10 ** 3,
  15. "KB": 10 ** 3,
  16. "MB": 10 ** 6,
  17. "GB": 10 ** 9,
  18. "TB": 10 ** 12,
  19. "KiB": 2 ** 10,
  20. "MiB": 2 ** 20,
  21. "GiB": 2 ** 30,
  22. "TiB": 2 ** 40,
  23. }
  24. def _data_size_to_bytes(size_with_unit: str) -> int:
  25. match = re.match(r"^([\d\.]+)\s*([A-Za-z]+)?$", size_with_unit)
  26. if not match:
  27. raise ValueError("Unable to parse data size {!r}".format(size_with_unit))
  28. unit_symbol = match.group(2)
  29. if unit_symbol:
  30. try:
  31. byte_conversion_factor = _DATA_SIZE_UNIT_BYTE_CONVERSION_FACTOR[unit_symbol]
  32. except KeyError as exc:
  33. raise ValueError(
  34. "Unknown data size unit symbol {!r}".format(unit_symbol)
  35. ) from exc
  36. else:
  37. byte_conversion_factor = 1
  38. byte_size = float(match.group(1)) * byte_conversion_factor
  39. return int(round(byte_size, 0))
  40. def _main():
  41. argparser = argparse.ArgumentParser(description=__doc__)
  42. argparser.add_argument("-d", "--debug", action="store_true")
  43. argparser.add_argument(
  44. "--free-bytes",
  45. type=_data_size_to_bytes,
  46. required=True,
  47. help="examples: 1024, 1024B, 4KiB, 4KB, 2TB",
  48. )
  49. argparser.add_argument("root_dir_path", metavar="ROOT_DIR")
  50. args = argparser.parse_args()
  51. logging.basicConfig(
  52. level=logging.DEBUG if args.debug else logging.INFO,
  53. format="%(asctime)s:%(levelname)s:%(message)s",
  54. datefmt="%Y-%m-%dT%H:%M:%S%z",
  55. )
  56. logging.debug("Required free bytes: %d", args.free_bytes)
  57. disk_usage = shutil.disk_usage(args.root_dir_path)
  58. logging.debug(disk_usage)
  59. if disk_usage.free >= args.free_bytes:
  60. logging.debug("Requirement already fulfilled")
  61. return
  62. file_paths = [
  63. os.path.join(dirpath, filename)
  64. for dirpath, _, filenames in os.walk(args.root_dir_path)
  65. for filename in filenames
  66. ]
  67. file_mtime_paths = [(os.stat(p).st_mtime, p) for p in file_paths]
  68. file_mtime_paths.sort()
  69. removed_files_counter = 0
  70. last_mtime = None
  71. for file_mtime, file_path in file_mtime_paths:
  72. if shutil.disk_usage(args.root_dir_path).free >= args.free_bytes:
  73. break
  74. os.remove(file_path)
  75. logging.debug("Removed file %s", file_path)
  76. removed_files_counter += 1
  77. last_mtime = file_mtime
  78. if removed_files_counter == 0:
  79. logging.warning("No files to remove")
  80. else:
  81. logging.info(
  82. "Removed %d file(s) with modification date <= %sZ",
  83. removed_files_counter,
  84. datetime.datetime.utcfromtimestamp(last_mtime).isoformat("T"),
  85. )
  86. if __name__ == "__main__":
  87. _main()