Coverage for src/album_sync_orchestrator.py: 100%
68 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +0000
1"""Album synchronization orchestration module.
3This module contains the main album sync orchestration logic
4that coordinates photo filtering, download collection, and parallel execution.
5"""
7___author___ = "Mandar Patil <mandarons@pm.me>"
9import os
10from typing import Any
12from src import DEFAULT_ENUMERATION_CHUNK_SIZE, config_parser, get_logger
13from src.hardlink_registry import HardlinkRegistry
14from src.photo_download_manager import (
15 DownloadTaskInfo,
16 collect_download_task,
17 execute_parallel_downloads,
18)
19from src.photo_filter_utils import is_photo_wanted
20from src.photo_path_utils import normalize_file_path
22LOGGER = get_logger()
24# DEFAULT_ENUMERATION_CHUNK_SIZE lives in src/__init__.py (with the other
25# DEFAULT_* config constants) and is re-exported here for backward-compat.
26# Picked to keep peak RSS bounded at ~10–20 MB of DownloadTaskInfo objects
27# per chunk on typical iCloud libraries while still giving
28# execute_parallel_downloads enough work to amortize HTTP connection setup.
29# Users can override via ``photos.enumeration_chunk_size`` in config.yaml.
32def sync_album_photos(
33 album,
34 destination_path: str,
35 file_sizes: list[str],
36 extensions: list[str] | None = None,
37 files: set[str] | None = None,
38 folder_format: str | None = None,
39 hardlink_registry: HardlinkRegistry | None = None,
40 config=None,
41) -> tuple[int, int] | None:
42 """Sync photos from given album.
44 This function orchestrates the synchronization of a single album by:
45 1. Creating the destination directory
46 2. Collecting download tasks for wanted photos
47 3. Executing downloads in parallel
48 4. Recursively syncing subalbums
50 Args:
51 album: Album object from iCloudPy
52 destination_path: Path where photos should be saved
53 file_sizes: List of file size variants to download
54 extensions: List of allowed file extensions (None = all allowed)
55 files: Set to track downloaded files
56 folder_format: strftime format string for folder organization
57 hardlink_registry: Registry for tracking downloaded files for hardlinks
58 config: Configuration dictionary
60 Returns:
61 Tuple of (total_successful, total_failed) download counts, or None on invalid input
62 """
63 if album is None or destination_path is None or file_sizes is None:
64 return None
66 # Create destination directory with normalized path
67 normalized_destination = normalize_file_path(destination_path)
68 os.makedirs(normalized_destination, exist_ok=True)
69 LOGGER.info(f"Syncing {album.title}")
71 # Stream the album in fixed-size chunks: collect → download →
72 # release → next chunk. Memory is bounded by chunk_size × per-task
73 # size (~10 MB at chunk=1000) instead of len(album) × per-task size
74 # (which OOM-kills containers on ~100K+ libraries: empirically a
75 # 111K-photo library peaks at ~4 GB RSS without chunking, kernel-
76 # confirmed via cgroup OOM at the 4 GB cap).
77 chunk_size = config_parser.get_photos_enumeration_chunk_size(config=config)
78 total_successful, total_failed = _collect_and_execute_album_in_chunks(
79 album,
80 normalized_destination,
81 file_sizes,
82 extensions,
83 files,
84 folder_format,
85 hardlink_registry,
86 config,
87 chunk_size=chunk_size,
88 )
90 # Recursively sync subalbums and aggregate counts
91 sub_successful, sub_failed = _sync_subalbums(
92 album,
93 normalized_destination,
94 file_sizes,
95 extensions,
96 files,
97 folder_format,
98 hardlink_registry,
99 config,
100 )
101 total_successful += sub_successful
102 total_failed += sub_failed
104 return total_successful, total_failed
107def _collect_photo_download_tasks(
108 photo: Any,
109 destination_path: str,
110 file_sizes: list[str],
111 extensions: list[str] | None,
112 files: set[str] | None,
113 folder_format: str | None,
114 hardlink_registry: HardlinkRegistry | None,
115) -> list[DownloadTaskInfo]:
116 """Collect download tasks for a single photo, handling errors gracefully.
118 Wraps per-photo processing so that exceptions (e.g. binascii.Error from
119 iCloudPy's base64-encoded filename decoding) are caught at the photo level
120 rather than inside the album iteration loop (avoids PERF203).
122 Args:
123 photo: Photo object from iCloudPy
124 destination_path: Path where photos should be saved
125 file_sizes: List of file size variants to download
126 extensions: List of allowed file extensions
127 files: Set to track downloaded files
128 folder_format: strftime format string for folder organization
129 hardlink_registry: Registry for tracking downloaded files
131 Returns:
132 List of download tasks for this photo (empty on error or if unwanted)
133 """
134 try:
135 if not is_photo_wanted(photo, extensions):
136 LOGGER.debug(f"Skipping the unwanted photo {photo.filename}.")
137 return []
138 tasks: list[DownloadTaskInfo] = []
139 for file_size in file_sizes:
140 download_info = collect_download_task(
141 photo,
142 file_size,
143 destination_path,
144 files,
145 folder_format,
146 hardlink_registry,
147 )
148 if download_info:
149 tasks.append(download_info)
150 # Live Photos: add "live_video_original" (or _medium/_thumb) to
151 # photos.filters.file_sizes to pull the paired .mov. It flows through
152 # the loop above like any other version; non-Live-Photos don't have
153 # those versions and are skipped (quietly -- see collect_download_task).
154 return tasks
155 except Exception as e:
156 try:
157 photo_id = photo.id
158 except AttributeError:
159 photo_id = "<unknown>"
160 LOGGER.warning(
161 f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}",
162 )
163 return []
166def _collect_and_execute_album_in_chunks(
167 album,
168 destination_path: str,
169 file_sizes: list[str],
170 extensions: list[str] | None,
171 files: set[str] | None,
172 folder_format: str | None,
173 hardlink_registry: HardlinkRegistry | None,
174 config,
175 chunk_size: int = DEFAULT_ENUMERATION_CHUNK_SIZE,
176) -> tuple[int, int]:
177 """Stream album → fixed-size chunks → download → release.
179 Buffers up to ``chunk_size`` download tasks, then drains them via
180 ``execute_parallel_downloads`` and clears the buffer before
181 collecting the next chunk. Memory is bounded by chunk_size, not by
182 len(album). Semantically equivalent to building the full task list
183 and downloading once — same total counts, same per-photo
184 side-effects — but resident-set stays flat instead of growing
185 monotonically through enumeration.
187 Args:
188 album: Album object from iCloudPy
189 destination_path: Path where photos should be saved
190 file_sizes: List of file size variants to download
191 extensions: List of allowed file extensions
192 files: Set to track downloaded files
193 folder_format: strftime format string for folder organization
194 hardlink_registry: Registry for tracking downloaded files
195 config: Configuration dictionary (passed through to
196 ``execute_parallel_downloads`` for per-album thread count)
197 chunk_size: Tasks to buffer before draining. Smaller = lower
198 peak memory but more per-chunk HTTP setup overhead.
200 Returns:
201 Tuple of (total_successful, total_failed) summed across chunks.
202 """
203 if chunk_size <= 0:
204 # Degenerate config; fall back to default rather than refusing
205 # to sync. Logging is at INFO so operators see the fallback.
206 LOGGER.info(
207 f"Invalid photos.enumeration_chunk_size={chunk_size!r}; "
208 f"using default {DEFAULT_ENUMERATION_CHUNK_SIZE}.",
209 )
210 chunk_size = DEFAULT_ENUMERATION_CHUNK_SIZE
212 buffer: list[DownloadTaskInfo] = []
213 total_successful = 0
214 total_failed = 0
216 def _drain():
217 nonlocal total_successful, total_failed, buffer
218 if not buffer:
219 return
220 succ, fail = execute_parallel_downloads(buffer, config)
221 total_successful += succ
222 total_failed += fail
223 # Rebind to a fresh list rather than clearing in place: the old
224 # list was just handed to execute_parallel_downloads, so a fresh
225 # object gives each chunk an independent buffer (no aliasing of
226 # an already-passed list) and lets the old chunk's task objects
227 # (and the photo refs they hold) be collected immediately.
228 buffer = []
230 for photo in album:
231 buffer.extend(
232 _collect_photo_download_tasks(
233 photo,
234 destination_path,
235 file_sizes,
236 extensions,
237 files,
238 folder_format,
239 hardlink_registry,
240 ),
241 )
242 if len(buffer) >= chunk_size:
243 _drain()
245 _drain() # final partial chunk
247 return total_successful, total_failed
250def _sync_subalbums(
251 album,
252 destination_path: str,
253 file_sizes: list[str],
254 extensions: list[str] | None,
255 files: set[str] | None,
256 folder_format: str | None,
257 hardlink_registry: HardlinkRegistry | None,
258 config,
259) -> tuple[int, int]:
260 """Recursively sync all subalbums.
262 Args:
263 album: Album object from iCloudPy
264 destination_path: Base path where subalbums should be created
265 file_sizes: List of file size variants to download
266 extensions: List of allowed file extensions
267 files: Set to track downloaded files
268 folder_format: strftime format string for folder organization
269 hardlink_registry: Registry for tracking downloaded files
270 config: Configuration dictionary
272 Returns:
273 Tuple of (total_successful, total_failed) aggregated across all subalbums
274 """
275 total_successful, total_failed = 0, 0
276 for subalbum in album.subalbums:
277 result = sync_album_photos(
278 album.subalbums[subalbum],
279 os.path.join(destination_path, subalbum),
280 file_sizes,
281 extensions,
282 files,
283 folder_format,
284 hardlink_registry,
285 config,
286 )
287 if result is not None:
288 sub_successful, sub_failed = result
289 total_successful += sub_successful
290 total_failed += sub_failed
291 return total_successful, total_failed