Coverage for puhdistusalue/cli.py: 61.00%

1#! /usr/bin/env python 

2# -*- coding: utf-8 -*- 

3"""Purge monotonically named files in folders keeping range endpoints. 

4 

5Implementation uses sha256 hashes for identity and assumes that 

6the natural order relates to the notion of fresher or better. 

7""" 

8import datetime as dti 

9import os 

10import sys 

11import typing 

12 

13from puristaa.puristaa import prefix_compression # type: ignore 

14 

15from puhdistusalue.puhdistusalue import read_folder, triage_hashes 

16 

17DEBUG = os.getenv('PURGE_RANGE_DEBUG') 

18 

19 

20@typing.no_type_check 

21def humanize_mass(total_less_bytes: int): 

22 """DRY""" 

23 if total_less_bytes >= 1e9: 

24 return f'{round(total_less_bytes / 1024 / 1024 / 1024, 3) :.3f}', 'total gigabytes' 

25 if total_less_bytes >= 1e6: 

26 return f'{round(total_less_bytes / 1024 / 1024, 3) :.3f}', 'total megabytes' 

27 if total_less_bytes >= 1e3: 

28 return f'{round(total_less_bytes / 1024, 3) :.3f}', 'total kilobytes' 

29 

30 return f'{total_less_bytes :d}', 'total bytes' 

31 

32 

33@typing.no_type_check 

34def humanize_duration(duration_seconds: float): 

35 """DRY""" 

36 if duration_seconds >= 3600: 

37 return f'{round(duration_seconds / 60 / 60, 3) :.3f}', 'hours' 

38 if duration_seconds >= 60: 

39 return f'{round(duration_seconds / 60, 3) :.3f}', 'minutes' 

40 if duration_seconds >= 1: 

41 return f'{round(duration_seconds, 3) :.3f}', 'seconds' 

42 

43 return f'{round(duration_seconds * 1e3, 3) :.3f}', 'millis' 

44 

45 

46# pylint: disable=expression-not-assigned 

47@typing.no_type_check 

48def main(argv=None): 

49 """Process the files separately per folder.""" 

50 start_time = dti.datetime.now(dti.UTC) 

51 argv = sys.argv[1:] if argv is None else argv 

52 verbose = bool('-v' in argv or '--verbose' in argv) 

53 human = bool('-H' in argv or '--human' in argv) 

54 folder_paths = [entry for entry in argv if entry.strip() and entry not in ('-H', '--human', '-v', '--verbose')] 

55 total_removed, total_less_bytes = 0, 0 

56 for a_path in folder_paths: 

57 hash_map = {} 

58 try: 

59 hash_map = read_folder(a_path) 

60 except FileNotFoundError as err: 

61 print(f'WARNING: Skipping non-existing path ({a_path}) -> "{err}"') 

62 if hash_map: 

63 keep_these, remove_those = triage_hashes(hash_map) 

64 for this in keep_these: 

65 DEBUG and print(f'KEEP file {this}') 

66 folder_removed, folder_less_bytes = 0, 0 

67 for that in remove_those: 67 ↛ 68line 67 didn't jump to line 68, because the loop on line 67 never started

68 DEBUG and print(f'REMOVE file {that}') 

69 target = os.path.join(a_path, that) 

70 folder_less_bytes += os.path.getsize(target) 

71 os.remove(target) 

72 folder_removed += 1 

73 

74 if verbose: 74 ↛ 75line 74 didn't jump to line 75, because the condition on line 74 was never true

75 print( 

76 f'removed {folder_removed} redundant objects or {folder_less_bytes}' 

77 f' combined bytes from folder at {a_path}' 

78 ) 

79 total_less_bytes += folder_less_bytes 

80 total_removed += folder_removed 

81 

82 prefix, rel_paths = prefix_compression(folder_paths, policy=lambda x: x == '/') 

83 if len(rel_paths) > 5: 83 ↛ 84line 83 didn't jump to line 84, because the condition on line 83 was never true

84 folders_disp = f"{prefix}[{', '.join(rel_paths[:3])}, ... {rel_paths[-1]}]" 

85 else: 

86 folders_disp = f'{folder_paths}' if folder_paths else '[<EMPTY>]' 

87 

88 duration_seconds = (dti.datetime.now(dti.UTC) - start_time).total_seconds() 

89 if human: 89 ↛ 90line 89 didn't jump to line 90, because the condition on line 89 was never true

90 m_quantity, m_unit = humanize_mass(total_less_bytes) 

91 d_quantity, d_unit = humanize_duration(duration_seconds) 

92 else: 

93 m_quantity, m_unit = f'{total_less_bytes :d}', 'total bytes' 

94 d_quantity, d_unit = f'{round(duration_seconds, 3) :.3f}', 'seconds' 

95 

96 print( 

97 f'removed {total_removed} total redundant objects or' 

98 f' {m_quantity} {m_unit} from folders at {folders_disp}' 

99 f' in {d_quantity} {d_unit}' 

100 )