Coverage for src/migration_check.py: 100%
147 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +0000
1"""Per-file dry-run validation — does mandarons see the right files?
3Walks each iCloud photo library AND iCloud Drive and, for a sample
4(or all) items, computes the on-disk path mandarons WOULD write to
5using the live config (library_destinations, folder_format,
6filename_format for photos; mirror-tree for drive) and checks whether
7the file is already there with the matching size.
9Used by ``--dry-run --check-files`` from ``main.py`` so users
10migrating from a different downloader (e.g. boredazfcuk's
11icloud_photos_downloader) can confirm the size-based existence check
12will actually find their existing files BEFORE mandarons launches a
13real sync. Without this check, a single misconfiguration —
14filename_format wrong, folder_format missing, library_destinations
15mapping a non-existent key, drive destination pointing at the wrong
16mount — silently triggers a full re-download of the user's entire
17library.
19Pure read: no downloads, no keyring writes, no cookie writes.
20"""
22__author__ = "Mandar Patil (mandarons@pm.me)"
24import os
25import unicodedata
26from pathlib import Path
27from typing import Any
28from urllib.parse import unquote
30from src import config_parser, get_logger
31from src.photo_path_utils import (
32 generate_photo_filename_with_metadata,
33 resolve_folder_path,
34)
36# These two symbols ship in companion PRs (``feat/photos-filename-format-simple``
37# and ``feat/photos-library-destinations``). When those land first the
38# real functions are used; when this PR is reviewed/merged in isolation
39# the no-op fallbacks let the suite import + run, so the migration-check
40# at least authenticates and walks the libraries (just without per-
41# library subdirectories or simple-filename naming). The dry-run still
42# reports something useful: "would mandarons-default paths line up with
43# what's on disk?"
44try:
45 from src.photo_path_utils import set_default_filename_format # type: ignore[attr-defined]
46except (
47 ImportError
48): # pragma: no cover — only when feat/photos-filename-format-simple isn't merged
50 def set_default_filename_format(_filename_format: str) -> None:
51 """No-op fallback — feat/photos-filename-format-simple not merged."""
54try:
55 from src.sync_photos import _library_destination # type: ignore[attr-defined]
56except (
57 ImportError
58): # pragma: no cover — only when feat/photos-library-destinations isn't merged
60 def _library_destination(
61 base_destination: str,
62 library: str,
63 library_destinations: dict,
64 ) -> str:
65 """Fallback that always returns the base destination.
67 feat/photos-library-destinations introduces per-library subdir
68 mapping; without it, mandarons writes every library to the
69 single base destination. The migration-check reports against
70 that same path.
71 """
72 return base_destination
75LOGGER = get_logger()
78def _check_one_photo(
79 photo,
80 library_dest: str,
81 folder_format: str | None,
82) -> tuple[str, str, int, int]:
83 """Compute target path + status for a single photo. Returns
84 ``(status, path, expected_size, actual_size)`` where ``status`` is
85 one of ``would_skip`` / ``size_mismatch`` / ``not_found`` /
86 ``error``."""
87 try:
88 file_size = "original"
89 if file_size not in photo.versions:
90 return "error", "", 0, 0
91 folder_path = resolve_folder_path(library_dest, folder_format, photo)
92 filename = generate_photo_filename_with_metadata(photo, file_size)
93 target_path = os.path.join(folder_path, filename)
94 expected = int(photo.versions[file_size]["size"])
95 except Exception as e:
96 LOGGER.debug(
97 f"check_migration: failed to compute path for {getattr(photo, 'filename', '?')}: {e!s}",
98 )
99 return "error", "", 0, 0
101 if not os.path.isfile(target_path):
102 return "not_found", target_path, expected, 0
103 try:
104 actual = os.path.getsize(target_path)
105 except OSError:
106 return "error", target_path, expected, 0
107 if actual == expected:
108 return "would_skip", target_path, expected, actual
109 return "size_mismatch", target_path, expected, actual
112def check_library(
113 library,
114 library_name: str,
115 photos_base: str,
116 mapping: dict,
117 folder_format: str | None,
118 sample: int,
119) -> dict[str, Any]:
120 """Walk a single library and accumulate per-status counters.
122 ``sample=0`` walks every photo (slow on large libraries).
123 ``sample>0`` walks the first N (newest-first per icloudpy's
124 iterator). Pagination cost is proportional to the number of
125 photos walked, so a sample of 200 is usually sub-minute; a sample
126 of 5000 can take several minutes on a 100K-photo library.
128 Note on bias: iCloud's iterator is newest-first, so a small N
129 skews toward recent photos and gives you no signal about whether
130 older files (which a migration tool would have downloaded years
131 ago) will match. Use a sample size proportional to the time
132 range you care about validating.
133 """
134 library_dest = _library_destination(photos_base, library_name, mapping)
135 # Match the real sync's default path layout: ``_sync_all_photos_in_library``
136 # in src/sync_photos.py iterates ``library.all`` (not
137 # ``library.albums["All Photos"]``) and writes under
138 # ``<library_dest>/all/<folder_format>/<filename>``. Earlier versions
139 # of this checker walked the album view and reported existing files
140 # as ``not_found`` because the on-disk path it computed was missing
141 # the trailing ``/all/`` segment.
142 check_dest = os.path.join(library_dest, "all")
143 stats = {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}
144 samples = {"would_skip": [], "size_mismatch": [], "not_found": []}
146 seen = 0
147 checked = 0
148 try:
149 for photo in library.all:
150 if sample > 0 and checked >= sample:
151 break
152 seen += 1
153 checked += 1
154 status, path, expected, actual = _check_one_photo(
155 photo,
156 check_dest,
157 folder_format,
158 )
159 stats[status] = stats.get(status, 0) + 1
160 if status in samples and len(samples[status]) < 3:
161 if status == "size_mismatch":
162 samples[status].append((path, expected, actual))
163 else:
164 samples[status].append((path, expected))
165 except Exception as e:
166 LOGGER.warning(f"check_migration: walk of {library_name} stopped early: {e!s}")
168 return {
169 "library_dest": library_dest,
170 "checked": checked,
171 "seen": seen,
172 "stats": stats,
173 "samples": samples,
174 }
177def check_migration(api, config: dict, sample: int = 0) -> dict[str, Any]:
178 """Walk every photo library and report what a real sync would do.
180 Args:
181 api: Authenticated ICloudPyService instance.
182 config: Live config dict (read_config output).
183 sample: Photos per library to check. ``0`` means all (slow on
184 large libraries — only use after a small-sample run has
185 confirmed the mapping looks right).
187 Returns:
188 Dict keyed by library name → per-library result dict (see
189 ``check_library``).
190 """
191 # Read-only path resolution — never call prepare_*_destination from
192 # the dry-run path. Those helpers ``os.makedirs`` the destination,
193 # which would leave stub directories on disk if the user
194 # misconfigured the mount they're trying to validate (exactly what
195 # --dry-run is supposed to catch BEFORE writing anything).
196 photos_base = os.path.join(
197 config_parser.get_root_destination_path(config=config),
198 config_parser.get_photos_destination_path(config=config),
199 )
200 # Defensive: feat/photos-library-destinations may not be merged yet.
201 # Falls back to an empty mapping (every library writes to the same
202 # photos_base) so this PR is independent of PR 3.
203 if hasattr(config_parser, "get_photos_library_destinations"):
204 mapping = config_parser.get_photos_library_destinations(config=config)
205 else: # pragma: no cover — only when feat/photos-library-destinations isn't merged
206 mapping = {}
207 folder_format = config_parser.get_photos_folder_format(config=config)
209 # mandarons' sync_photos.sync_photos() normally sets this singleton
210 # via set_default_filename_format(). When the user invokes us via
211 # --dry-run we never go through that path, so we have to set it
212 # ourselves — otherwise every call to
213 # generate_photo_filename_with_metadata returns the legacy metadata-
214 # style name regardless of the config setting.
215 filename_format = (config.get("photos", {}) or {}).get("filename_format")
216 if filename_format:
217 set_default_filename_format(filename_format)
219 results: dict[str, Any] = {}
220 for library_name in api.photos.libraries:
221 LOGGER.info(
222 f"check_migration: walking library {library_name} (sample={sample or 'all'}) ...",
223 )
224 library = api.photos.libraries[library_name]
225 results[library_name] = check_library(
226 library=library,
227 library_name=library_name,
228 photos_base=photos_base,
229 mapping=mapping,
230 folder_format=folder_format,
231 sample=sample,
232 )
233 return results
236def _check_one_drive_file(item, local_path: str) -> tuple[str, str, int, int]:
237 """Compute status for a single Drive file item.
239 Returns ``(status, path, expected_size, actual_size)`` where
240 ``status`` is one of ``would_skip`` / ``size_mismatch`` /
241 ``not_found`` / ``error``.
243 Handles BOTH on-disk forms a Drive item can take:
244 - regular file (most items, plus packages that mandarons couldn't
245 unpack like .key / .jmb — bytes saved flat)
246 - directory tree (packages mandarons successfully unpacked, e.g.
247 .band GarageBand projects)
249 Size comparison matches mandarons' real-sync ``file_exists`` /
250 ``package_exists`` semantics: flat-file size for regular files,
251 sum of contained file sizes for directory packages.
252 """
253 try:
254 expected = int(item.size) if getattr(item, "size", None) is not None else 0
255 except Exception as e:
256 LOGGER.debug(
257 f"check_migration: drive item bad size for {getattr(item, 'name', '?')}: {e!s}",
258 )
259 return "error", local_path, 0, 0
261 if os.path.isdir(local_path):
262 try:
263 actual = sum(
264 f.stat().st_size for f in Path(local_path).glob("**/*") if f.is_file()
265 )
266 except OSError:
267 return "error", local_path, expected, 0
268 elif os.path.isfile(local_path):
269 try:
270 actual = os.path.getsize(local_path)
271 except OSError:
272 return "error", local_path, expected, 0
273 else:
274 return "not_found", local_path, expected, 0
276 if actual == expected:
277 return "would_skip", local_path, expected, actual
278 return "size_mismatch", local_path, expected, actual
281def _walk_drive_recursive(
282 folder,
283 destination_path: str,
284 sample: int,
285 state: dict,
286) -> None:
287 """Recursively walk a Drive folder, mutating ``state`` in place.
289 ``state`` shape:
290 {'checked': int, 'stats': {...}, 'samples': {...}}
292 ``sample > 0`` caps the total file count walked (depth-first across
293 the tree). ``sample == 0`` walks everything.
295 Folders are followed unconditionally; only file items count against
296 the sample cap (folders themselves aren't validated against disk).
297 """
298 if sample > 0 and state["checked"] >= sample:
299 return
300 try:
301 items_index = folder.dir()
302 except Exception as e:
303 LOGGER.debug(f"check_migration: drive folder dir() failed: {e!s}")
304 return
305 if not items_index:
306 return
308 for name in items_index:
309 if sample > 0 and state["checked"] >= sample:
310 return
311 try:
312 item = folder[name]
313 except Exception as e:
314 LOGGER.debug(f"check_migration: drive item access failed for {name}: {e!s}")
315 state["stats"]["error"] = state["stats"].get("error", 0) + 1
316 continue
318 item_type = getattr(item, "type", None)
319 if item_type in ("folder", "app_library"):
320 try:
321 decoded = unquote(getattr(item, "name", name))
322 except Exception:
323 decoded = name
324 sub_dest = unicodedata.normalize(
325 "NFC",
326 os.path.join(destination_path, decoded),
327 )
328 _walk_drive_recursive(item, sub_dest, sample, state)
329 elif item_type == "file":
330 try:
331 decoded = unquote(getattr(item, "name", name))
332 except Exception:
333 decoded = name
334 local_path = unicodedata.normalize(
335 "NFC",
336 os.path.join(destination_path, decoded),
337 )
338 status, path, expected, actual = _check_one_drive_file(item, local_path)
339 state["stats"][status] = state["stats"].get(status, 0) + 1
340 state["checked"] += 1
341 if status in state["samples"] and len(state["samples"][status]) < 3:
342 if status == "size_mismatch":
343 state["samples"][status].append((path, expected, actual))
344 else:
345 state["samples"][status].append((path, expected))
348def check_drive(drive, drive_destination: str, sample: int) -> dict[str, Any]:
349 """Walk iCloud Drive and report per-file dry-run status.
351 Args:
352 drive: ``api.drive`` (root drive node from icloudpy).
353 drive_destination: Local path where mandarons would write Drive
354 content (matches what ``sync_drive`` uses).
355 sample: ``0`` walks every file; ``N > 0`` walks up to N files
356 total (depth-first across the folder tree). A small N gives
357 quick sanity feedback; a large N (or 0) gives statistical
358 confidence at the cost of pagination time.
360 Returns:
361 Dict with keys: ``drive_destination``, ``checked``, ``stats``,
362 ``samples``. Same shape as ``check_library`` minus the
363 library-specific fields.
364 """
365 state: dict[str, Any] = {
366 "checked": 0,
367 "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0},
368 "samples": {"would_skip": [], "size_mismatch": [], "not_found": []},
369 }
370 try:
371 _walk_drive_recursive(drive, drive_destination, sample, state)
372 except Exception as e:
373 LOGGER.warning(f"check_migration: drive walk stopped early: {e!s}")
374 return {
375 "drive_destination": drive_destination,
376 "checked": state["checked"],
377 "stats": state["stats"],
378 "samples": state["samples"],
379 }
382def check_drive_migration(api, config: dict, sample: int = 0) -> dict[str, Any] | None:
383 """Wrapper that resolves drive destination from config then walks.
385 Returns None if there's no ``drive:`` section in the config — caller
386 treats absence as "drive sync would be skipped at real-sync time."
387 """
388 if "drive" not in (config or {}):
389 return None
390 try:
391 # Read-only resolution — see check_migration above for rationale.
392 drive_destination = os.path.join(
393 config_parser.get_root_destination_path(config=config),
394 config_parser.get_drive_destination_path(config=config),
395 )
396 except Exception as e:
397 LOGGER.warning(f"check_migration: drive destination resolution failed: {e!s}")
398 return None
399 LOGGER.info(f"check_migration: walking iCloud Drive (sample={sample or 'all'}) ...")
400 return check_drive(
401 drive=api.drive,
402 drive_destination=drive_destination,
403 sample=sample,
404 )