__init__.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  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 re
  10. import shutil
  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(f"Unable to parse data size {size_with_unit!r}")
  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(f"Unknown data size unit symbol {unit_symbol!r}") from exc
  34. else:
  35. byte_conversion_factor = 1
  36. byte_size = float(match.group(1)) * byte_conversion_factor
  37. return int(round(byte_size, 0))
  38. def _main() -> None:
  39. argparser = argparse.ArgumentParser(description=__doc__)
  40. argparser.add_argument("-d", "--debug", action="store_true")
  41. argparser.add_argument(
  42. "--free-bytes",
  43. type=_data_size_to_bytes,
  44. required=True,
  45. help="examples: 1024, 1024B, 4KiB, 4KB, 2TB",
  46. )
  47. argparser.add_argument("root_dir_path", metavar="ROOT_DIR")
  48. args = argparser.parse_args()
  49. logging.basicConfig(
  50. level=logging.DEBUG if args.debug else logging.INFO,
  51. format="%(asctime)s:%(levelname)s:%(message)s",
  52. datefmt="%Y-%m-%dT%H:%M:%S%z",
  53. )
  54. logging.debug("Required free bytes: %d", args.free_bytes)
  55. disk_usage = shutil.disk_usage(args.root_dir_path)
  56. logging.debug(disk_usage)
  57. if disk_usage.free >= args.free_bytes:
  58. logging.debug("Requirement already fulfilled")
  59. return
  60. file_paths = [
  61. os.path.join(dirpath, filename)
  62. for dirpath, _, filenames in os.walk(args.root_dir_path)
  63. for filename in filenames
  64. ]
  65. file_mtime_paths = [(os.stat(p).st_mtime, p) for p in file_paths]
  66. file_mtime_paths.sort()
  67. removed_files_counter = 0
  68. last_mtime = None
  69. for file_mtime, file_path in file_mtime_paths:
  70. if shutil.disk_usage(args.root_dir_path).free >= args.free_bytes:
  71. break
  72. os.remove(file_path)
  73. logging.debug("Removed file %s", file_path)
  74. removed_files_counter += 1
  75. last_mtime = file_mtime
  76. if removed_files_counter == 0:
  77. logging.warning("No files to remove")
  78. else:
  79. assert last_mtime is not None # for mypy
  80. logging.info(
  81. "Removed %d file(s) with modification date <= %sZ",
  82. removed_files_counter,
  83. datetime.datetime.utcfromtimestamp(last_mtime).isoformat("T"),
  84. )