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, 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(destination_path=destination_path, files=files) 

96 

97 return files 

98 

99 

100def _process_folder_item( 

101 item: Any, 

102 destination_path: str, 

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

104 ignore: list[str] | None, 

105 root: str, 

106 files: set[str], 

107 config: Any | None, 

108) -> None: 

109 """Process a single folder item. 

110 

111 Args: 

112 item: iCloud folder item 

113 destination_path: Local destination directory 

114 filters: Dictionary of filters 

115 ignore: List of ignore patterns 

116 root: Root directory 

117 files: Set to update with processed files 

118 config: Configuration object 

119 """ 

120 new_folder = process_folder( 

121 item=item, 

122 destination_path=destination_path, 

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

124 ignore=ignore, 

125 root=root, 

126 ) 

127 if not new_folder: 

128 return 

129 

130 try: 

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

132 # Recursively sync subdirectory 

133 subdirectory_files = sync_directory( 

134 drive=item, 

135 destination_path=new_folder, 

136 items=item.dir(), 

137 root=root, 

138 top=False, 

139 filters=filters, 

140 ignore=ignore, 

141 config=config, 

142 ) 

143 files.update(subdirectory_files) 

144 except Exception: 

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

146 pass 

147 

148 

149def _process_file_item( 

150 item: Any, 

151 destination_path: str, 

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

153 ignore: list[str] | None, 

154 root: str, 

155 files: set[str], 

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

157 config: Any | None = None, 

158) -> None: 

159 """Process a single file item. 

160 

161 Args: 

162 item: iCloud file item 

163 destination_path: Local destination directory 

164 filters: Dictionary of filters 

165 ignore: List of ignore patterns 

166 root: Root directory 

167 files: Set to update with processed files 

168 download_tasks: List to append download tasks to 

169 config: Configuration object 

170 """ 

171 if not wanted_parent_folder( 

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

173 ignore=ignore, 

174 root=root, 

175 folder_path=destination_path, 

176 ): 

177 return 

178 

179 try: 

180 download_info = collect_file_for_download( 

181 item=item, 

182 destination_path=destination_path, 

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

184 ignore=ignore, 

185 files=files, 

186 config=config, 

187 ) 

188 if download_info: 

189 download_tasks.append(download_info) 

190 except Exception: 

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

192 pass 

193 

194 

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

196 """Execute parallel downloads. 

197 

198 Args: 

199 download_tasks: List of download task dictionaries 

200 config: Configuration object 

201 """ 

202 max_threads = sync_drive.get_max_threads(config) 

203 execute_parallel_downloads(download_tasks, max_threads)