Coverage for src/album_sync_orchestrator.py: 100%

1"""Album synchronization orchestration module. 

2 

3This module contains the main album sync orchestration logic 

4that coordinates photo filtering, download collection, and parallel execution. 

5""" 

6 

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

8 

9import os 

10 

11from src import get_logger 

12from src.hardlink_registry import HardlinkRegistry 

13from src.photo_download_manager import ( 

14 collect_download_task, 

15 execute_parallel_downloads, 

16) 

17from src.photo_filter_utils import is_photo_wanted 

18from src.photo_path_utils import normalize_file_path 

19 

20LOGGER = get_logger() 

21 

22 

23def sync_album_photos( 

24 album, 

25 destination_path: str, 

26 file_sizes: list[str], 

27 extensions: list[str] | None = None, 

28 files: set[str] | None = None, 

29 folder_format: str | None = None, 

30 hardlink_registry: HardlinkRegistry | None = None, 

31 config=None, 

32) -> bool | None: 

33 """Sync photos from given album. 

34 

35 This function orchestrates the synchronization of a single album by: 

36 1. Creating the destination directory 

37 2. Collecting download tasks for wanted photos 

38 3. Executing downloads in parallel 

39 4. Recursively syncing subalbums 

40 

41 Args: 

42 album: Album object from iCloudPy 

43 destination_path: Path where photos should be saved 

44 file_sizes: List of file size variants to download 

45 extensions: List of allowed file extensions (None = all allowed) 

46 files: Set to track downloaded files 

47 folder_format: strftime format string for folder organization 

48 hardlink_registry: Registry for tracking downloaded files for hardlinks 

49 config: Configuration dictionary 

50 

51 Returns: 

52 True on success, None on invalid input 

53 """ 

54 if album is None or destination_path is None or file_sizes is None: 

55 return None 

56 

57 # Create destination directory with normalized path 

58 normalized_destination = normalize_file_path(destination_path) 

59 os.makedirs(normalized_destination, exist_ok=True) 

60 LOGGER.info(f"Syncing {album.title}") 

61 

62 # Collect download tasks for photos 

63 download_tasks = _collect_album_download_tasks( 

64 album, 

65 normalized_destination, 

66 file_sizes, 

67 extensions, 

68 files, 

69 folder_format, 

70 hardlink_registry, 

71 ) 

72 

73 # Execute downloads in parallel if there are tasks 

74 if download_tasks: 

75 execute_parallel_downloads(download_tasks, config) 

76 

77 # Recursively sync subalbums 

78 _sync_subalbums( 

79 album, 

80 normalized_destination, 

81 file_sizes, 

82 extensions, 

83 files, 

84 folder_format, 

85 hardlink_registry, 

86 config, 

87 ) 

88 

89 return True 

90 

91 

92def _collect_album_download_tasks( 

93 album, 

94 destination_path: str, 

95 file_sizes: list[str], 

96 extensions: list[str] | None, 

97 files: set[str] | None, 

98 folder_format: str | None, 

99 hardlink_registry: HardlinkRegistry | None, 

100) -> list: 

101 """Collect download tasks for all photos in an album. 

102 

103 Args: 

104 album: Album object from iCloudPy 

105 destination_path: Path where photos should be saved 

106 file_sizes: List of file size variants to download 

107 extensions: List of allowed file extensions 

108 files: Set to track downloaded files 

109 folder_format: strftime format string for folder organization 

110 hardlink_registry: Registry for tracking downloaded files 

111 

112 Returns: 

113 List of download tasks to execute 

114 """ 

115 download_tasks = [] 

116 

117 for photo in album: 

118 if is_photo_wanted(photo, extensions): 

119 for file_size in file_sizes: 

120 download_info = collect_download_task( 

121 photo, 

122 file_size, 

123 destination_path, 

124 files, 

125 folder_format, 

126 hardlink_registry, 

127 ) 

128 if download_info: 

129 download_tasks.append(download_info) 

130 else: 

131 LOGGER.debug(f"Skipping the unwanted photo {photo.filename}.") 

132 

133 return download_tasks 

134 

135 

136def _sync_subalbums( 

137 album, 

138 destination_path: str, 

139 file_sizes: list[str], 

140 extensions: list[str] | None, 

141 files: set[str] | None, 

142 folder_format: str | None, 

143 hardlink_registry: HardlinkRegistry | None, 

144 config, 

145) -> None: 

146 """Recursively sync all subalbums. 

147 

148 Args: 

149 album: Album object from iCloudPy 

150 destination_path: Base path where subalbums should be created 

151 file_sizes: List of file size variants to download 

152 extensions: List of allowed file extensions 

153 files: Set to track downloaded files 

154 folder_format: strftime format string for folder organization 

155 hardlink_registry: Registry for tracking downloaded files 

156 config: Configuration dictionary 

157 """ 

158 for subalbum in album.subalbums: 

159 sync_album_photos( 

160 album.subalbums[subalbum], 

161 os.path.join(destination_path, subalbum), 

162 file_sizes, 

163 extensions, 

164 files, 

165 folder_format, 

166 hardlink_registry, 

167 config, 

168 )