Coverage for src/photo_cleanup_utils.py: 100%

1"""Photo file cleanup utilities module. 

2 

3This module contains utilities for cleaning up obsolete photo files 

4that are no longer on the server. 

5""" 

6 

7___author___ = "Mandar Patil <mandarons@pm.me>" 

8 

9from pathlib import Path 

10 

11from src import DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT, get_logger 

12 

13LOGGER = get_logger() 

14 

15 

16def remove_obsolete_files( 

17 destination_path: str | None, 

18 tracked_files: set[str] | None, 

19 exclude_filenames: set[str] | None = None, 

20 limit_percent: int = DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT, 

21 recycle_bin=None, 

22) -> set[str]: 

23 """Remove local obsolete files that are no longer on server. 

24 

25 Args: 

26 destination_path: Path to search for obsolete files 

27 tracked_files: Set of files that should be kept (files on server) 

28 exclude_filenames: Set of filenames (basename only) that must never 

29 be removed even if they are not in ``tracked_files``. Used to 

30 protect the mount-marker sentinel file from cleanup. 

31 recycle_bin: ``RecycleBin`` to move removed files to instead of 

32 deleting them. The delete limit applies the same either way. 

33 

34 Returns: 

35 Set of paths that were removed 

36 """ 

37 removed_paths: set[str] = set() 

38 

39 if not (destination_path and tracked_files is not None): 

40 return removed_paths 

41 if recycle_bin: 

42 recycle_bin.purge() 

43 

44 # Decide everything before deleting anything. Cleanup is the only 

45 # destructive step in a sync, and it infers deletions from the absence 

46 # of a path in ``tracked_files`` -- so any bug that leaves paths 

47 # untracked reads as "the server no longer has these" and is acted on 

48 # at full speed. Counting first makes that recoverable. 

49 candidates = [] 

50 present = 0 

51 for path in Path(destination_path).rglob("*"): 

52 if not path.is_file(): 

53 continue 

54 local_file = str(path.absolute()) 

55 if recycle_bin and recycle_bin.contains(local_file): 

56 continue # what is in the bin is not part of the library 

57 present += 1 

58 if local_file in tracked_files: 

59 continue 

60 if exclude_filenames and path.name in exclude_filenames: 

61 continue 

62 candidates.append((path, local_file)) 

63 

64 if not candidates: 

65 return removed_paths 

66 

67 if limit_percent > 0 and present: 

68 delete_percent = 100 * len(candidates) / present 

69 if delete_percent > limit_percent: 

70 LOGGER.error( 

71 f"Refusing to remove {len(candidates)} of {present} files " 

72 f"({delete_percent:.0f}%) under {destination_path} — over the " 

73 f"{limit_percent}% obsolete-delete limit. A sync does not " 

74 f"normally make this share of a library obsolete, so this is " 

75 f"treated as a fault rather than acted on. Check the library " 

76 f"is fully enumerated and the destination is correct, then " 

77 f"raise photos.obsolete_delete_limit_percent (0 disables the " 

78 f"limit) if the deletions are genuinely wanted.", 

79 ) 

80 return removed_paths 

81 

82 for path, local_file in candidates: 

83 if recycle_bin: 

84 LOGGER.info(f"Moving {local_file} to the recycle bin ...") 

85 recycle_bin.discard(local_file, destination_path) 

86 else: 

87 LOGGER.info(f"Removing {local_file} ...") 

88 path.unlink(missing_ok=True) 

89 removed_paths.add(local_file) 

90 

91 return removed_paths