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