Coverage for src/sync_photos.py: 100%
206 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 17:25 +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
11from src import config_parser, configure_icloudpy_logging, get_logger
12from src.album_sync_orchestrator import sync_album_photos
13from src.hardlink_registry import create_hardlink_registry
14from src.photo_cleanup_utils import remove_obsolete_files
16# Configure icloudpy logging immediately after import
17configure_icloudpy_logging()
19LOGGER = get_logger()
22# Legacy functions preserved for backward compatibility with existing tests
23# These functions are now implemented using the new modular architecture
26def get_max_threads(config):
27 """Get maximum number of threads for parallel downloads.
29 Legacy function - now delegates to config_parser.
31 Args:
32 config: Configuration dictionary
34 Returns:
35 Maximum number of threads to use for downloads
36 """
37 return config_parser.get_app_max_threads(config)
40def get_name_and_extension(photo, file_size):
41 """Extract filename and extension.
43 Legacy function - now delegates to photo_path_utils.
45 Args:
46 photo: Photo object from iCloudPy
47 file_size: File size variant
49 Returns:
50 Tuple of (name, extension)
51 """
52 from src.photo_path_utils import get_photo_name_and_extension
54 return get_photo_name_and_extension(photo, file_size)
57def photo_wanted(photo, extensions):
58 """Check if photo is wanted based on extension.
60 Legacy function - now delegates to photo_filter_utils.
62 Args:
63 photo: Photo object from iCloudPy
64 extensions: List of allowed extensions
66 Returns:
67 True if photo should be synced, False otherwise
68 """
69 from src.photo_filter_utils import is_photo_wanted
71 return is_photo_wanted(photo, extensions)
74def generate_file_name(photo, file_size, destination_path, folder_format):
75 """Generate full path to file.
77 Legacy function - now delegates to photo_download_manager.
79 Args:
80 photo: Photo object from iCloudPy
81 file_size: File size variant
82 destination_path: Base destination path
83 folder_format: Folder format string
85 Returns:
86 Full file path
87 """
88 from src.photo_download_manager import generate_photo_path
90 return generate_photo_path(photo, file_size, destination_path, folder_format)
93def photo_exists(photo, file_size, local_path):
94 """Check if photo exist locally.
96 Legacy function - now delegates to photo_file_utils.
98 Args:
99 photo: Photo object from iCloudPy
100 file_size: File size variant
101 local_path: Local file path to check
103 Returns:
104 True if photo exists with correct size, False otherwise
105 """
106 from src.photo_file_utils import check_photo_exists
108 return check_photo_exists(photo, file_size, local_path)
111def create_hardlink(source_path, destination_path):
112 """Create a hard link from source to destination.
114 Legacy function - now delegates to photo_file_utils.
116 Args:
117 source_path: Path to source file
118 destination_path: Path for new hardlink
120 Returns:
121 True if successful, False otherwise
122 """
123 from src.photo_file_utils import create_hardlink as create_hardlink_impl
125 return create_hardlink_impl(source_path, destination_path)
128def download_photo(photo, file_size, destination_path):
129 """Download photo from server.
131 Legacy function - now delegates to photo_file_utils.
133 Args:
134 photo: Photo object from iCloudPy
135 file_size: File size variant
136 destination_path: Where to save the photo
138 Returns:
139 True if successful, False otherwise
140 """
141 from src.photo_file_utils import download_photo_from_server
143 return download_photo_from_server(photo, file_size, destination_path)
146def process_photo(photo, file_size, destination_path, files, folder_format, hardlink_registry=None):
147 """Process photo details (legacy function for backward compatibility).
149 Args:
150 photo: Photo object from iCloudPy
151 file_size: File size variant
152 destination_path: Base destination path
153 files: Set to track downloaded files
154 folder_format: Folder format string
155 hardlink_registry: Registry for hardlinks (legacy dict format)
157 Returns:
158 True if photo was processed successfully, False otherwise
159 """
160 from src.photo_download_manager import collect_download_task, execute_download_task
162 # Convert legacy hardlink registry dict to new registry format if needed
163 converted_registry = None
164 if hardlink_registry is not None:
165 from src.hardlink_registry import HardlinkRegistry
167 converted_registry = HardlinkRegistry()
168 for key, path in hardlink_registry.items():
169 # Legacy format: photo_id_file_size -> path
170 if "_" in key:
171 parts = key.rsplit("_", 1)
172 if len(parts) == 2:
173 photo_id, file_sz = parts
174 converted_registry.register_photo_path(photo_id, file_sz, path)
176 # Collect download task
177 task_info = collect_download_task(
178 photo,
179 file_size,
180 destination_path,
181 files,
182 folder_format,
183 converted_registry,
184 )
186 if task_info is None:
187 return False
189 # Execute task
190 result = execute_download_task(task_info)
192 # Update legacy registry if provided
193 if result and hardlink_registry is not None:
194 photo_key = f"{photo.id}_{file_size}"
195 hardlink_registry[photo_key] = task_info.photo_path
197 return result
200def collect_photo_for_download(photo, file_size, destination_path, files, folder_format, hardlink_registry=None):
201 """Collect photo info for parallel download without immediately downloading.
203 Legacy function - now delegates to photo_download_manager.
205 Args:
206 photo: Photo object from iCloudPy
207 file_size: File size variant
208 destination_path: Base destination path
209 files: Set to track downloaded files
210 folder_format: Folder format string
211 hardlink_registry: Registry for hardlinks (legacy dict format)
213 Returns:
214 Download task info or None
215 """
216 from src.photo_download_manager import collect_download_task
218 # Convert legacy hardlink registry dict to new registry format if needed
219 converted_registry = None
220 if hardlink_registry is not None:
221 from src.hardlink_registry import HardlinkRegistry
223 converted_registry = HardlinkRegistry()
224 for key, path in hardlink_registry.items():
225 if "_" in key:
226 parts = key.rsplit("_", 1)
227 if len(parts) == 2:
228 photo_id, file_sz = parts
229 converted_registry.register_photo_path(photo_id, file_sz, path)
231 task_info = collect_download_task(
232 photo,
233 file_size,
234 destination_path,
235 files,
236 folder_format,
237 converted_registry,
238 )
240 if task_info is None:
241 return None
243 # Convert back to legacy format for compatibility
244 return {
245 "photo": task_info.photo,
246 "file_size": task_info.file_size,
247 "photo_path": task_info.photo_path,
248 "hardlink_source": task_info.hardlink_source,
249 "hardlink_registry": hardlink_registry,
250 }
253def download_photo_task(download_info):
254 """Download a single photo or create hardlink as part of parallel execution.
256 Legacy function - maintains original implementation for backward compatibility.
258 Args:
259 download_info: Dictionary with download task information
261 Returns:
262 True if successful, False otherwise
263 """
264 photo = download_info["photo"]
265 file_size = download_info["file_size"]
266 photo_path = download_info["photo_path"]
267 hardlink_source = download_info.get("hardlink_source")
268 hardlink_registry = download_info.get("hardlink_registry")
270 LOGGER.debug(f"[Thread] Starting processing of {photo_path}")
272 try:
273 # Try hardlink first if source exists
274 if hardlink_source:
275 if create_hardlink(hardlink_source, photo_path):
276 LOGGER.debug(f"[Thread] Created hardlink for {photo_path}")
277 return True
278 else:
279 # Fallback to download if hard link creation fails
280 LOGGER.warning(f"Hard link creation failed, downloading {photo_path} instead")
282 # Download the photo - this maintains the original function call for test compatibility
283 result = download_photo(photo, file_size, photo_path)
284 if result:
285 # Register for future hard links if enabled
286 if hardlink_registry is not None:
287 photo_key = f"{photo.id}_{file_size}"
288 hardlink_registry[photo_key] = photo_path
289 LOGGER.debug(f"[Thread] Completed download of {photo_path}")
290 return result
291 except Exception as e:
292 LOGGER.error(f"[Thread] Failed to process {photo_path}: {e!s}")
293 return False
296def sync_album(
297 album,
298 destination_path,
299 file_sizes,
300 extensions=None,
301 files=None,
302 folder_format=None,
303 hardlink_registry=None,
304 config=None,
305):
306 """Sync given album.
308 Legacy function - now delegates to album_sync_orchestrator with conversion
309 for legacy hardlink registry format.
311 Args:
312 album: Album object from iCloudPy
313 destination_path: Path where photos should be saved
314 file_sizes: List of file size variants to download
315 extensions: List of allowed file extensions
316 files: Set to track downloaded files
317 folder_format: Folder format string
318 hardlink_registry: Registry for hardlinks (legacy dict format)
319 config: Configuration dictionary
321 Returns:
322 True on success, None on invalid input
323 """
324 # Convert legacy hardlink registry dict to new registry format if needed
325 converted_registry = None
326 if hardlink_registry is not None:
327 from src.hardlink_registry import HardlinkRegistry
329 converted_registry = HardlinkRegistry()
330 for key, path in hardlink_registry.items():
331 if "_" in key:
332 parts = key.rsplit("_", 1)
333 if len(parts) == 2:
334 photo_id, file_sz = parts
335 converted_registry.register_photo_path(photo_id, file_sz, path)
337 result = sync_album_photos(
338 album=album,
339 destination_path=destination_path,
340 file_sizes=file_sizes,
341 extensions=extensions,
342 files=files,
343 folder_format=folder_format,
344 hardlink_registry=converted_registry,
345 config=config,
346 )
348 # Update legacy registry if provided and new registry was created
349 if hardlink_registry is not None and converted_registry is not None:
350 # This is a simplified approach - in practice, we'd need to track new entries
351 # But for legacy compatibility, we'll maintain the existing behavior
352 pass
354 return result
357def remove_obsolete(destination_path, files):
358 """Remove local obsolete file.
360 Legacy function - now delegates to photo_cleanup_utils.
362 Args:
363 destination_path: Path to search for obsolete files
364 files: Set of files that should be kept
366 Returns:
367 Set of removed file paths
368 """
369 return remove_obsolete_files(destination_path, files)
372def sync_photos(config, photos):
373 """Sync all photos.
375 Main orchestration function that coordinates the entire photo sync process.
376 This function has been refactored to use the new modular architecture while
377 maintaining backward compatibility.
379 Args:
380 config: Configuration dictionary
381 photos: Photos object from iCloudPy
383 Returns:
384 Tuple of (total_successful, total_failed) download counts
385 """
386 # Parse configuration using centralized config parser
387 destination_path = config_parser.prepare_photos_destination(config=config)
388 library_destinations = config_parser.get_photos_library_destinations(config=config)
389 filters = config_parser.get_photos_filters(config=config)
390 files = set()
391 download_all = config_parser.get_photos_all_albums(config=config)
392 use_hardlinks = config_parser.get_photos_use_hardlinks(config=config)
393 libraries = filters["libraries"] if filters["libraries"] is not None else photos.libraries
394 folder_format = config_parser.get_photos_folder_format(config=config)
396 # Initialize hard link registry using new modular approach
397 hardlink_registry = create_hardlink_registry(use_hardlinks)
399 total_successful, total_failed = 0, 0
401 # Special handling for "All Photos" when hardlinks are enabled
402 if use_hardlinks and download_all:
403 sub_successful, sub_failed = _sync_all_photos_first_for_hardlinks(
404 photos,
405 libraries,
406 destination_path,
407 filters,
408 files,
409 folder_format,
410 hardlink_registry,
411 config,
412 library_destinations=library_destinations,
413 )
414 total_successful += sub_successful
415 total_failed += sub_failed
417 # Sync albums based on configuration
418 sub_successful, sub_failed = _sync_albums_by_configuration(
419 photos,
420 libraries,
421 download_all,
422 destination_path,
423 filters,
424 files,
425 folder_format,
426 hardlink_registry,
427 config,
428 library_destinations=library_destinations,
429 )
430 total_successful += sub_successful
431 total_failed += sub_failed
433 # Clean up obsolete files if enabled. When per-library destinations are
434 # configured we walk each library's subdir independently, otherwise the
435 # legacy single-destination walk preserves backward compatibility.
436 if config_parser.get_photos_remove_obsolete(config=config):
437 marker_filename = config_parser.get_mount_marker_filename(config=config)
438 exclude = {marker_filename}
439 if library_destinations:
440 for library in libraries:
441 lib_dest = _library_destination(destination_path, library, library_destinations)
442 remove_obsolete_files(lib_dest, files, exclude_filenames=exclude)
443 else:
444 remove_obsolete_files(destination_path, files, exclude_filenames=exclude)
446 return total_successful, total_failed
449def _library_destination(base_destination: str, library: str, library_destinations: dict) -> str:
450 """Resolve the on-disk destination for a given iCloud photo library.
452 When ``library_destinations`` provides a mapping for ``library``, joins
453 the configured subdirectory under ``base_destination`` and ensures the
454 directory exists. Otherwise returns ``base_destination`` unchanged
455 (preserving mandarons' legacy single-destination behaviour).
457 Library-name matching has three rules, in priority order:
459 1. **Exact match.** ``library_destinations[library]`` if present.
460 2. **Role alias for `SharedLibrary`.** Apple's modern iCloud Shared
461 Photo Library is exposed by icloudpy under a GUID-based zone name
462 like ``SharedSync-3C977B4A-C15A-46E4-9854-585B9342C409``. A config
463 key of ``SharedLibrary`` matches any zone whose name starts with
464 ``SharedSync-`` so users don't need to discover and hardcode the
465 per-account GUID. (Configs that already use the literal current
466 Apple zone name still work via rule 1.)
467 3. **Fallthrough.** Returns ``base_destination`` unchanged.
468 """
469 if not library_destinations:
470 return base_destination
471 # ``get`` returns None only when the key is absent -- the config parser
472 # coerces every configured value to str, so an explicit "" stays "".
473 # Key on ``is None`` (not falsiness) so an explicitly-mapped library
474 # always wins over the SharedLibrary alias and the default, honouring
475 # the rule 1 > rule 2 > rule 3 priority documented above.
476 subdir = library_destinations.get(library)
477 if subdir is None and library.startswith("SharedSync-"):
478 subdir = library_destinations.get("SharedLibrary")
479 if subdir is None:
480 return base_destination
481 dest = os.path.join(base_destination, subdir)
482 os.makedirs(dest, exist_ok=True)
483 return dest
486def _sync_all_photos_first_for_hardlinks(
487 photos,
488 libraries,
489 destination_path,
490 filters,
491 files,
492 folder_format,
493 hardlink_registry,
494 config,
495 library_destinations: dict | None = None,
496) -> tuple[int, int]:
497 """Sync 'All Photos' album first to populate hardlink registry.
499 Args:
500 photos: Photos object from iCloudPy
501 libraries: List of photo libraries to sync
502 destination_path: Base destination path
503 filters: Photo filters configuration
504 files: Set to track downloaded files
505 folder_format: Folder format string
506 hardlink_registry: Registry for tracking downloaded files
507 config: Configuration dictionary
509 Returns:
510 Tuple of (total_successful, total_failed) download counts
511 """
512 for library in libraries:
513 if library == "PrimarySync" and "All Photos" in photos.libraries[library].albums:
514 LOGGER.info("Syncing 'All Photos' album first for hard link reference...")
515 lib_dest = _library_destination(destination_path, library, library_destinations or {})
516 result = sync_album_photos(
517 album=photos.libraries[library].albums["All Photos"],
518 destination_path=os.path.join(lib_dest, "All Photos"),
519 file_sizes=filters["file_sizes"],
520 extensions=filters["extensions"],
521 files=files,
522 folder_format=folder_format,
523 hardlink_registry=hardlink_registry,
524 config=config,
525 )
526 if hardlink_registry:
527 LOGGER.info(
528 f"'All Photos' sync complete. Hard link registry populated with "
529 f"{hardlink_registry.get_registry_size()} reference files.",
530 )
531 if result is not None:
532 return result
533 break
534 return 0, 0
537def _sync_albums_by_configuration(
538 photos,
539 libraries,
540 download_all,
541 destination_path,
542 filters,
543 files,
544 folder_format,
545 hardlink_registry,
546 config,
547 library_destinations: dict | None = None,
548) -> tuple[int, int]:
549 """Sync albums based on configuration settings.
551 Args:
552 photos: Photos object from iCloudPy
553 libraries: List of photo libraries to sync
554 download_all: Whether to download all albums
555 destination_path: Base destination path
556 filters: Photo filters configuration
557 files: Set to track downloaded files
558 folder_format: Folder format string
559 hardlink_registry: Registry for tracking downloaded files
560 config: Configuration dictionary
562 Returns:
563 Tuple of (total_successful, total_failed) aggregated across all libraries
564 """
565 total_successful, total_failed = 0, 0
566 for library in libraries:
567 lib_dest = _library_destination(destination_path, library, library_destinations or {})
568 if download_all and library == "PrimarySync":
569 sub_successful, sub_failed = _sync_all_albums_except_filtered(
570 photos,
571 library,
572 filters,
573 lib_dest,
574 files,
575 folder_format,
576 hardlink_registry,
577 config,
578 )
579 elif filters["albums"] and library == "PrimarySync":
580 sub_successful, sub_failed = _sync_filtered_albums(
581 photos,
582 library,
583 filters,
584 lib_dest,
585 files,
586 folder_format,
587 hardlink_registry,
588 config,
589 )
590 elif filters["albums"]:
591 sub_successful, sub_failed = _sync_filtered_albums_in_library(
592 photos,
593 library,
594 filters,
595 lib_dest,
596 files,
597 folder_format,
598 hardlink_registry,
599 config,
600 )
601 else:
602 sub_successful, sub_failed = _sync_all_photos_in_library(
603 photos,
604 library,
605 lib_dest,
606 filters,
607 files,
608 folder_format,
609 hardlink_registry,
610 config,
611 )
612 total_successful += sub_successful
613 total_failed += sub_failed
614 return total_successful, total_failed
617def _sync_all_albums_except_filtered(
618 photos,
619 library,
620 filters,
621 destination_path,
622 files,
623 folder_format,
624 hardlink_registry,
625 config,
626) -> tuple[int, int]:
627 """Sync all albums except those in the filter exclusion list.
629 Args:
630 photos: Photos object from iCloudPy
631 library: Library name to sync
632 filters: Photo filters configuration
633 destination_path: Base destination path
634 files: Set to track downloaded files
635 folder_format: Folder format string
636 hardlink_registry: Registry for tracking downloaded files
637 config: Configuration dictionary
639 Returns:
640 Tuple of (total_successful, total_failed) aggregated across all synced albums
641 """
642 total_successful, total_failed = 0, 0
643 for album in photos.libraries[library].albums.keys():
644 # Skip All Photos if we already synced it first
645 if hardlink_registry and album == "All Photos":
646 continue
647 if filters["albums"] and album in iter(filters["albums"]):
648 continue
649 result = sync_album_photos(
650 album=photos.libraries[library].albums[album],
651 destination_path=os.path.join(destination_path, album),
652 file_sizes=filters["file_sizes"],
653 extensions=filters["extensions"],
654 files=files,
655 folder_format=folder_format,
656 hardlink_registry=hardlink_registry,
657 config=config,
658 )
659 if result is not None:
660 sub_successful, sub_failed = result
661 total_successful += sub_successful
662 total_failed += sub_failed
663 return total_successful, total_failed
666def _sync_filtered_albums(
667 photos,
668 library,
669 filters,
670 destination_path,
671 files,
672 folder_format,
673 hardlink_registry,
674 config,
675) -> tuple[int, int]:
676 """Sync only albums specified in filters.
678 Args:
679 photos: Photos object from iCloudPy
680 library: Library name to sync
681 filters: Photo filters configuration
682 destination_path: Base destination path
683 files: Set to track downloaded files
684 folder_format: Folder format string
685 hardlink_registry: Registry for tracking downloaded files
686 config: Configuration dictionary
688 Returns:
689 Tuple of (total_successful, total_failed) aggregated across all synced albums
690 """
691 total_successful, total_failed = 0, 0
692 for album in iter(filters["albums"]):
693 result = sync_album_photos(
694 album=photos.libraries[library].albums[album],
695 destination_path=os.path.join(destination_path, album),
696 file_sizes=filters["file_sizes"],
697 extensions=filters["extensions"],
698 files=files,
699 folder_format=folder_format,
700 hardlink_registry=hardlink_registry,
701 config=config,
702 )
703 if result is not None:
704 sub_successful, sub_failed = result
705 total_successful += sub_successful
706 total_failed += sub_failed
707 return total_successful, total_failed
710def _sync_filtered_albums_in_library(
711 photos,
712 library,
713 filters,
714 destination_path,
715 files,
716 folder_format,
717 hardlink_registry,
718 config,
719) -> tuple[int, int]:
720 """Sync filtered albums in a specific library.
722 Args:
723 photos: Photos object from iCloudPy
724 library: Library name to sync
725 filters: Photo filters configuration
726 destination_path: Base destination path
727 files: Set to track downloaded files
728 folder_format: Folder format string
729 hardlink_registry: Registry for tracking downloaded files
730 config: Configuration dictionary
732 Returns:
733 Tuple of (total_successful, total_failed) aggregated across all synced albums
734 """
735 total_successful, total_failed = 0, 0
736 for album in iter(filters["albums"]):
737 if album in photos.libraries[library].albums:
738 result = sync_album_photos(
739 album=photos.libraries[library].albums[album],
740 destination_path=os.path.join(destination_path, album),
741 file_sizes=filters["file_sizes"],
742 extensions=filters["extensions"],
743 files=files,
744 folder_format=folder_format,
745 hardlink_registry=hardlink_registry,
746 config=config,
747 )
748 if result is not None:
749 sub_successful, sub_failed = result
750 total_successful += sub_successful
751 total_failed += sub_failed
752 else:
753 LOGGER.warning(f"Album {album} not found in {library}. Skipping the album {album} ...")
754 return total_successful, total_failed
757def _sync_all_photos_in_library(
758 photos,
759 library,
760 destination_path,
761 filters,
762 files,
763 folder_format,
764 hardlink_registry,
765 config,
766) -> tuple[int, int]:
767 """Sync all photos in a library.
769 Args:
770 photos: Photos object from iCloudPy
771 library: Library name to sync
772 destination_path: Base destination path
773 filters: Photo filters configuration
774 files: Set to track downloaded files
775 folder_format: Folder format string
776 hardlink_registry: Registry for tracking downloaded files
777 config: Configuration dictionary
779 Returns:
780 Tuple of (total_successful, total_failed) download counts
781 """
782 result = sync_album_photos(
783 album=photos.libraries[library].all,
784 destination_path=os.path.join(destination_path, "all"),
785 file_sizes=filters["file_sizes"],
786 extensions=filters["extensions"],
787 files=files,
788 folder_format=folder_format,
789 hardlink_registry=hardlink_registry,
790 config=config,
791 )
792 if result is not None:
793 return result
794 return 0, 0