free_disk.py 2.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  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:
  33. raise ValueError('Unknown data size unit symbol {!r}'.format(unit_symbol))
  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():
  39. argparser = argparse.ArgumentParser(description=__doc__)
  40. argparser.add_argument('-d', '--debug', action='store_true')
  41. argparser.add_argument('--free-bytes', type=data_size_to_bytes, required=True,
  42. help='examples: 1024, 1024B, 4KiB, 4KB, 2TB')
  43. argparser.add_argument('root_dir_path', metavar='ROOT_DIR')
  44. args = argparser.parse_args()
  45. logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO,
  46. format='%(asctime)s:%(levelname)s:%(message)s',
  47. datefmt='%Y-%m-%dT%H:%M:%S%z')
  48. logging.debug('Required free bytes: %d', args.free_bytes)
  49. disk_usage = shutil.disk_usage(args.root_dir_path)
  50. logging.debug(disk_usage)
  51. if disk_usage.free >= args.free_bytes:
  52. logging.debug('Requirement already fulfilled')
  53. return
  54. file_paths = [os.path.join(dirpath, filename)
  55. for dirpath, _, filenames in os.walk(args.root_dir_path)
  56. for filename in filenames]
  57. file_mtime_paths = [(os.stat(p).st_mtime, p) for p in file_paths]
  58. file_mtime_paths.sort()
  59. removed_files_counter = 0
  60. last_mtime = None
  61. for file_mtime, file_path in file_mtime_paths:
  62. if shutil.disk_usage(args.root_dir_path).free >= args.free_bytes:
  63. break
  64. os.remove(file_path)
  65. logging.debug('Removed file %s', file_path)
  66. removed_files_counter += 1
  67. last_mtime = file_mtime
  68. if removed_files_counter == 0:
  69. logging.warning('No files to remove')
  70. else:
  71. logging.info('Removed %d file(s) with modification date <= %s', removed_files_counter,
  72. datetime.datetime.utcfromtimestamp(last_mtime).isoformat('T'))
  73. if __name__ == '__main__':
  74. main()