Coverage for src/drive_sync_directory.py: 100%

1"""Drive sync directory orchestration. 

2 

3This module provides the main sync directory coordination functionality, 

4orchestrating folder processing, file collection, and parallel downloads per SRP. 

5""" 

6 

7__author__ = "Mandar Patil (mandarons@pm.me)" 

8 

9import unicodedata 

10from typing import Any 

11 

12from src import configure_icloudpy_logging, get_logger, recycle_bin, sync_drive 

13from src.drive_cleanup import remove_obsolete 

14from src.drive_filtering import wanted_parent_folder 

15from src.drive_folder_processing import process_folder 

16from src.drive_parallel_download import collect_file_for_download, execute_parallel_downloads 

17 

18# Configure icloudpy logging immediately after import 

19configure_icloudpy_logging() 

20 

21LOGGER = get_logger() 

22 

23 

24def sync_directory( 

25 drive: Any, 

26 destination_path: str, 

27 items: Any, 

28 root: str, 

29 top: bool = True, 

30 filters: dict[str, list[str]] | None = None, 

31 ignore: list[str] | None = None, 

32 remove: bool = False, 

33 config: Any | None = None, 

34) -> set[str]: 

35 """Synchronize a directory from iCloud Drive to local filesystem. 

36 

37 This function orchestrates the entire sync process by: 

38 1. Processing folders and recursively syncing subdirectories 

39 2. Collecting files for parallel download 

40 3. Executing parallel downloads 

41 4. Cleaning up obsolete files if requested 

42 

43 Args: 

44 drive: iCloud drive service instance 

45 destination_path: Local destination directory 

46 items: iCloud items to process 

47 root: Root directory for relative path calculations 

48 top: Whether this is the top-level sync call 

49 filters: Dictionary of filters (folders, file_extensions) 

50 ignore: List of ignore patterns 

51 remove: Whether to remove obsolete local files 

52 config: Configuration object 

53 

54 Returns: 

55 Set of all processed file paths 

56 """ 

57 files = set() 

58 download_tasks = [] 

59 

60 if not (drive and destination_path and items and root): 

61 return files 

62 

63 # First pass: process folders and collect download tasks 

64 for i in items: 

65 item = drive[i] 

66 

67 if item.type in ("folder", "app_library"): 

68 _process_folder_item( 

69 item, 

70 destination_path, 

71 filters, 

72 ignore, 

73 root, 

74 files, 

75 config, 

76 ) 

77 elif item.type == "file": 

78 _process_file_item( 

79 item, 

80 destination_path, 

81 filters, 

82 ignore, 

83 root, 

84 files, 

85 download_tasks, 

86 config, 

87 ) 

88 

89 # Second pass: execute downloads in parallel 

90 if download_tasks: 

91 _execute_downloads(download_tasks, config) 

92 

93 # Final cleanup if this is the top-level call 

94 if top and remove: 

95 remove_obsolete( 

96 destination_path=destination_path, 

97 files=files, 

98 recycle_bin=recycle_bin.from_config(config, "drive"), 

99 ) 

100 

101 return files 

102 

103 

104def _process_folder_item( 

105 item: Any, 

106 destination_path: str, 

107 filters: dict[str, list[str]] | None, 

108 ignore: list[str] | None, 

109 root: str, 

110 files: set[str], 

111 config: Any | None, 

112) -> None: 

113 """Process a single folder item. 

114 

115 Args: 

116 item: iCloud folder item 

117 destination_path: Local destination directory 

118 filters: Dictionary of filters 

119 ignore: List of ignore patterns 

120 root: Root directory 

121 files: Set to update with processed files 

122 config: Configuration object 

123 """ 

124 new_folder = process_folder( 

125 item=item, 

126 destination_path=destination_path, 

127 filters=filters["folders"] if filters and "folders" in filters else None, 

128 ignore=ignore, 

129 root=root, 

130 ) 

131 if not new_folder: 

132 return 

133 

134 try: 

135 files.add(unicodedata.normalize("NFC", new_folder)) 

136 # Recursively sync subdirectory 

137 subdirectory_files = sync_directory( 

138 drive=item, 

139 destination_path=new_folder, 

140 items=item.dir(), 

141 root=root, 

142 top=False, 

143 filters=filters, 

144 ignore=ignore, 

145 config=config, 

146 ) 

147 files.update(subdirectory_files) 

148 except Exception: 

149 # Continue execution to next item, without crashing the app 

150 pass 

151 

152 

153def _process_file_item( 

154 item: Any, 

155 destination_path: str, 

156 filters: dict[str, list[str]] | None, 

157 ignore: list[str] | None, 

158 root: str, 

159 files: set[str], 

160 download_tasks: list[dict[str, Any]], 

161 config: Any | None = None, 

162) -> None: 

163 """Process a single file item. 

164 

165 Args: 

166 item: iCloud file item 

167 destination_path: Local destination directory 

168 filters: Dictionary of filters 

169 ignore: List of ignore patterns 

170 root: Root directory 

171 files: Set to update with processed files 

172 download_tasks: List to append download tasks to 

173 config: Configuration object 

174 """ 

175 if not wanted_parent_folder( 

176 filters=filters["folders"] if filters and "folders" in filters else None, 

177 ignore=ignore, 

178 root=root, 

179 folder_path=destination_path, 

180 ): 

181 return 

182 

183 try: 

184 download_info = collect_file_for_download( 

185 item=item, 

186 destination_path=destination_path, 

187 filters=filters["file_extensions"] if filters and "file_extensions" in filters else None, 

188 ignore=ignore, 

189 files=files, 

190 config=config, 

191 ) 

192 if download_info: 

193 download_tasks.append(download_info) 

194 except Exception: 

195 # Continue execution to next item, without crashing the app 

196 pass 

197 

198 

199def _execute_downloads(download_tasks: list[dict[str, Any]], config: Any) -> None: 

200 """Execute parallel downloads. 

201 

202 Args: 

203 download_tasks: List of download task dictionaries 

204 config: Configuration object 

205 """ 

206 max_threads = sync_drive.get_max_threads(config) 

207 execute_parallel_downloads(download_tasks, max_threads)