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 ) 

87 

88 # Second pass: execute downloads in parallel 

89 if download_tasks: 

90 _execute_downloads(download_tasks, config) 

91 

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

93 if top and remove: 

94 remove_obsolete(destination_path=destination_path, files=files) 

95 

96 return files 

97 

98 

99def _process_folder_item( 

100 item: Any, 

101 destination_path: str, 

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

103 ignore: list[str] | None, 

104 root: str, 

105 files: set[str], 

106 config: Any | None, 

107) -> None: 

108 """Process a single folder item. 

109 

110 Args: 

111 item: iCloud folder item 

112 destination_path: Local destination directory 

113 filters: Dictionary of filters 

114 ignore: List of ignore patterns 

115 root: Root directory 

116 files: Set to update with processed files 

117 config: Configuration object 

118 """ 

119 new_folder = process_folder( 

120 item=item, 

121 destination_path=destination_path, 

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

123 ignore=ignore, 

124 root=root, 

125 ) 

126 if not new_folder: 

127 return 

128 

129 try: 

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

131 # Recursively sync subdirectory 

132 subdirectory_files = sync_directory( 

133 drive=item, 

134 destination_path=new_folder, 

135 items=item.dir(), 

136 root=root, 

137 top=False, 

138 filters=filters, 

139 ignore=ignore, 

140 config=config, 

141 ) 

142 files.update(subdirectory_files) 

143 except Exception: 

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

145 pass 

146 

147 

148def _process_file_item( 

149 item: Any, 

150 destination_path: str, 

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

152 ignore: list[str] | None, 

153 root: str, 

154 files: set[str], 

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

156) -> None: 

157 """Process a single file item. 

158 

159 Args: 

160 item: iCloud file item 

161 destination_path: Local destination directory 

162 filters: Dictionary of filters 

163 ignore: List of ignore patterns 

164 root: Root directory 

165 files: Set to update with processed files 

166 download_tasks: List to append download tasks to 

167 """ 

168 if not wanted_parent_folder( 

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

170 ignore=ignore, 

171 root=root, 

172 folder_path=destination_path, 

173 ): 

174 return 

175 

176 try: 

177 download_info = collect_file_for_download( 

178 item=item, 

179 destination_path=destination_path, 

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

181 ignore=ignore, 

182 files=files, 

183 ) 

184 if download_info: 

185 download_tasks.append(download_info) 

186 except Exception: 

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

188 pass 

189 

190 

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

192 """Execute parallel downloads. 

193 

194 Args: 

195 download_tasks: List of download task dictionaries 

196 config: Configuration object 

197 """ 

198 max_threads = sync_drive.get_max_threads(config) 

199 execute_parallel_downloads(download_tasks, max_threads)