Coverage for src/drive_parallel_download.py: 100%
78 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 21:05 +0000
1"""Parallel download utilities.
3This module provides parallel download coordination for iCloud Drive sync,
4separating parallel execution logic from sync operations per SRP.
5"""
7__author__ = "Mandar Patil (mandarons@pm.me)"
9import os
10import unicodedata
11from concurrent.futures import ThreadPoolExecutor, as_completed
12from pathlib import Path
13from threading import Lock
14from typing import Any
15from urllib.parse import unquote
17from src import (
18 DEFAULT_REQUEST_TIMEOUT_SEC,
19 config_parser,
20 configure_icloudpy_logging,
21 get_logger,
22)
23from src.drive_file_download import download_file
24from src.drive_file_existence import file_exists, is_package, package_bundle_unchanged, package_exists
25from src.drive_filtering import wanted_file
27# Configure icloudpy logging immediately after import
28configure_icloudpy_logging()
30LOGGER = get_logger()
32# Thread-safe lock for file set operations
33files_lock = Lock()
36def collect_file_for_download(
37 item: Any,
38 destination_path: str,
39 filters: list[str] | None,
40 ignore: list[str] | None,
41 files: set[str],
42 config: dict | None = None,
43) -> dict[str, Any] | None:
44 """Collect file information for parallel download without immediately downloading.
46 Args:
47 item: iCloud file item
48 destination_path: Local destination directory
49 filters: File extension filters
50 ignore: Ignore patterns
51 files: Set to track processed files (thread-safe updates)
52 config: Configuration dictionary (used to resolve request timeout)
54 Returns:
55 Download task info dict, or None if file should be skipped
56 """
57 if not (item and destination_path and files is not None):
58 return None
60 # Decode URL-encoded filename from iCloud API
61 # This handles special characters like %CC%88 (combining diacritical marks)
62 decoded_name = unquote(item.name)
63 local_file = os.path.join(destination_path, decoded_name)
64 local_file = unicodedata.normalize("NFC", local_file)
66 if not wanted_file(filters=filters, ignore=ignore, file_path=local_file):
67 return None
69 # Thread-safe file set update
70 with files_lock:
71 files.add(local_file)
73 flatten_packages = (
74 bool(config_parser.get_drive_flatten_packages(config))
75 if config
76 else False
77 )
79 # Check local existence FIRST to avoid unnecessary network requests.
80 # is_package() makes an HTTP call for every file, which is very slow
81 # when syncing thousands of already-up-to-date files.
82 if os.path.isfile(local_file):
83 # It's a regular file locally — check if it's up-to-date (no network call)
84 if file_exists(item=item, local_file=local_file):
85 return None
86 # File is outdated; still need to determine its type for re-download
87 elif os.path.isdir(local_file):
88 # A directory at this path means the item was previously downloaded as a
89 # package. iCloud Drive items do not change type between file and package,
90 # so package_exists() is the correct check here (no is_package() needed).
91 # Note: package_exists() deletes the directory if it is outdated.
92 if package_exists(item=item, local_package_path=local_file):
93 with files_lock:
94 for f in Path(local_file).glob("**/*"):
95 files.add(str(f))
96 return None
97 # Directory was deleted by package_exists(); schedule re-download as a package
98 return {
99 "item": item,
100 "local_file": local_file,
101 "is_package": True,
102 "files": files,
103 "timeout": config_parser.get_drive_request_timeout(config),
104 "flatten_packages": flatten_packages,
105 }
107 # File/directory doesn't exist locally (or was an outdated regular file that needs
108 # re-download) — make the network call to determine the item type
109 timeout = config_parser.get_drive_request_timeout(config)
110 item_is_package = is_package(item=item, timeout=timeout)
112 # A package stored as a flat single-file bundle (unrecognised-mime package,
113 # or flatten_packages=true) reports item.size as the package's *logical*
114 # size, which never equals the on-disk bundle's byte count -- so file_exists()
115 # above sees a spurious size mismatch and we'd re-download the bundle on every
116 # sync. When it IS a package and the on-disk bundle's mtime already matches the
117 # remote, the bytes are unchanged; skip the wasteful re-download. is_package()
118 # was just called above, so this adds no extra network round-trip.
119 if item_is_package and package_bundle_unchanged(item=item, local_file=local_file):
120 return None
122 # Return download task info
123 return {
124 "item": item,
125 "local_file": local_file,
126 "is_package": item_is_package,
127 "files": files,
128 "timeout": timeout,
129 "flatten_packages": flatten_packages,
130 }
133def download_file_task(download_info: dict[str, Any]) -> bool:
134 """Download a single file as part of parallel execution.
136 Args:
137 download_info: Dictionary containing download task information
139 Returns:
140 True if download succeeded, False otherwise
141 """
142 item = download_info["item"]
143 local_file = download_info["local_file"]
144 is_package = download_info["is_package"]
145 files = download_info["files"]
147 LOGGER.debug(f"[Thread] Starting download of {local_file}")
149 try:
150 downloaded_file = download_file(
151 item=item,
152 local_file=local_file,
153 timeout=download_info.get("timeout", DEFAULT_REQUEST_TIMEOUT_SEC),
154 flatten_packages=download_info.get("flatten_packages", False),
155 )
156 if not downloaded_file:
157 return False
159 if is_package:
160 # Record the package's contents exactly as they are on disk, so
161 # obsolete-file cleanup keeps them. They used to be renamed to the
162 # NFD form of their whole path first, which on Linux failed under
163 # a non-ASCII parent folder and otherwise moved the package's
164 # mtime off iCloud's, so package_exists re-downloaded it every sync.
165 with files_lock:
166 files.update(str(f) for f in Path(downloaded_file).glob("**/*"))
168 LOGGER.debug(f"[Thread] Completed download of {local_file}")
169 return True
170 except Exception as e:
171 LOGGER.error(f"[Thread] Failed to download {local_file}: {e!s}")
172 return False
175def execute_parallel_downloads(
176 download_tasks: list[dict[str, Any]],
177 max_threads: int,
178) -> tuple[int, int]:
179 """Execute multiple file downloads in parallel.
181 Args:
182 download_tasks: List of download task dictionaries
183 max_threads: Maximum number of concurrent threads
185 Returns:
186 Tuple of (successful_downloads, failed_downloads) counts
187 """
188 if not download_tasks:
189 return 0, 0
191 LOGGER.info(
192 f"Starting parallel downloads with {max_threads} threads for {len(download_tasks)} files...",
193 )
195 successful_downloads = 0
196 failed_downloads = 0
198 with ThreadPoolExecutor(max_workers=max_threads) as executor:
199 # Submit all download tasks
200 future_to_task = {
201 executor.submit(download_file_task, task): task for task in download_tasks
202 }
204 # Process completed downloads
205 for future in as_completed(future_to_task):
206 try:
207 result = future.result()
208 if result:
209 successful_downloads += 1
210 else:
211 failed_downloads += 1
212 except Exception as e: # noqa: PERF203
213 LOGGER.error(f"Download task failed with exception: {e!s}")
214 failed_downloads += 1
216 LOGGER.info(
217 f"Parallel downloads completed: {successful_downloads} successful, {failed_downloads} failed",
218 )
219 return successful_downloads, failed_downloads