Coverage for src/sync_photos.py: 100%
204 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +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 if library_destinations:
438 for library in libraries:
439 lib_dest = _library_destination(destination_path, library, library_destinations)
440 remove_obsolete_files(lib_dest, files)
441 else:
442 remove_obsolete_files(destination_path, files)
444 return total_successful, total_failed
447def _library_destination(base_destination: str, library: str, library_destinations: dict) -> str:
448 """Resolve the on-disk destination for a given iCloud photo library.
450 When ``library_destinations`` provides a mapping for ``library``, joins
451 the configured subdirectory under ``base_destination`` and ensures the
452 directory exists. Otherwise returns ``base_destination`` unchanged
453 (preserving mandarons' legacy single-destination behaviour).
455 Library-name matching has three rules, in priority order:
457 1. **Exact match.** ``library_destinations[library]`` if present.
458 2. **Role alias for `SharedLibrary`.** Apple's modern iCloud Shared
459 Photo Library is exposed by icloudpy under a GUID-based zone name
460 like ``SharedSync-3C977B4A-C15A-46E4-9854-585B9342C409``. A config
461 key of ``SharedLibrary`` matches any zone whose name starts with
462 ``SharedSync-`` so users don't need to discover and hardcode the
463 per-account GUID. (Configs that already use the literal current
464 Apple zone name still work via rule 1.)
465 3. **Fallthrough.** Returns ``base_destination`` unchanged.
466 """
467 if not library_destinations:
468 return base_destination
469 # ``get`` returns None only when the key is absent -- the config parser
470 # coerces every configured value to str, so an explicit "" stays "".
471 # Key on ``is None`` (not falsiness) so an explicitly-mapped library
472 # always wins over the SharedLibrary alias and the default, honouring
473 # the rule 1 > rule 2 > rule 3 priority documented above.
474 subdir = library_destinations.get(library)
475 if subdir is None and library.startswith("SharedSync-"):
476 subdir = library_destinations.get("SharedLibrary")
477 if subdir is None:
478 return base_destination
479 dest = os.path.join(base_destination, subdir)
480 os.makedirs(dest, exist_ok=True)
481 return dest
484def _sync_all_photos_first_for_hardlinks(
485 photos,
486 libraries,
487 destination_path,
488 filters,
489 files,
490 folder_format,
491 hardlink_registry,
492 config,
493 library_destinations: dict | None = None,
494) -> tuple[int, int]:
495 """Sync 'All Photos' album first to populate hardlink registry.
497 Args:
498 photos: Photos object from iCloudPy
499 libraries: List of photo libraries to sync
500 destination_path: Base destination path
501 filters: Photo filters configuration
502 files: Set to track downloaded files
503 folder_format: Folder format string
504 hardlink_registry: Registry for tracking downloaded files
505 config: Configuration dictionary
507 Returns:
508 Tuple of (total_successful, total_failed) download counts
509 """
510 for library in libraries:
511 if library == "PrimarySync" and "All Photos" in photos.libraries[library].albums:
512 LOGGER.info("Syncing 'All Photos' album first for hard link reference...")
513 lib_dest = _library_destination(destination_path, library, library_destinations or {})
514 result = sync_album_photos(
515 album=photos.libraries[library].albums["All Photos"],
516 destination_path=os.path.join(lib_dest, "All Photos"),
517 file_sizes=filters["file_sizes"],
518 extensions=filters["extensions"],
519 files=files,
520 folder_format=folder_format,
521 hardlink_registry=hardlink_registry,
522 config=config,
523 )
524 if hardlink_registry:
525 LOGGER.info(
526 f"'All Photos' sync complete. Hard link registry populated with "
527 f"{hardlink_registry.get_registry_size()} reference files.",
528 )
529 if result is not None:
530 return result
531 break
532 return 0, 0
535def _sync_albums_by_configuration(
536 photos,
537 libraries,
538 download_all,
539 destination_path,
540 filters,
541 files,
542 folder_format,
543 hardlink_registry,
544 config,
545 library_destinations: dict | None = None,
546) -> tuple[int, int]:
547 """Sync albums based on configuration settings.
549 Args:
550 photos: Photos object from iCloudPy
551 libraries: List of photo libraries to sync
552 download_all: Whether to download all albums
553 destination_path: Base destination path
554 filters: Photo filters configuration
555 files: Set to track downloaded files
556 folder_format: Folder format string
557 hardlink_registry: Registry for tracking downloaded files
558 config: Configuration dictionary
560 Returns:
561 Tuple of (total_successful, total_failed) aggregated across all libraries
562 """
563 total_successful, total_failed = 0, 0
564 for library in libraries:
565 lib_dest = _library_destination(destination_path, library, library_destinations or {})
566 if download_all and library == "PrimarySync":
567 sub_successful, sub_failed = _sync_all_albums_except_filtered(
568 photos,
569 library,
570 filters,
571 lib_dest,
572 files,
573 folder_format,
574 hardlink_registry,
575 config,
576 )
577 elif filters["albums"] and library == "PrimarySync":
578 sub_successful, sub_failed = _sync_filtered_albums(
579 photos,
580 library,
581 filters,
582 lib_dest,
583 files,
584 folder_format,
585 hardlink_registry,
586 config,
587 )
588 elif filters["albums"]:
589 sub_successful, sub_failed = _sync_filtered_albums_in_library(
590 photos,
591 library,
592 filters,
593 lib_dest,
594 files,
595 folder_format,
596 hardlink_registry,
597 config,
598 )
599 else:
600 sub_successful, sub_failed = _sync_all_photos_in_library(
601 photos,
602 library,
603 lib_dest,
604 filters,
605 files,
606 folder_format,
607 hardlink_registry,
608 config,
609 )
610 total_successful += sub_successful
611 total_failed += sub_failed
612 return total_successful, total_failed
615def _sync_all_albums_except_filtered(
616 photos,
617 library,
618 filters,
619 destination_path,
620 files,
621 folder_format,
622 hardlink_registry,
623 config,
624) -> tuple[int, int]:
625 """Sync all albums except those in the filter exclusion list.
627 Args:
628 photos: Photos object from iCloudPy
629 library: Library name to sync
630 filters: Photo filters configuration
631 destination_path: Base destination path
632 files: Set to track downloaded files
633 folder_format: Folder format string
634 hardlink_registry: Registry for tracking downloaded files
635 config: Configuration dictionary
637 Returns:
638 Tuple of (total_successful, total_failed) aggregated across all synced albums
639 """
640 total_successful, total_failed = 0, 0
641 for album in photos.libraries[library].albums.keys():
642 # Skip All Photos if we already synced it first
643 if hardlink_registry and album == "All Photos":
644 continue
645 if filters["albums"] and album in iter(filters["albums"]):
646 continue
647 result = sync_album_photos(
648 album=photos.libraries[library].albums[album],
649 destination_path=os.path.join(destination_path, album),
650 file_sizes=filters["file_sizes"],
651 extensions=filters["extensions"],
652 files=files,
653 folder_format=folder_format,
654 hardlink_registry=hardlink_registry,
655 config=config,
656 )
657 if result is not None:
658 sub_successful, sub_failed = result
659 total_successful += sub_successful
660 total_failed += sub_failed
661 return total_successful, total_failed
664def _sync_filtered_albums(
665 photos,
666 library,
667 filters,
668 destination_path,
669 files,
670 folder_format,
671 hardlink_registry,
672 config,
673) -> tuple[int, int]:
674 """Sync only albums specified in filters.
676 Args:
677 photos: Photos object from iCloudPy
678 library: Library name to sync
679 filters: Photo filters configuration
680 destination_path: Base destination path
681 files: Set to track downloaded files
682 folder_format: Folder format string
683 hardlink_registry: Registry for tracking downloaded files
684 config: Configuration dictionary
686 Returns:
687 Tuple of (total_successful, total_failed) aggregated across all synced albums
688 """
689 total_successful, total_failed = 0, 0
690 for album in iter(filters["albums"]):
691 result = sync_album_photos(
692 album=photos.libraries[library].albums[album],
693 destination_path=os.path.join(destination_path, album),
694 file_sizes=filters["file_sizes"],
695 extensions=filters["extensions"],
696 files=files,
697 folder_format=folder_format,
698 hardlink_registry=hardlink_registry,
699 config=config,
700 )
701 if result is not None:
702 sub_successful, sub_failed = result
703 total_successful += sub_successful
704 total_failed += sub_failed
705 return total_successful, total_failed
708def _sync_filtered_albums_in_library(
709 photos,
710 library,
711 filters,
712 destination_path,
713 files,
714 folder_format,
715 hardlink_registry,
716 config,
717) -> tuple[int, int]:
718 """Sync filtered albums in a specific library.
720 Args:
721 photos: Photos object from iCloudPy
722 library: Library name to sync
723 filters: Photo filters configuration
724 destination_path: Base destination path
725 files: Set to track downloaded files
726 folder_format: Folder format string
727 hardlink_registry: Registry for tracking downloaded files
728 config: Configuration dictionary
730 Returns:
731 Tuple of (total_successful, total_failed) aggregated across all synced albums
732 """
733 total_successful, total_failed = 0, 0
734 for album in iter(filters["albums"]):
735 if album in photos.libraries[library].albums:
736 result = sync_album_photos(
737 album=photos.libraries[library].albums[album],
738 destination_path=os.path.join(destination_path, album),
739 file_sizes=filters["file_sizes"],
740 extensions=filters["extensions"],
741 files=files,
742 folder_format=folder_format,
743 hardlink_registry=hardlink_registry,
744 config=config,
745 )
746 if result is not None:
747 sub_successful, sub_failed = result
748 total_successful += sub_successful
749 total_failed += sub_failed
750 else:
751 LOGGER.warning(f"Album {album} not found in {library}. Skipping the album {album} ...")
752 return total_successful, total_failed
755def _sync_all_photos_in_library(
756 photos,
757 library,
758 destination_path,
759 filters,
760 files,
761 folder_format,
762 hardlink_registry,
763 config,
764) -> tuple[int, int]:
765 """Sync all photos in a library.
767 Args:
768 photos: Photos object from iCloudPy
769 library: Library name to sync
770 destination_path: Base destination path
771 filters: Photo filters configuration
772 files: Set to track downloaded files
773 folder_format: Folder format string
774 hardlink_registry: Registry for tracking downloaded files
775 config: Configuration dictionary
777 Returns:
778 Tuple of (total_successful, total_failed) download counts
779 """
780 result = sync_album_photos(
781 album=photos.libraries[library].all,
782 destination_path=os.path.join(destination_path, "all"),
783 file_sizes=filters["file_sizes"],
784 extensions=filters["extensions"],
785 files=files,
786 folder_format=folder_format,
787 hardlink_registry=hardlink_registry,
788 config=config,
789 )
790 if result is not None:
791 return result
792 return 0, 0