Coverage for src/sync_photos.py: 100%
256 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 21:05 +0000
1"""Sync photos module.
3This module provides the main photo synchronization functionality,
4orchestrating the downloading of photos from iCloud to local storage.
5"""
7___author___ = "Mandar Patil <mandarons@pm.me>"
9import os
11import requests
12from icloudpy import exceptions
14from src import config_parser, configure_icloudpy_logging, get_logger, recycle_bin
15from src.album_sync_orchestrator import sync_album_photos
16from src.hardlink_registry import create_hardlink_registry
17from src.photo_cleanup_utils import remove_obsolete_files
19# Configure icloudpy logging immediately after import
20configure_icloudpy_logging()
22LOGGER = get_logger()
25# Legacy functions preserved for backward compatibility with existing tests
26# These functions are now implemented using the new modular architecture
29def get_max_threads(config):
30 """Get maximum number of threads for parallel downloads.
32 Legacy function - now delegates to config_parser.
34 Args:
35 config: Configuration dictionary
37 Returns:
38 Maximum number of threads to use for downloads
39 """
40 return config_parser.get_app_max_threads(config)
43def get_name_and_extension(photo, file_size):
44 """Extract filename and extension.
46 Legacy function - now delegates to photo_path_utils.
48 Args:
49 photo: Photo object from iCloudPy
50 file_size: File size variant
52 Returns:
53 Tuple of (name, extension)
54 """
55 from src.photo_path_utils import get_photo_name_and_extension
57 return get_photo_name_and_extension(photo, file_size)
60def photo_wanted(photo, extensions):
61 """Check if photo is wanted based on extension.
63 Legacy function - now delegates to photo_filter_utils.
65 Args:
66 photo: Photo object from iCloudPy
67 extensions: List of allowed extensions
69 Returns:
70 True if photo should be synced, False otherwise
71 """
72 from src.photo_filter_utils import is_photo_wanted
74 return is_photo_wanted(photo, extensions)
77def generate_file_name(photo, file_size, destination_path, folder_format):
78 """Generate full path to file.
80 Legacy function - now delegates to photo_download_manager.
82 Args:
83 photo: Photo object from iCloudPy
84 file_size: File size variant
85 destination_path: Base destination path
86 folder_format: Folder format string
88 Returns:
89 Full file path
90 """
91 from src.photo_download_manager import generate_photo_path
93 return generate_photo_path(photo, file_size, destination_path, folder_format)
96def photo_exists(photo, file_size, local_path):
97 """Check if photo exist locally.
99 Legacy function - now delegates to photo_file_utils.
101 Args:
102 photo: Photo object from iCloudPy
103 file_size: File size variant
104 local_path: Local file path to check
106 Returns:
107 True if photo exists with correct size, False otherwise
108 """
109 from src.photo_file_utils import check_photo_exists
111 return check_photo_exists(photo, file_size, local_path)
114def create_hardlink(source_path, destination_path):
115 """Create a hard link from source to destination.
117 Legacy function - now delegates to photo_file_utils.
119 Args:
120 source_path: Path to source file
121 destination_path: Path for new hardlink
123 Returns:
124 True if successful, False otherwise
125 """
126 from src.photo_file_utils import create_hardlink as create_hardlink_impl
128 return create_hardlink_impl(source_path, destination_path)
131def download_photo(photo, file_size, destination_path):
132 """Download photo from server.
134 Legacy function - now delegates to photo_file_utils.
136 Args:
137 photo: Photo object from iCloudPy
138 file_size: File size variant
139 destination_path: Where to save the photo
141 Returns:
142 True if successful, False otherwise
143 """
144 from src.photo_file_utils import download_photo_from_server
146 return download_photo_from_server(photo, file_size, destination_path)
149def process_photo(photo, file_size, destination_path, files, folder_format, hardlink_registry=None):
150 """Process photo details (legacy function for backward compatibility).
152 Args:
153 photo: Photo object from iCloudPy
154 file_size: File size variant
155 destination_path: Base destination path
156 files: Set to track downloaded files
157 folder_format: Folder format string
158 hardlink_registry: Registry for hardlinks (legacy dict format)
160 Returns:
161 True if photo was processed successfully, False otherwise
162 """
163 from src.photo_download_manager import collect_download_task, execute_download_task
165 # Convert legacy hardlink registry dict to new registry format if needed
166 converted_registry = None
167 if hardlink_registry is not None:
168 from src.hardlink_registry import HardlinkRegistry
170 converted_registry = HardlinkRegistry()
171 for key, path in hardlink_registry.items():
172 # Legacy format: photo_id_file_size -> path
173 if "_" in key:
174 parts = key.rsplit("_", 1)
175 if len(parts) == 2:
176 photo_id, file_sz = parts
177 converted_registry.register_photo_path(photo_id, file_sz, path)
179 # Collect download task
180 task_info = collect_download_task(
181 photo,
182 file_size,
183 destination_path,
184 files,
185 folder_format,
186 converted_registry,
187 )
189 if task_info is None:
190 return False
192 # Execute task
193 result = execute_download_task(task_info)
195 # Update legacy registry if provided
196 if result and hardlink_registry is not None:
197 photo_key = f"{photo.id}_{file_size}"
198 hardlink_registry[photo_key] = task_info.photo_path
200 return result
203def collect_photo_for_download(photo, file_size, destination_path, files, folder_format, hardlink_registry=None):
204 """Collect photo info for parallel download without immediately downloading.
206 Legacy function - now delegates to photo_download_manager.
208 Args:
209 photo: Photo object from iCloudPy
210 file_size: File size variant
211 destination_path: Base destination path
212 files: Set to track downloaded files
213 folder_format: Folder format string
214 hardlink_registry: Registry for hardlinks (legacy dict format)
216 Returns:
217 Download task info or None
218 """
219 from src.photo_download_manager import collect_download_task
221 # Convert legacy hardlink registry dict to new registry format if needed
222 converted_registry = None
223 if hardlink_registry is not None:
224 from src.hardlink_registry import HardlinkRegistry
226 converted_registry = HardlinkRegistry()
227 for key, path in hardlink_registry.items():
228 if "_" in key:
229 parts = key.rsplit("_", 1)
230 if len(parts) == 2:
231 photo_id, file_sz = parts
232 converted_registry.register_photo_path(photo_id, file_sz, path)
234 task_info = collect_download_task(
235 photo,
236 file_size,
237 destination_path,
238 files,
239 folder_format,
240 converted_registry,
241 )
243 if task_info is None:
244 return None
246 # Convert back to legacy format for compatibility
247 return {
248 "photo": task_info.photo,
249 "file_size": task_info.file_size,
250 "photo_path": task_info.photo_path,
251 "hardlink_source": task_info.hardlink_source,
252 "hardlink_registry": hardlink_registry,
253 }
256def download_photo_task(download_info):
257 """Download a single photo or create hardlink as part of parallel execution.
259 Legacy function - maintains original implementation for backward compatibility.
261 Args:
262 download_info: Dictionary with download task information
264 Returns:
265 True if successful, False otherwise
266 """
267 photo = download_info["photo"]
268 file_size = download_info["file_size"]
269 photo_path = download_info["photo_path"]
270 hardlink_source = download_info.get("hardlink_source")
271 hardlink_registry = download_info.get("hardlink_registry")
273 LOGGER.debug(f"[Thread] Starting processing of {photo_path}")
275 try:
276 # Try hardlink first if source exists
277 if hardlink_source:
278 if create_hardlink(hardlink_source, photo_path):
279 LOGGER.debug(f"[Thread] Created hardlink for {photo_path}")
280 return True
281 else:
282 # Fallback to download if hard link creation fails
283 LOGGER.warning(f"Hard link creation failed, downloading {photo_path} instead")
285 # Download the photo - this maintains the original function call for test compatibility
286 result = download_photo(photo, file_size, photo_path)
287 if result:
288 # Register for future hard links if enabled
289 if hardlink_registry is not None:
290 photo_key = f"{photo.id}_{file_size}"
291 hardlink_registry[photo_key] = photo_path
292 LOGGER.debug(f"[Thread] Completed download of {photo_path}")
293 return result
294 except Exception as e:
295 LOGGER.error(f"[Thread] Failed to process {photo_path}: {e!s}")
296 return False
299def sync_album(
300 album,
301 destination_path,
302 file_sizes,
303 extensions=None,
304 files=None,
305 folder_format=None,
306 hardlink_registry=None,
307 config=None,
308):
309 """Sync given album.
311 Legacy function - now delegates to album_sync_orchestrator with conversion
312 for legacy hardlink registry format.
314 Args:
315 album: Album object from iCloudPy
316 destination_path: Path where photos should be saved
317 file_sizes: List of file size variants to download
318 extensions: List of allowed file extensions
319 files: Set to track downloaded files
320 folder_format: Folder format string
321 hardlink_registry: Registry for hardlinks (legacy dict format)
322 config: Configuration dictionary
324 Returns:
325 True on success, None on invalid input
326 """
327 # Convert legacy hardlink registry dict to new registry format if needed
328 converted_registry = None
329 if hardlink_registry is not None:
330 from src.hardlink_registry import HardlinkRegistry
332 converted_registry = HardlinkRegistry()
333 for key, path in hardlink_registry.items():
334 if "_" in key:
335 parts = key.rsplit("_", 1)
336 if len(parts) == 2:
337 photo_id, file_sz = parts
338 converted_registry.register_photo_path(photo_id, file_sz, path)
340 result = sync_album_photos(
341 album=album,
342 destination_path=destination_path,
343 file_sizes=file_sizes,
344 extensions=extensions,
345 files=files,
346 folder_format=folder_format,
347 hardlink_registry=converted_registry,
348 config=config,
349 )
351 # Update legacy registry if provided and new registry was created
352 if hardlink_registry is not None and converted_registry is not None:
353 # This is a simplified approach - in practice, we'd need to track new entries
354 # But for legacy compatibility, we'll maintain the existing behavior
355 pass
357 return result
360def remove_obsolete(destination_path, files):
361 """Remove local obsolete file.
363 Legacy function - now delegates to photo_cleanup_utils.
365 Args:
366 destination_path: Path to search for obsolete files
367 files: Set of files that should be kept
369 Returns:
370 Set of removed file paths
371 """
372 return remove_obsolete_files(destination_path, files)
375def sync_photos(config, photos):
376 """Sync all photos.
378 Main orchestration function that coordinates the entire photo sync process.
379 This function has been refactored to use the new modular architecture while
380 maintaining backward compatibility.
382 Args:
383 config: Configuration dictionary
384 photos: Photos object from iCloudPy
386 Returns:
387 Tuple of (total_successful, total_failed) download counts
388 """
389 # Parse configuration using centralized config parser
390 destination_path = config_parser.prepare_photos_destination(config=config)
391 library_destinations = config_parser.get_photos_library_destinations(config=config)
392 # Apply the requested filename convention for this sync run. The module-level
393 # default in photo_path_utils flows down through generate_photo_path -> all
394 # downloads without requiring a per-call signature change.
395 from src.photo_path_utils import set_default_filename_format, set_file_format
396 set_default_filename_format(config_parser.get_photos_filename_format(config=config))
397 set_file_format(
398 config_parser.get_photos_file_format(config=config),
399 config_parser.get_photos_variant_separator(config=config),
400 )
401 filters = config_parser.get_photos_filters(config=config)
402 # preserve_originals_as_bak invariant: at least ONE visible file per
403 # photo must always land on disk. The suffix only hides ``original``
404 # behind ``.bak`` -- so if the user isn't ALSO downloading
405 # ``original_alt`` (the visible "current view" of an edited photo),
406 # enabling the suffix would leave nothing visible. Gate the toggle
407 # on both ``preserve_originals_as_bak: true`` AND ``original_alt`` in
408 # ``file_sizes``. If a photo is later removed from iCloud, the normal
409 # remove_obsolete sweep cascades to both the visible file AND the
410 # ``.bak`` sidecar (neither is in tracked_files, both get deleted) --
411 # so there's no risk of orphan ``.bak`` files outliving their parent.
412 from src.photo_path_utils import set_preserve_originals_as_bak
414 set_preserve_originals_as_bak(
415 config_parser.get_photos_preserve_originals_as_bak(config=config)
416 and "original_alt" in (filters.get("file_sizes") or []),
417 )
418 files = set()
419 download_all = config_parser.get_photos_all_albums(config=config)
420 use_hardlinks = config_parser.get_photos_use_hardlinks(config=config)
421 libraries = filters["libraries"] if filters["libraries"] is not None else photos.libraries
422 folder_format = config_parser.get_photos_folder_format(config=config)
424 # Initialize hard link registry using new modular approach
425 hardlink_registry = create_hardlink_registry(use_hardlinks)
427 total_successful, total_failed = 0, 0
428 # Libraries we could not read. They must be barred from the cleanup
429 # below: a library that failed contributes nothing to ``files``, so
430 # cleaning its destination would read as "the server has none of these"
431 # and delete every local copy.
432 failed_libraries: set = set()
434 # Special handling for "All Photos" when hardlinks are enabled
435 if use_hardlinks and download_all:
436 sub_successful, sub_failed = _sync_all_photos_first_for_hardlinks(
437 photos,
438 libraries,
439 destination_path,
440 filters,
441 files,
442 folder_format,
443 hardlink_registry,
444 config,
445 library_destinations=library_destinations,
446 failed_libraries=failed_libraries,
447 )
448 total_successful += sub_successful
449 total_failed += sub_failed
451 # Sync albums based on configuration
452 sub_successful, sub_failed = _sync_albums_by_configuration(
453 photos,
454 libraries,
455 download_all,
456 destination_path,
457 filters,
458 files,
459 folder_format,
460 hardlink_registry,
461 config,
462 library_destinations=library_destinations,
463 failed_libraries=failed_libraries,
464 )
465 total_successful += sub_successful
466 total_failed += sub_failed
468 # Clean up obsolete files if enabled. When per-library destinations are
469 # configured we walk each library's subdir independently, otherwise the
470 # legacy single-destination walk preserves backward compatibility.
471 if config_parser.get_photos_remove_obsolete(config=config):
472 marker_filename = config_parser.get_mount_marker_filename(config=config)
473 exclude = {marker_filename}
474 limit = config_parser.get_photos_obsolete_delete_limit_percent(config=config)
475 bin_ = recycle_bin.from_config(config, "photos")
476 if library_destinations:
477 for library in libraries:
478 if library in failed_libraries:
479 LOGGER.warning(
480 f"Skipping obsolete-file cleanup for {library}: it could not "
481 f"be read this run, so its local files cannot be verified.",
482 )
483 continue
484 lib_dest = _library_destination(destination_path, library, library_destinations)
485 if lib_dest == destination_path:
486 # An unmapped library falls through to the shared root,
487 # which holds every other library's tree -- and anything
488 # else the user keeps there. Cleaning it on behalf of one
489 # library would delete all of them. Apple invents
490 # libraries (backend migration zones), so this is reached
491 # by doing nothing wrong.
492 LOGGER.warning(
493 f"Skipping obsolete-file cleanup for {library}: it has no "
494 f"library_destinations entry, so its destination is the "
495 f"shared root and cleaning it would affect other libraries.",
496 )
497 continue
498 remove_obsolete_files(
499 lib_dest,
500 files,
501 exclude_filenames=exclude,
502 limit_percent=limit,
503 recycle_bin=bin_ and bin_.within(_bin_subfolder(lib_dest, destination_path, library)),
504 )
505 elif failed_libraries:
506 # One shared destination: ``files`` cannot say which library a
507 # path came from, so a single failure makes the whole set
508 # untrustworthy for deletion.
509 LOGGER.warning(
510 "Skipping obsolete-file cleanup: "
511 f"{len(failed_libraries)} library(ies) could not be read this run.",
512 )
513 else:
514 remove_obsolete_files(
515 destination_path,
516 files,
517 exclude_filenames=exclude,
518 limit_percent=limit,
519 recycle_bin=bin_,
520 )
522 return total_successful, total_failed
525def _bin_subfolder(lib_dest: str, destination_path: str, library: str) -> str:
526 """Where a library's files go inside the shared Photos bin.
528 The library's folder under the Photos destination, as it appears on
529 disk, so a restore is the same relative move. A mapping that points
530 outside the destination has no such folder, so the library's own name
531 stands in for it.
532 """
533 relative = os.path.relpath(lib_dest, destination_path)
534 if relative == os.pardir or relative.startswith(os.pardir + os.sep):
535 return library
536 return relative
539def _library_destination(base_destination: str, library: str, library_destinations: dict) -> str:
540 """Resolve the on-disk destination for a given iCloud photo library.
542 When ``library_destinations`` provides a mapping for ``library``, joins
543 the configured subdirectory under ``base_destination`` and ensures the
544 directory exists. Otherwise returns ``base_destination`` unchanged
545 (preserving mandarons' legacy single-destination behaviour).
547 Library-name matching has three rules, in priority order:
549 1. **Exact match.** ``library_destinations[library]`` if present.
550 2. **Role alias for `SharedLibrary`.** Apple's modern iCloud Shared
551 Photo Library is exposed by icloudpy under a GUID-based zone name
552 like ``SharedSync-<guid>``. A config
553 key of ``SharedLibrary`` matches any zone whose name starts with
554 ``SharedSync-`` so users don't need to discover and hardcode the
555 per-account GUID. (Configs that already use the literal current
556 Apple zone name still work via rule 1.)
557 3. **Fallthrough.** Returns ``base_destination`` unchanged.
558 """
559 if not library_destinations:
560 return base_destination
561 # ``get`` returns None only when the key is absent -- the config parser
562 # coerces every configured value to str, so an explicit "" stays "".
563 # Key on ``is None`` (not falsiness) so an explicitly-mapped library
564 # always wins over the SharedLibrary alias and the default, honouring
565 # the rule 1 > rule 2 > rule 3 priority documented above.
566 subdir = library_destinations.get(library)
567 if subdir is None and library.startswith("SharedSync-"):
568 subdir = library_destinations.get("SharedLibrary")
569 if subdir is None:
570 return base_destination
571 dest = os.path.join(base_destination, subdir)
572 os.makedirs(dest, exist_ok=True)
573 return dest
576# An account can be shown libraries it never created -- zones left behind
577# by Apple's own backend migrations -- and some of them answer every query
578# with ZONE_NOT_FOUND or BAD_REQUEST. One of those must not be able to
579# abort the sync for the libraries that work.
580_LIBRARY_FAULTS = (exceptions.ICloudPyAPIResponseException, requests.exceptions.RequestException)
583def _signal_library(action: str, library: str, **kwargs) -> None:
584 """Publish per-library progress for the dashboard, best effort.
586 A library can take an hour, so "which one is running right now" is
587 only knowable while it runs. Wrapped because the web-signal module is
588 optional and a dashboard write must never affect the sync.
589 """
590 try:
591 from src import web_signals
593 getattr(web_signals, action)(library, **kwargs)
594 except Exception as e: # noqa: BLE001 -- reporting must not break syncing
595 LOGGER.debug(f"web_signals: {action} for {library} raised: {e!s}")
598def is_photos_indexing(error) -> bool:
599 """True when ``error`` is Apple saying a photo library is still indexing.
601 icloudpy raises this as a plain ServiceNotActivated, with only the
602 message to tell it apart, and it is account-wide rather than
603 per-library: ``PhotosService`` checks the primary zone's indexing
604 state in its own constructor, and its ``libraries`` property builds a
605 ``PhotoLibrary`` for every zone of the account in one pass -- each of
606 which raises this if that zone's index is unfinished, with nothing
607 cached on failure. So while any one zone is indexing, Photos cannot
608 be opened at all.
609 """
610 return isinstance(error, exceptions.ICloudPyServiceNotActivatedException) and "indexing" in str(error)
613def _raise_if_photos_indexing(error) -> None:
614 """Let a Photos-wide indexing wait escape the per-library handler.
616 ``_LIBRARY_FAULTS`` covers ServiceNotActivated so that one dead zone
617 cannot stop the others, but indexing is not one library's problem:
618 none of them can be read (see ``is_photos_indexing``). Isolating it
619 would mark every configured library failed and run the pass to
620 completion, when the whole of Photos simply has to wait.
621 """
622 if is_photos_indexing(error):
623 raise error
626def _note_library_failure(library, error, failed_libraries) -> None:
627 """Record a library we could not read, and say so in the log.
629 Never silent: the count is reported as failures by the caller, and the
630 library is barred from obsolete-file cleanup so an unreadable library
631 can never be mistaken for an empty one.
632 """
633 LOGGER.error(f"Library {library} could not be synced, skipping it: {error!s}")
634 if failed_libraries is not None:
635 failed_libraries.add(library)
638def _sync_all_photos_first_for_hardlinks(
639 photos,
640 libraries,
641 destination_path,
642 filters,
643 files,
644 folder_format,
645 hardlink_registry,
646 config,
647 library_destinations: dict | None = None,
648 failed_libraries: set | None = None,
649) -> tuple[int, int]:
650 """Sync 'All Photos' album first to populate hardlink registry.
652 Args:
653 photos: Photos object from iCloudPy
654 libraries: List of photo libraries to sync
655 destination_path: Base destination path
656 filters: Photo filters configuration
657 files: Set to track downloaded files
658 folder_format: Folder format string
659 hardlink_registry: Registry for tracking downloaded files
660 config: Configuration dictionary
662 Returns:
663 Tuple of (total_successful, total_failed) download counts
664 """
665 for library in libraries:
666 try:
667 if library == "PrimarySync" and "All Photos" in photos.libraries[library].albums:
668 LOGGER.info("Syncing 'All Photos' album first for hard link reference...")
669 lib_dest = _library_destination(destination_path, library, library_destinations or {})
670 result = sync_album_photos(
671 album=photos.libraries[library].albums["All Photos"],
672 destination_path=os.path.join(lib_dest, "All Photos"),
673 file_sizes=filters["file_sizes"],
674 extensions=filters["extensions"],
675 files=files,
676 folder_format=folder_format,
677 hardlink_registry=hardlink_registry,
678 config=config,
679 )
680 if hardlink_registry:
681 LOGGER.info(
682 f"'All Photos' sync complete. Hard link registry populated with "
683 f"{hardlink_registry.get_registry_size()} reference files.",
684 )
685 if result is not None:
686 return result
687 break
688 except _LIBRARY_FAULTS as e:
689 _raise_if_photos_indexing(e)
690 _note_library_failure(library, e, failed_libraries)
691 return 0, 0
694def _sync_albums_by_configuration(
695 photos,
696 libraries,
697 download_all,
698 destination_path,
699 filters,
700 files,
701 folder_format,
702 hardlink_registry,
703 config,
704 library_destinations: dict | None = None,
705 failed_libraries: set | None = None,
706) -> tuple[int, int]:
707 """Sync albums based on configuration settings.
709 Args:
710 photos: Photos object from iCloudPy
711 libraries: List of photo libraries to sync
712 download_all: Whether to download all albums
713 destination_path: Base destination path
714 filters: Photo filters configuration
715 files: Set to track downloaded files
716 folder_format: Folder format string
717 hardlink_registry: Registry for tracking downloaded files
718 config: Configuration dictionary
720 Returns:
721 Tuple of (total_successful, total_failed) aggregated across all libraries
722 """
723 total_successful, total_failed = 0, 0
724 for library in libraries:
725 try:
726 _signal_library("record_library_started", library)
727 lib_dest = _library_destination(destination_path, library, library_destinations or {})
728 if download_all and library == "PrimarySync":
729 sub_successful, sub_failed = _sync_all_albums_except_filtered(
730 photos,
731 library,
732 filters,
733 lib_dest,
734 files,
735 folder_format,
736 hardlink_registry,
737 config,
738 )
739 elif filters["albums"] and library == "PrimarySync":
740 sub_successful, sub_failed = _sync_filtered_albums(
741 photos,
742 library,
743 filters,
744 lib_dest,
745 files,
746 folder_format,
747 hardlink_registry,
748 config,
749 )
750 elif filters["albums"]:
751 sub_successful, sub_failed = _sync_filtered_albums_in_library(
752 photos,
753 library,
754 filters,
755 lib_dest,
756 files,
757 folder_format,
758 hardlink_registry,
759 config,
760 )
761 else:
762 sub_successful, sub_failed = _sync_all_photos_in_library(
763 photos,
764 library,
765 lib_dest,
766 filters,
767 files,
768 folder_format,
769 hardlink_registry,
770 config,
771 )
772 total_successful += sub_successful
773 total_failed += sub_failed
774 _signal_library(
775 "record_library_finished",
776 library,
777 ok=True,
778 downloaded=sub_successful,
779 skipped=sub_failed,
780 )
781 # Per-iteration by design: isolating one library from the next is
782 # the whole point, so the handler cannot be hoisted out of the loop.
783 except _LIBRARY_FAULTS as e: # noqa: PERF203
784 _raise_if_photos_indexing(e)
785 _note_library_failure(library, e, failed_libraries)
786 _signal_library("record_library_finished", library, ok=False, error=str(e))
787 return total_successful, total_failed
790def _sync_all_albums_except_filtered(
791 photos,
792 library,
793 filters,
794 destination_path,
795 files,
796 folder_format,
797 hardlink_registry,
798 config,
799) -> tuple[int, int]:
800 """Sync all albums except those in the filter exclusion list.
802 Args:
803 photos: Photos object from iCloudPy
804 library: Library name to sync
805 filters: Photo filters configuration
806 destination_path: Base destination path
807 files: Set to track downloaded files
808 folder_format: Folder format string
809 hardlink_registry: Registry for tracking downloaded files
810 config: Configuration dictionary
812 Returns:
813 Tuple of (total_successful, total_failed) aggregated across all synced albums
814 """
815 total_successful, total_failed = 0, 0
816 for album in photos.libraries[library].albums.keys():
817 # Skip All Photos if we already synced it first
818 if hardlink_registry and album == "All Photos":
819 continue
820 if filters["albums"] and album in iter(filters["albums"]):
821 continue
822 result = sync_album_photos(
823 album=photos.libraries[library].albums[album],
824 destination_path=os.path.join(destination_path, album),
825 file_sizes=filters["file_sizes"],
826 extensions=filters["extensions"],
827 files=files,
828 folder_format=folder_format,
829 hardlink_registry=hardlink_registry,
830 config=config,
831 )
832 if result is not None:
833 sub_successful, sub_failed = result
834 total_successful += sub_successful
835 total_failed += sub_failed
836 return total_successful, total_failed
839def _sync_filtered_albums(
840 photos,
841 library,
842 filters,
843 destination_path,
844 files,
845 folder_format,
846 hardlink_registry,
847 config,
848) -> tuple[int, int]:
849 """Sync only albums specified in filters.
851 Args:
852 photos: Photos object from iCloudPy
853 library: Library name to sync
854 filters: Photo filters configuration
855 destination_path: Base destination path
856 files: Set to track downloaded files
857 folder_format: Folder format string
858 hardlink_registry: Registry for tracking downloaded files
859 config: Configuration dictionary
861 Returns:
862 Tuple of (total_successful, total_failed) aggregated across all synced albums
863 """
864 total_successful, total_failed = 0, 0
865 for album in iter(filters["albums"]):
866 result = sync_album_photos(
867 album=photos.libraries[library].albums[album],
868 destination_path=os.path.join(destination_path, album),
869 file_sizes=filters["file_sizes"],
870 extensions=filters["extensions"],
871 files=files,
872 folder_format=folder_format,
873 hardlink_registry=hardlink_registry,
874 config=config,
875 )
876 if result is not None:
877 sub_successful, sub_failed = result
878 total_successful += sub_successful
879 total_failed += sub_failed
880 return total_successful, total_failed
883def _sync_filtered_albums_in_library(
884 photos,
885 library,
886 filters,
887 destination_path,
888 files,
889 folder_format,
890 hardlink_registry,
891 config,
892) -> tuple[int, int]:
893 """Sync filtered albums in a specific library.
895 Args:
896 photos: Photos object from iCloudPy
897 library: Library name to sync
898 filters: Photo filters configuration
899 destination_path: Base destination path
900 files: Set to track downloaded files
901 folder_format: Folder format string
902 hardlink_registry: Registry for tracking downloaded files
903 config: Configuration dictionary
905 Returns:
906 Tuple of (total_successful, total_failed) aggregated across all synced albums
907 """
908 total_successful, total_failed = 0, 0
909 for album in iter(filters["albums"]):
910 if album in photos.libraries[library].albums:
911 result = sync_album_photos(
912 album=photos.libraries[library].albums[album],
913 destination_path=os.path.join(destination_path, album),
914 file_sizes=filters["file_sizes"],
915 extensions=filters["extensions"],
916 files=files,
917 folder_format=folder_format,
918 hardlink_registry=hardlink_registry,
919 config=config,
920 )
921 if result is not None:
922 sub_successful, sub_failed = result
923 total_successful += sub_successful
924 total_failed += sub_failed
925 else:
926 LOGGER.warning(f"Album {album} not found in {library}. Skipping the album {album} ...")
927 return total_successful, total_failed
930def _sync_all_photos_in_library(
931 photos,
932 library,
933 destination_path,
934 filters,
935 files,
936 folder_format,
937 hardlink_registry,
938 config,
939) -> tuple[int, int]:
940 """Sync all photos in a library.
942 Args:
943 photos: Photos object from iCloudPy
944 library: Library name to sync
945 destination_path: Base destination path
946 filters: Photo filters configuration
947 files: Set to track downloaded files
948 folder_format: Folder format string
949 hardlink_registry: Registry for tracking downloaded files
950 config: Configuration dictionary
952 Returns:
953 Tuple of (total_successful, total_failed) download counts
954 """
955 result = sync_album_photos(
956 album=photos.libraries[library].all,
957 destination_path=os.path.join(destination_path, "all"),
958 file_sizes=filters["file_sizes"],
959 extensions=filters["extensions"],
960 files=files,
961 folder_format=folder_format,
962 hardlink_registry=hardlink_registry,
963 config=config,
964 )
965 if result is not None:
966 return result
967 return 0, 0