Coverage for src/recycle_bin.py: 100%

1"""Recycle bin for obsolete-file cleanup. 

2 

3When ``app.recycle_bin.enabled`` is set, files and folders that obsolete-file 

4cleanup would delete are moved to ``<app.root>/Recently Deleted/<service>/ 

5<YYYY-MM-DD>/<path relative to the service destination>`` instead (with the 

6library's folder first when Photos libraries have their own destinations), and kept 

7for ``app.recycle_bin.retention_days`` (or forever when that is unset). 

8 

9The bin only changes *how* cleanup removes something, never *what* it 

10removes: with ``remove_obsolete`` off, nothing is cleaned up and nothing 

11reaches the bin. 

12""" 

13 

14import copy 

15import datetime 

16import os 

17import shutil 

18from pathlib import Path 

19 

20from src import get_logger 

21 

22LOGGER = get_logger() 

23 

24BIN_DIRECTORY_NAME = "Recently Deleted" 

25 

26 

27class RecycleBin: 

28 """Where one service's cleanup puts what it removes.""" 

29 

30 def __init__( 

31 self, 

32 root: str, 

33 service: str, 

34 retention_days: int | None = None, 

35 today: datetime.date | None = None, 

36 ): 

37 """Create the bin for ``service`` under ``root``. 

38 

39 Args: 

40 root: ``app.root``, the parent of every service destination 

41 service: ``drive`` or ``photos``; each gets its own folder 

42 retention_days: Days to keep a day's folder; None keeps forever 

43 today: Override for tests 

44 """ 

45 self.path = os.path.abspath(os.path.join(root, BIN_DIRECTORY_NAME, service)) 

46 self.retention_days = retention_days 

47 self.today = today or datetime.date.today() 

48 self.subfolder = "" 

49 

50 def within(self, subfolder: str) -> "RecycleBin": 

51 """The same bin, filing what it receives under ``subfolder`` in each day. 

52 

53 Several destinations can share one service's bin -- each Photos 

54 library in ``library_destinations`` has its own -- and the same 

55 relative path in two of them must land in two places, or nothing 

56 says where either belongs. 

57 """ 

58 scoped = copy.copy(self) 

59 scoped.subfolder = subfolder 

60 return scoped 

61 

62 def contains(self, path: str) -> bool: 

63 """True if ``path`` is the bin or inside it. 

64 

65 A destination can be ``app.root`` itself, so cleanup walks would 

66 otherwise find the bin and treat everything in it as obsolete. 

67 """ 

68 absolute = os.path.abspath(path) 

69 top = os.path.dirname(self.path) 

70 return absolute == top or absolute.startswith(top + os.sep) 

71 

72 def discard(self, path: str, destination_path: str) -> str: 

73 """Move ``path`` into today's folder, keeping its place in the tree. 

74 

75 Args: 

76 path: File or folder that cleanup would delete 

77 destination_path: The service destination ``path`` lives under 

78 

79 Returns: 

80 Where it was moved to 

81 """ 

82 relative = os.path.relpath( 

83 os.path.abspath(path), os.path.abspath(destination_path), 

84 ) 

85 day = os.path.join(self.path, self.today.isoformat()) 

86 target = os.path.normpath(os.path.join(day, self.subfolder, relative)) 

87 if not target.startswith(day + os.sep): 

88 msg = f"Recycle bin: {path} would be filed outside the bin, at {target}" 

89 raise ValueError(msg) 

90 os.makedirs(os.path.dirname(target), exist_ok=True) 

91 # The same path can be removed twice in a day (re-downloaded, then 

92 # dropped again); keep both rather than overwrite the first. The 

93 # number goes before a file's extension, so renaming it back to the 

94 # original name is all a restore takes. 

95 stem, extension = (target, "") if os.path.isdir(path) else os.path.splitext(target) 

96 candidate, n = target, 2 

97 while os.path.lexists(candidate): 

98 candidate = f"{stem} ({n}){extension}" 

99 n += 1 

100 shutil.move(path, candidate) 

101 return candidate 

102 

103 def purge(self) -> int: 

104 """Delete day folders older than the retention period. 

105 

106 Returns: 

107 Number of day folders deleted 

108 """ 

109 if self.retention_days is None or not os.path.isdir(self.path): 

110 return 0 

111 cutoff = self.today - datetime.timedelta(days=self.retention_days) 

112 purged = 0 

113 for day in sorted(Path(self.path).iterdir()): 

114 try: 

115 day_date = datetime.date.fromisoformat(day.name) 

116 except ValueError: 

117 continue # not a folder the bin made; leave it alone 

118 if day.is_dir() and day_date < cutoff: 

119 LOGGER.info( 

120 f"Recycle bin: deleting {day}, older than {self.retention_days} days ...", 

121 ) 

122 shutil.rmtree(day) 

123 purged += 1 

124 return purged 

125 

126 

127def from_config(config: dict | None, service: str) -> RecycleBin | None: 

128 """The bin for ``service`` if ``app.recycle_bin.enabled``, else None.""" 

129 from src import config_parser 

130 

131 if not config or not config_parser.get_recycle_bin_enabled(config=config): 

132 return None 

133 return RecycleBin( 

134 root=config_parser.get_root_destination_path(config=config), 

135 service=service, 

136 retention_days=config_parser.get_recycle_bin_retention_days(config=config), 

137 )