Coverage for src/album_sync_orchestrator.py: 100%
83 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-26 21:28 +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
10import time
11from typing import Any
13from src import BUFFERED_DOWNLOAD_MAX_AGE_SEC, DEFAULT_ENUMERATION_CHUNK_SIZE, config_parser, get_logger
14from src.hardlink_registry import HardlinkRegistry
15from src.photo_download_manager import (
16 DownloadTaskInfo,
17 collect_download_task,
18 execute_parallel_downloads,
19 would_fall_back_to_original,
20)
21from src.photo_filter_utils import is_photo_wanted
22from src.photo_path_utils import normalize_file_path
24LOGGER = get_logger()
27def _now() -> float:
28 """Monotonic clock for buffer ageing; a seam for tests."""
29 return time.monotonic()
32# DEFAULT_ENUMERATION_CHUNK_SIZE lives in src/__init__.py (with the other
33# DEFAULT_* config constants) and is re-exported here for backward-compat.
34# Picked to keep peak RSS bounded at ~10–20 MB of DownloadTaskInfo objects
35# per chunk on typical iCloud libraries while still giving
36# execute_parallel_downloads enough work to amortize HTTP connection setup.
37# Users can override via ``photos.enumeration_chunk_size`` in config.yaml.
40def sync_album_photos(
41 album,
42 destination_path: str,
43 file_sizes: list[str],
44 extensions: list[str] | None = None,
45 files: set[str] | None = None,
46 folder_format: str | None = None,
47 hardlink_registry: HardlinkRegistry | None = None,
48 config=None,
49) -> tuple[int, int] | None:
50 """Sync photos from given album.
52 This function orchestrates the synchronization of a single album by:
53 1. Creating the destination directory
54 2. Collecting download tasks for wanted photos
55 3. Executing downloads in parallel
56 4. Recursively syncing subalbums
58 Args:
59 album: Album object from iCloudPy
60 destination_path: Path where photos should be saved
61 file_sizes: List of file size variants to download
62 extensions: List of allowed file extensions (None = all allowed)
63 files: Set to track downloaded files
64 folder_format: strftime format string for folder organization
65 hardlink_registry: Registry for tracking downloaded files for hardlinks
66 config: Configuration dictionary
68 Returns:
69 Tuple of (total_successful, total_failed) download counts, or None on invalid input
70 """
71 if album is None or destination_path is None or file_sizes is None:
72 return None
74 # Create destination directory with normalized path
75 normalized_destination = normalize_file_path(destination_path)
76 os.makedirs(normalized_destination, exist_ok=True)
77 LOGGER.info(f"Syncing {album.title}")
79 # Stream the album in fixed-size chunks: collect → download →
80 # release → next chunk. Memory is bounded by chunk_size × per-task
81 # size (~10 MB at chunk=1000) instead of len(album) × per-task size
82 # (which OOM-kills containers on ~100K+ libraries: empirically a
83 # 111K-photo library peaks at ~4 GB RSS without chunking, kernel-
84 # confirmed via cgroup OOM at the 4 GB cap).
85 chunk_size = config_parser.get_photos_enumeration_chunk_size(config=config)
86 total_successful, total_failed = _collect_and_execute_album_in_chunks(
87 album,
88 normalized_destination,
89 file_sizes,
90 extensions,
91 files,
92 folder_format,
93 hardlink_registry,
94 config,
95 chunk_size=chunk_size,
96 )
98 # Recursively sync subalbums and aggregate counts
99 sub_successful, sub_failed = _sync_subalbums(
100 album,
101 normalized_destination,
102 file_sizes,
103 extensions,
104 files,
105 folder_format,
106 hardlink_registry,
107 config,
108 )
109 total_successful += sub_successful
110 total_failed += sub_failed
112 return total_successful, total_failed
115def _collect_photo_download_tasks(
116 photo: Any,
117 destination_path: str,
118 file_sizes: list[str],
119 extensions: list[str] | None,
120 files: set[str] | None,
121 folder_format: str | None,
122 hardlink_registry: HardlinkRegistry | None,
123) -> list[DownloadTaskInfo]:
124 """Collect download tasks for a single photo, handling errors gracefully.
126 Wraps per-photo processing so that exceptions (e.g. binascii.Error from
127 iCloudPy's base64-encoded filename decoding) are caught at the photo level
128 rather than inside the album iteration loop (avoids PERF203).
130 Args:
131 photo: Photo object from iCloudPy
132 destination_path: Path where photos should be saved
133 file_sizes: List of file size variants to download
134 extensions: List of allowed file extensions
135 files: Set to track downloaded files
136 folder_format: strftime format string for folder organization
137 hardlink_registry: Registry for tracking downloaded files
139 Returns:
140 List of download tasks for this photo (empty on error or if unwanted)
141 """
142 try:
143 if not is_photo_wanted(photo, extensions):
144 LOGGER.debug(f"Skipping the unwanted photo {photo.filename}.")
145 return []
146 tasks: list[DownloadTaskInfo] = []
147 # With ``fallback_to_original`` on, two requested sizes a photo lacks
148 # (e.g. ``[adjusted, alternative]`` for an unedited photo) both
149 # resolve to ``original``. Claim that substitution once per photo so
150 # the parallel executor never queues two downloads onto one path.
151 fallback_claimed = False
152 for file_size in file_sizes:
153 if would_fall_back_to_original(photo, file_size):
154 if fallback_claimed:
155 LOGGER.debug(
156 f"Already downloading the original for {photo.filename}; "
157 f"skipping the {file_size} fallback.",
158 )
159 continue
160 fallback_claimed = True
161 download_info = collect_download_task(
162 photo,
163 file_size,
164 destination_path,
165 files,
166 folder_format,
167 hardlink_registry,
168 )
169 if download_info:
170 tasks.append(download_info)
171 # Live Photos: add "live_video_original" (or _medium/_thumb) to
172 # photos.filters.file_sizes to pull the paired .mov. It flows through
173 # the loop above like any other version; non-Live-Photos don't have
174 # those versions and are skipped (quietly -- see collect_download_task).
175 return tasks
176 except Exception as e:
177 try:
178 photo_id = photo.id
179 except AttributeError:
180 photo_id = "<unknown>"
181 LOGGER.warning(
182 f"Error processing photo (id: {photo_id}), skipping: {type(e).__name__}: {e!s}",
183 )
184 return []
187def _collect_and_execute_album_in_chunks(
188 album,
189 destination_path: str,
190 file_sizes: list[str],
191 extensions: list[str] | None,
192 files: set[str] | None,
193 folder_format: str | None,
194 hardlink_registry: HardlinkRegistry | None,
195 config,
196 chunk_size: int = DEFAULT_ENUMERATION_CHUNK_SIZE,
197) -> tuple[int, int]:
198 """Stream album → fixed-size chunks → download → release.
200 Buffers up to ``chunk_size`` download tasks, then drains them via
201 ``execute_parallel_downloads`` and clears the buffer before
202 collecting the next chunk. The buffer also drains once its oldest
203 task is ``BUFFERED_DOWNLOAD_MAX_AGE_SEC`` old, so download URLs (which
204 expire ~30-40 minutes after listing) do not wait out a long walk of
205 photos already on disk. That is checked between photos, so it is a
206 bound on the usual wait, not a guarantee; a URL that still expires is
207 recovered by the expired-URL refresh. Memory is bounded by chunk_size, not by
208 len(album). Semantically equivalent to building the full task list
209 and downloading once — same total counts, same per-photo
210 side-effects — but resident-set stays flat instead of growing
211 monotonically through enumeration.
213 Args:
214 album: Album object from iCloudPy
215 destination_path: Path where photos should be saved
216 file_sizes: List of file size variants to download
217 extensions: List of allowed file extensions
218 files: Set to track downloaded files
219 folder_format: strftime format string for folder organization
220 hardlink_registry: Registry for tracking downloaded files
221 config: Configuration dictionary (passed through to
222 ``execute_parallel_downloads`` for per-album thread count)
223 chunk_size: Tasks to buffer before draining. Smaller = lower
224 peak memory but more per-chunk HTTP setup overhead.
226 Returns:
227 Tuple of (total_successful, total_failed) summed across chunks.
228 """
229 if chunk_size <= 0:
230 # Degenerate config; fall back to default rather than refusing
231 # to sync. Logging is at INFO so operators see the fallback.
232 LOGGER.info(
233 f"Invalid photos.enumeration_chunk_size={chunk_size!r}; "
234 f"using default {DEFAULT_ENUMERATION_CHUNK_SIZE}.",
235 )
236 chunk_size = DEFAULT_ENUMERATION_CHUNK_SIZE
238 buffer: list[DownloadTaskInfo] = []
239 # When the buffer must drain regardless of size; None while it is empty.
240 drain_by: float | None = None
241 total_successful = 0
242 total_failed = 0
244 def _drain():
245 nonlocal total_successful, total_failed, buffer, drain_by
246 if not buffer:
247 return
248 succ, fail = execute_parallel_downloads(buffer, config)
249 total_successful += succ
250 total_failed += fail
251 # Rebind to a fresh list rather than clearing in place: the old
252 # list was just handed to execute_parallel_downloads, so a fresh
253 # object gives each chunk an independent buffer (no aliasing of
254 # an already-passed list) and lets the old chunk's task objects
255 # (and the photo refs they hold) be collected immediately.
256 buffer = []
257 drain_by = None
259 for photo in album:
260 tasks = _collect_photo_download_tasks(
261 photo,
262 destination_path,
263 file_sizes,
264 extensions,
265 files,
266 folder_format,
267 hardlink_registry,
268 )
269 if tasks and drain_by is None:
270 drain_by = _now() + BUFFERED_DOWNLOAD_MAX_AGE_SEC
271 buffer.extend(tasks)
272 aged_out = drain_by is not None and _now() >= drain_by
273 if len(buffer) >= chunk_size or aged_out:
274 _drain()
276 _drain() # final partial chunk
278 return total_successful, total_failed
281def _sync_subalbums(
282 album,
283 destination_path: str,
284 file_sizes: list[str],
285 extensions: list[str] | None,
286 files: set[str] | None,
287 folder_format: str | None,
288 hardlink_registry: HardlinkRegistry | None,
289 config,
290) -> tuple[int, int]:
291 """Recursively sync all subalbums.
293 Args:
294 album: Album object from iCloudPy
295 destination_path: Base path where subalbums should be created
296 file_sizes: List of file size variants to download
297 extensions: List of allowed file extensions
298 files: Set to track downloaded files
299 folder_format: strftime format string for folder organization
300 hardlink_registry: Registry for tracking downloaded files
301 config: Configuration dictionary
303 Returns:
304 Tuple of (total_successful, total_failed) aggregated across all subalbums
305 """
306 total_successful, total_failed = 0, 0
307 for subalbum in album.subalbums:
308 result = sync_album_photos(
309 album.subalbums[subalbum],
310 os.path.join(destination_path, subalbum),
311 file_sizes,
312 extensions,
313 files,
314 folder_format,
315 hardlink_registry,
316 config,
317 )
318 if result is not None:
319 sub_successful, sub_failed = result
320 total_successful += sub_successful
321 total_failed += sub_failed
322 return total_successful, total_failed