Coverage for src/photo_download_manager.py: 100%
147 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-26 21:28 +0000
1"""Photo download task management module.
3This module contains utilities for managing photo download tasks
4and parallel execution during photo synchronization.
5"""
7___author___ = "Mandar Patil <mandarons@pm.me>"
9import os
10from concurrent.futures import ThreadPoolExecutor, as_completed
11from threading import Lock
13from src import DEFAULT_REQUEST_TIMEOUT_SEC, config_parser, get_logger
14from src.hardlink_registry import HardlinkRegistry
15from src.photo_file_utils import create_hardlink, download_photo_from_server
16from src.photo_path_utils import (
17 _LIVE_VIDEO_SIZES,
18 create_folder_path_if_needed,
19 generate_photo_filename_with_metadata,
20 get_default_filename_format,
21 get_file_format,
22 normalize_file_path,
23 photo_has_edit,
24 rename_legacy_file_if_exists,
25)
27LOGGER = get_logger()
29# Thread-safe lock for file set operations
30files_lock = Lock()
32# ``photos.filters.fallback_to_original``: download ``original`` when an
33# asset has no copy of a requested size. ``sync_photos`` sets this once per
34# run from the parsed filters.
35_FALLBACK_TO_ORIGINAL = False
38def set_fallback_to_original(value: bool) -> None:
39 """Set whether an absent requested size falls back to ``original``."""
40 global _FALLBACK_TO_ORIGINAL
41 _FALLBACK_TO_ORIGINAL = bool(value)
44def _size_present(photo, file_size: str, versions: dict) -> bool:
45 """Whether ``file_size`` can be downloaded as itself.
47 A ``full`` that only the master record carries is Apple's display
48 rendition for a format iCloud does not render natively, not an edit (see
49 ``photo_path_utils.photo_has_edit``), so it counts as absent here too.
50 """
51 if file_size not in versions:
52 return False
53 return file_size != "full" or photo_has_edit(photo)
56def would_fall_back_to_original(photo, file_size: str) -> bool:
57 """Whether ``collect_download_task`` would download ``original`` instead.
59 The album orchestrator uses this to claim the substitution once per photo:
60 with two requested sizes a photo lacks (e.g. ``[adjusted, alternative]``),
61 both would otherwise resolve to the same ``original`` path and race.
62 """
63 if not _FALLBACK_TO_ORIGINAL:
64 return False
65 if file_size == "original" or file_size.startswith("live_video_"):
66 return False
67 versions = photo.versions
68 return not _size_present(photo, file_size, versions) and "original" in versions
71class DownloadTaskInfo:
72 """Information about a photo download task."""
74 def __init__(
75 self,
76 photo,
77 file_size: str,
78 photo_path: str,
79 hardlink_source: str | None = None,
80 hardlink_registry: HardlinkRegistry | None = None,
81 timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC,
82 ):
83 """Initialize download task info.
85 Args:
86 photo: Photo object from iCloudPy
87 file_size: File size variant (original, medium, thumb, etc.)
88 photo_path: Target path for photo download
89 hardlink_source: Path to existing file for hardlink creation
90 hardlink_registry: Registry for tracking downloaded files
91 timeout: HTTP read timeout for the download, in seconds
92 """
93 self.photo = photo
94 self.file_size = file_size
95 self.photo_path = photo_path
96 self.hardlink_source = hardlink_source
97 self.hardlink_registry = hardlink_registry
98 self.timeout = timeout
101def get_max_threads_for_download(config) -> int:
102 """Get maximum number of threads for parallel downloads.
104 Args:
105 config: Configuration dictionary
107 Returns:
108 Maximum number of threads to use for downloads
109 """
110 return config_parser.get_app_max_threads(config)
113def generate_photo_path(photo, file_size: str, destination_path: str, folder_format: str | None) -> str:
114 """Generate full file path for photo with legacy file renaming.
116 This function combines path generation, folder creation, and legacy
117 file renaming into a single operation to maintain backward compatibility.
119 Args:
120 photo: Photo object from iCloudPy
121 file_size: File size variant (original, medium, thumb, etc.)
122 destination_path: Base destination path
123 folder_format: strftime format string for folder creation
125 Returns:
126 Normalized full path where photo should be saved
127 """
128 # Generate filename with metadata
129 filename_with_metadata = generate_photo_filename_with_metadata(photo, file_size)
131 # Create folder path if needed
132 final_destination = create_folder_path_if_needed(
133 destination_path, folder_format, photo,
134 )
136 # Generate paths for legacy file format handling
137 filename = photo.filename
138 name, extension = filename.rsplit(".", 1) if "." in filename else [filename, ""]
140 # Legacy file paths that need to be renamed
141 file_path = os.path.join(destination_path, filename)
142 file_size_path = os.path.join(
143 destination_path,
144 (f"{'__'.join([name, file_size])}" if extension == "" else f"{'__'.join([name, file_size])}.{extension}"),
145 )
147 # Final path with normalization
148 final_file_path = os.path.join(final_destination, filename_with_metadata)
149 normalized_path = normalize_file_path(final_file_path)
151 # The path the still legitimately occupies under the active naming
152 # scheme. Every legacy rename below moves a file onto *this* variant's
153 # path, and os.rename replaces its target -- so a rename starting from
154 # the still destroys the still and whatever stood at the target. Without
155 # folder_format the destination and the photo's folder are the same
156 # directory, which makes the flat "legacy" paths below identical to the
157 # still's current one.
158 still_path = normalize_file_path(
159 os.path.join(
160 final_destination,
161 generate_photo_filename_with_metadata(photo, "original"),
162 ),
163 )
164 _moves_the_still = file_size in _LIVE_VIDEO_SIZES
166 # Rename legacy files if they exist
167 if not (_moves_the_still and file_path == still_path):
168 rename_legacy_file_if_exists(file_path, normalized_path)
169 if not (_moves_the_still and file_size_path == still_path):
170 rename_legacy_file_if_exists(file_size_path, normalized_path)
172 # Self-heal the earlier .HEIC mislabeling of Live Photo videos: an older
173 # version wrote the paired video with the still's extension. Rename that
174 # file to the corrected path instead of re-downloading it (which would also
175 # leave the broken duplicate behind).
176 if file_size in _LIVE_VIDEO_SIZES and extension:
177 root, _ = os.path.splitext(filename_with_metadata)
178 legacy_mislabeled = normalize_file_path(
179 os.path.join(final_destination, f"{root}.{extension}"),
180 )
181 # Under a naming scheme that keeps the photo's own filename, ``root``
182 # is the still's stem, so the candidate above IS the still's correct
183 # path -- renaming it would move the still onto the video's path and,
184 # because os.rename replaces its target, destroy the video too. The
185 # mislabeled file this heals only exists under a scheme that encodes
186 # the variant in the name, where the two paths cannot coincide.
187 # Confirm the candidate really is the paired video before moving it.
188 # os.rename replaces its target, so an unverified rename destroys two
189 # files at once and logs nothing. CloudKit already reports how long
190 # the video should be, so this costs a stat and no extra I/O: a file
191 # of some other length is not the video, whatever it is named.
192 expected_size = (photo.versions.get(file_size) or {}).get("size")
193 try:
194 candidate_size = (
195 os.path.getsize(legacy_mislabeled)
196 if os.path.isfile(legacy_mislabeled)
197 else None
198 )
199 except OSError: # pragma: no cover - raced away between the two calls
200 candidate_size = None
201 if (
202 legacy_mislabeled not in (normalized_path, still_path)
203 and expected_size is not None
204 and candidate_size == expected_size
205 ):
206 rename_legacy_file_if_exists(legacy_mislabeled, normalized_path)
208 # Handle existing file with different normalization
209 if os.path.isfile(final_file_path) and final_file_path != normalized_path:
210 rename_legacy_file_if_exists(final_file_path, normalized_path)
212 return normalized_path
215def collect_download_task(
216 photo,
217 file_size: str,
218 destination_path: str,
219 files: set[str] | None,
220 folder_format: str | None,
221 hardlink_registry: HardlinkRegistry | None,
222) -> DownloadTaskInfo | None:
223 """Collect photo info for parallel download without immediately downloading.
225 Args:
226 photo: Photo object from iCloudPy
227 file_size: File size variant (original, medium, thumb, etc.)
228 destination_path: Base destination path
229 files: Set to track downloaded files (thread-safe updates)
230 folder_format: strftime format string for folder creation
231 hardlink_registry: Registry for tracking downloaded files
233 Returns:
234 DownloadTaskInfo if photo needs to be processed, None if skipped
235 """
236 # Check if file size exists on server. With ``fallback_to_original`` on, an
237 # absent size downloads the original instead of skipping -- that is what
238 # makes ``[adjusted]`` mean "the edit, or the untouched original if the
239 # photo was never edited".
240 #
241 # ``full`` only counts as present when the asset record carries the edit
242 # (see ``photo_has_edit``): a master-carried ``full`` is Apple's display
243 # rendition for a format it does not render natively, and treating it as
244 # an edit would hide a PNG/RAW original behind ``.original.bak``.
245 #
246 # ``live_video_*`` never falls back: those versions are QuickTime movies,
247 # and writing the still image's bytes under a ``.mov`` name would be
248 # corrupt data rather than a substitute.
249 versions = photo.versions
250 if not _size_present(photo, file_size, versions):
251 if would_fall_back_to_original(photo, file_size):
252 LOGGER.debug(
253 f"File size {file_size} not found on server for {photo.filename}. "
254 "Downloading original instead.",
255 )
256 file_size = "original"
257 else:
258 photo_path = generate_photo_path(photo, file_size, destination_path, folder_format)
259 # A missing live_video_* version just means this isn't a Live Photo --
260 # expected for most assets, so log at DEBUG to avoid warning-spam when
261 # live_video_original is in file_sizes. Other sizes warn as before.
262 msg = f"File size {file_size} not found on server. Skipping the photo {photo_path} ..."
263 if file_size.startswith("live_video_"):
264 LOGGER.debug(msg)
265 else:
266 LOGGER.warning(msg)
267 return None
269 # Generate photo path
270 photo_path = generate_photo_path(photo, file_size, destination_path, folder_format)
272 # Check if photo already exists with correct size
273 from src.photo_file_utils import check_photo_exists
275 if check_photo_exists(photo, file_size, photo_path):
276 if files is not None:
277 with files_lock:
278 files.add(photo_path)
279 return None
281 # Filename-collision fallback (``simple`` mode only): a plain
282 # ``IMG_1234.HEIC`` path may already be claimed by a DIFFERENT iCloud
283 # photo that happens to share the human filename. Two collision sources:
284 #
285 # 1. On-disk collision: ``check_photo_exists`` returned False but the
286 # path is occupied -- size mismatch with a photo from a prior sync.
287 # 2. In-flight collision: an earlier call in this same
288 # ``_collect_album_download_tasks`` pass already claimed this plain
289 # path. Without this check the later parallel download silently
290 # overwrites the earlier one and we lose data. ``collect_download_task``
291 # runs sequentially during collection so a plain ``in files`` membership
292 # test under ``files_lock`` is sufficient; we hold the lock only long
293 # enough to read.
294 #
295 # In either case we route this photo to the metadata-suffix filename so
296 # both files coexist and both round-trip stably on future syncs. We use
297 # the ``get_default_filename_format()`` accessor (rather than importing
298 # the module-level constant) so ``set_default_filename_format`` updates
299 # are observed live on every call.
300 # Non-unique naming (simple mode, or a photos.file_format template that may
301 # omit a unique component) can collide; the metadata format cannot.
302 is_non_unique = get_default_filename_format() == "simple" or get_file_format() is not None
303 in_flight_collision = False
304 if is_non_unique and files is not None:
305 with files_lock:
306 in_flight_collision = photo_path in files
307 if is_non_unique and (os.path.isfile(photo_path) or in_flight_collision):
308 # Protect the file we are stepping around. This branch exists to
309 # "preserve both photos", but obsolete-file cleanup deletes anything
310 # absent from ``files`` -- so without this the run writes the suffix
311 # copy and then deletes the very file it just refused to overwrite.
312 # On a library with many repeated filenames that is not a rare edge
313 # case: it deletes a large share of the library every sync, and the
314 # next sync re-downloads it, because freeing the plain path changes
315 # the collision outcome. Only an on-disk collision needs this; an
316 # in-flight one means this run already claimed the path.
317 if files is not None and os.path.isfile(photo_path):
318 with files_lock:
319 files.add(photo_path)
320 suffix_folder = create_folder_path_if_needed(
321 destination_path, folder_format, photo,
322 )
323 suffix_basename = generate_photo_filename_with_metadata(
324 photo, file_size, "metadata",
325 )
326 photo_path = normalize_file_path(os.path.join(suffix_folder, suffix_basename))
327 LOGGER.info(
328 f"Filename collision for {photo.filename} (id={photo.id}); "
329 f"using suffix path {photo_path} to preserve both photos.",
330 )
331 if check_photo_exists(photo, file_size, photo_path):
332 if files is not None:
333 with files_lock:
334 files.add(photo_path)
335 return None
337 if files is not None:
338 with files_lock:
339 files.add(photo_path)
341 # Check for existing hardlink source
342 hardlink_source = None
343 if hardlink_registry is not None:
344 hardlink_source = hardlink_registry.get_existing_path(photo.id, file_size)
346 return DownloadTaskInfo(
347 photo=photo,
348 file_size=file_size,
349 photo_path=photo_path,
350 hardlink_source=hardlink_source,
351 hardlink_registry=hardlink_registry,
352 )
355def execute_download_task(task_info: DownloadTaskInfo) -> bool:
356 """Download a single photo or create hardlink as part of parallel execution.
358 Args:
359 task_info: Download task information
361 Returns:
362 True if task completed successfully, False otherwise
363 """
364 LOGGER.debug(f"[Thread] Starting processing of {task_info.photo_path}")
366 try:
367 # Try hardlink first if source exists
368 if task_info.hardlink_source:
369 if create_hardlink(task_info.hardlink_source, task_info.photo_path):
370 LOGGER.debug(f"[Thread] Created hardlink for {task_info.photo_path}")
371 return True
372 else:
373 # Fallback to download if hard link creation fails
374 LOGGER.warning(
375 f"Hard link creation failed, downloading {task_info.photo_path} instead",
376 )
378 # Download the photo
379 result = download_photo_from_server(
380 task_info.photo,
381 task_info.file_size,
382 task_info.photo_path,
383 timeout=task_info.timeout,
384 )
385 if result and task_info.hardlink_registry is not None:
386 # Register for future hard links if enabled
387 task_info.hardlink_registry.register_photo_path(
388 task_info.photo.id,
389 task_info.file_size,
390 task_info.photo_path,
391 )
392 LOGGER.debug(f"[Thread] Completed download of {task_info.photo_path}")
394 return result
396 except Exception as e:
397 LOGGER.error(f"[Thread] Failed to process {task_info.photo_path}: {e!s}")
398 return False
401def execute_parallel_downloads(
402 download_tasks: list[DownloadTaskInfo], config,
403) -> tuple[int, int]:
404 """Execute download tasks in parallel using thread pool.
406 Args:
407 download_tasks: List of download tasks to execute
408 config: Configuration dictionary for thread settings
410 Returns:
411 Tuple of (successful_downloads, failed_downloads)
412 """
413 if not download_tasks:
414 return 0, 0
416 max_threads = get_max_threads_for_download(config)
418 # Resolved once here rather than per task: without it a stalled CDN
419 # connection blocks its worker thread for the life of the process.
420 timeout = config_parser.get_photos_request_timeout(config)
421 for task in download_tasks:
422 task.timeout = timeout
424 # Count hardlink tasks vs download tasks for logging
425 hardlink_tasks = sum(1 for task in download_tasks if task.hardlink_source)
426 download_only_tasks = len(download_tasks) - hardlink_tasks
428 if hardlink_tasks > 0:
429 LOGGER.info(
430 f"Starting parallel processing with {max_threads} threads: "
431 f"{hardlink_tasks} hard links, {download_only_tasks} downloads...",
432 )
433 else:
434 LOGGER.info(
435 f"Starting parallel photo downloads with {max_threads} threads for {len(download_tasks)} photos...",
436 )
438 successful_downloads = 0
439 failed_downloads = 0
441 with ThreadPoolExecutor(max_workers=max_threads) as executor:
442 # Submit all download tasks
443 future_to_task = {
444 executor.submit(execute_download_task, task): task
445 for task in download_tasks
446 }
448 # Process completed downloads
449 for future in as_completed(future_to_task):
450 try:
451 result = future.result()
452 if result:
453 successful_downloads += 1
454 else:
455 failed_downloads += 1
456 except Exception as e: # noqa: PERF203
457 LOGGER.error(f"Unexpected error during photo download: {e!s}")
458 failed_downloads += 1
460 LOGGER.info(
461 f"Photo processing complete: {successful_downloads} successful, {failed_downloads} failed",
462 )
463 return successful_downloads, failed_downloads