Coverage for src/photo_file_utils.py: 100%
111 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 17:25 +0000
1"""Photo file operations module.
3This module contains utilities for photo file operations including
4downloading, hardlink creation, and file existence checking.
5"""
7___author___ = "Mandar Patil <mandarons@pm.me>"
9import json
10import os
11import shutil
12import threading
13from datetime import timezone
14from urllib.parse import urlencode
16from src import get_logger
18LOGGER = get_logger()
20# Module-level lock to protect thread-safe mutation of photo._versions during retries
21_versions_refresh_lock = threading.Lock()
23# Consecutive download-URL refresh failures, and how often to escalate them to WARNING.
24# Downloads run in parallel, so the counter is guarded by its own lock.
25_refresh_failure_lock = threading.Lock()
26_consecutive_refresh_failures = 0
27_REFRESH_FAILURE_WARN_INTERVAL = 3
29# CloudKit fields to request when re-fetching a photo record for fresh download URLs.
30# Mirrors the desiredKeys list used by icloudpy's PhotoAlbum._list_query_gen().
31_DESIRED_KEYS = [
32 "resJPEGFullWidth",
33 "resJPEGFullHeight",
34 "resJPEGFullFileType",
35 "resJPEGFullFingerprint",
36 "resJPEGFullRes",
37 "resJPEGLargeWidth",
38 "resJPEGLargeHeight",
39 "resJPEGLargeFileType",
40 "resJPEGLargeFingerprint",
41 "resJPEGLargeRes",
42 "resJPEGMedWidth",
43 "resJPEGMedHeight",
44 "resJPEGMedFileType",
45 "resJPEGMedFingerprint",
46 "resJPEGMedRes",
47 "resJPEGThumbWidth",
48 "resJPEGThumbHeight",
49 "resJPEGThumbFileType",
50 "resJPEGThumbFingerprint",
51 "resJPEGThumbRes",
52 "resVidFullWidth",
53 "resVidFullHeight",
54 "resVidFullFileType",
55 "resVidFullFingerprint",
56 "resVidFullRes",
57 "resVidMedWidth",
58 "resVidMedHeight",
59 "resVidMedFileType",
60 "resVidMedFingerprint",
61 "resVidMedRes",
62 "resVidSmallWidth",
63 "resVidSmallHeight",
64 "resVidSmallFileType",
65 "resVidSmallFingerprint",
66 "resVidSmallRes",
67 "resSidecarWidth",
68 "resSidecarHeight",
69 "resSidecarFileType",
70 "resSidecarFingerprint",
71 "resSidecarRes",
72 "itemType",
73 "dataClassType",
74 "filenameEnc",
75 "originalOrientation",
76 "resOriginalWidth",
77 "resOriginalHeight",
78 "resOriginalFileType",
79 "resOriginalFingerprint",
80 "resOriginalRes",
81 "resOriginalAltWidth",
82 "resOriginalAltHeight",
83 "resOriginalAltFileType",
84 "resOriginalAltFingerprint",
85 "resOriginalAltRes",
86 "resOriginalVidComplWidth",
87 "resOriginalVidComplHeight",
88 "resOriginalVidComplFileType",
89 "resOriginalVidComplFingerprint",
90 "resOriginalVidComplRes",
91 "isDeleted",
92 "isExpunged",
93 "dateExpunged",
94 "remappedRef",
95 "recordName",
96 "recordType",
97 "recordChangeTag",
98 "masterRef",
99 "adjustmentRenderType",
100 "assetDate",
101 "addedDate",
102 "isFavorite",
103 "isHidden",
104 "orientation",
105 "duration",
106 "assetSubtype",
107 "assetSubtypeV2",
108 "assetHDRType",
109 "burstFlags",
110 "burstFlagsExt",
111 "burstId",
112 "captionEnc",
113 "extendedDescEnc",
114 "locationEnc",
115 "locationV2Enc",
116 "locationLatitude",
117 "locationLongitude",
118 "adjustmentType",
119 "timeZoneOffset",
120 "vidComplDurValue",
121 "vidComplDurScale",
122 "vidComplDispValue",
123 "vidComplDispScale",
124 "vidComplVisibilityState",
125 "customRenderedValue",
126 "containerId",
127 "itemId",
128 "position",
129 "isKeyAsset",
130 "importedByBundleIdentifierEnc",
131 "importedByDisplayNameEnc",
132 "importedBy",
133]
136def _note_refresh_success() -> None:
137 """Reset the consecutive refresh-failure streak after a successful refresh."""
138 global _consecutive_refresh_failures # noqa: PLW0603
139 with _refresh_failure_lock:
140 _consecutive_refresh_failures = 0
143def _note_refresh_failure(record_name: str, reason: str) -> None:
144 """Record a failed URL refresh, escalating to WARNING once failures repeat.
146 URL refresh is the last line of defence before a download is abandoned, so a
147 systematically broken refresh path (rather than an occasional miss) should be
148 visible without turning on debug logging. Individual failures stay at DEBUG;
149 every ``_REFRESH_FAILURE_WARN_INTERVAL`` consecutive failures emits a WARNING.
151 Args:
152 record_name: CloudKit recordName of the photo being refreshed
153 reason: Human-readable description of why the refresh failed
154 """
155 global _consecutive_refresh_failures # noqa: PLW0603
156 with _refresh_failure_lock:
157 _consecutive_refresh_failures += 1
158 failures = _consecutive_refresh_failures
160 if failures % _REFRESH_FAILURE_WARN_INTERVAL == 0:
161 LOGGER.warning(
162 f"Download URL refresh has failed {failures} times in a row - expired-URL (HTTP 410) "
163 f"recovery is not working, so affected photos will be reported as failed downloads. "
164 f"Most recent failure: {record_name} - {reason}",
165 )
166 else:
167 LOGGER.debug(f"Failed to refresh download URL for {record_name}: {reason}")
170def _refresh_photo_download_url(photo) -> bool:
171 """Re-fetch the photo's master record from iCloud to obtain fresh download URLs.
173 iCloud download URLs are signed CDN tokens that expire after ~30–40 minutes.
174 When a URL expires (HTTP 410 Gone), clearing ``photo._versions`` alone is
175 insufficient because icloudpy re-parses the same stale ``_master_record``
176 which still contains the expired URL. This function makes a new
177 ``records/lookup`` API call to get an updated master record with fresh URLs,
178 then updates ``photo._master_record`` in place and clears ``_versions`` so
179 the next ``download()`` call uses the fresh URL.
181 ``records/lookup`` is used rather than ``records/query`` because ``CPLMaster``
182 is not a query-indexable CloudKit type: querying it fails every time with
183 ``Type is not marked indexable: CPLMaster (BAD_REQUEST)``. Lookup fetches
184 records by name and returns the same ``{"records": [...]}`` shape.
186 Args:
187 photo: PhotoAsset object from icloudpy
189 Returns:
190 True if the master record was successfully refreshed, False otherwise.
191 """
192 try:
193 record_name = photo._master_record["recordName"] # noqa: SLF001
194 except (AttributeError, KeyError, TypeError):
195 _note_refresh_failure("<unknown>", "photo missing _master_record or recordName")
196 return False
198 service = getattr(photo, "_service", None)
199 if service is None:
200 _note_refresh_failure(record_name, "photo missing _service")
201 return False
203 endpoint = getattr(service, "_service_endpoint", None)
204 session = getattr(service, "session", None)
205 params = getattr(service, "params", None)
206 zone_id = getattr(service, "zone_id", None)
208 if not all([endpoint, session, params, zone_id]):
209 _note_refresh_failure(record_name, "photo._service missing required attributes")
210 return False
212 try:
213 url = f"{endpoint}/records/lookup?{urlencode(params)}"
214 query = {
215 "records": [{"recordName": record_name}],
216 "desiredKeys": _DESIRED_KEYS,
217 "zoneID": zone_id,
218 }
219 request = session.post(
220 url,
221 data=json.dumps(query),
222 headers={"Content-type": "text/plain"},
223 )
224 response = request.json()
225 records = response.get("records", [])
227 for rec in records:
228 if rec.get("recordName") == record_name:
229 photo._master_record = rec # noqa: SLF001
230 with _versions_refresh_lock:
231 photo._versions = None # noqa: SLF001
232 LOGGER.debug(f"Refreshed download URL for {record_name}")
233 _note_refresh_success()
234 return True
236 _note_refresh_failure(record_name, "record not found in iCloud response")
237 return False
239 except Exception as e: # noqa: BLE001
240 _note_refresh_failure(record_name, str(e))
241 return False
244def check_photo_exists(photo, file_size: str, local_path: str) -> bool:
245 """Check if photo exists locally with correct size.
247 Args:
248 photo: Photo object from iCloudPy
249 file_size: File size variant (original, medium, thumb, etc.)
250 local_path: Local file path to check
252 Returns:
253 True if photo exists locally with correct size, False otherwise
254 """
255 if not (photo and local_path and os.path.isfile(local_path)):
256 return False
258 local_size = os.path.getsize(local_path)
259 remote_size = int(photo.versions[file_size]["size"])
261 if local_size == remote_size:
262 LOGGER.debug(f"No changes detected. Skipping the file {local_path} ...")
263 return True
264 else:
265 LOGGER.debug(f"Change detected: local_file_size is {local_size} and remote_file_size is {remote_size}.")
266 return False
269def create_hardlink(source_path: str, destination_path: str) -> bool:
270 """Create a hard link from source to destination.
272 Args:
273 source_path: Path to existing file to link from
274 destination_path: Path where hardlink should be created
276 Returns:
277 True if hardlink was created successfully, False otherwise
278 """
279 try:
280 # Ensure destination directory exists
281 os.makedirs(os.path.dirname(destination_path), exist_ok=True)
282 # Create hard link
283 os.link(source_path, destination_path)
284 LOGGER.info(f"Created hard link: {destination_path} (linked to existing file: {source_path})")
285 return True
286 except (OSError, FileNotFoundError) as e:
287 LOGGER.warning(f"Failed to create hard link {destination_path}: {e!s}")
288 return False
291def download_photo_from_server(photo, file_size: str, destination_path: str, max_retries: int = 1) -> bool:
292 """Download photo from iCloud server to local path.
294 This function implements automatic retry logic for HTTP 410 (Gone) errors,
295 which occur when iCloud download URLs expire. When a 410 error is detected,
296 the function re-fetches the photo's master record from iCloud to obtain
297 fresh download URLs, then retries the download.
299 Args:
300 photo: Photo object from iCloudPy
301 file_size: File size variant (original, medium, thumb, etc.)
302 destination_path: Local path where photo should be saved
303 max_retries: Maximum number of retries on 410 errors (default: 1)
305 Returns:
306 True if download was successful, False otherwise
307 """
308 if not (photo and file_size and destination_path):
309 return False
311 LOGGER.info(f"Downloading {destination_path} ...")
313 max_retries = max(0, max_retries) # Clamp to minimum 0 for predictable behavior
314 attempt = 0
315 max_attempts = max_retries + 1 # Initial attempt + retries
317 while attempt < max_attempts: # noqa: PERF203
318 try:
319 download = photo.download(file_size)
320 with open(destination_path, "wb") as file_out:
321 shutil.copyfileobj(download.raw, file_out)
323 # Set file modification time to photo's added date.
324 # iCloudPy returns added_date as an aware UTC datetime; replace() is a
325 # safe no-op here because tzinfo is already UTC. If it ever returns a
326 # naive datetime, replace(tzinfo=utc) correctly treats it as UTC.
327 local_modified_time = photo.added_date.replace(tzinfo=timezone.utc).timestamp()
328 os.utime(destination_path, (local_modified_time, local_modified_time))
330 return True
332 except Exception as e: # noqa: PERF203
333 # Enhanced error logging with file path context
334 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException
335 error_msg = str(e)
337 # Check for HTTP 410 Gone error - download URL has expired
338 # The iCloudPy library raises exceptions with "Gone (410)" in the message
339 # when the download URL has expired (typically after 30-40 minutes)
340 if "Gone (410)" in error_msg:
341 attempt += 1
342 if attempt < max_attempts:
343 LOGGER.warning(
344 f"Download URL expired (410) for {destination_path}. "
345 f"Refreshing URL and retrying (attempt {attempt}/{max_attempts})...",
346 )
347 # Re-fetch the master record from iCloud to obtain fresh download URLs.
348 # Simply clearing _versions is insufficient because icloudpy re-parses
349 # the same stale _master_record which still contains the expired URL.
350 _refresh_photo_download_url(photo)
351 continue
352 else:
353 LOGGER.error(
354 f"Failed to download {destination_path} after {max_retries} retries: {error_msg}",
355 )
356 return False
358 # Handle other errors
359 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg:
360 LOGGER.error(f"Photo not found in iCloud Photos - {destination_path}: {error_msg}")
361 else:
362 LOGGER.error(f"Failed to download {destination_path}: {error_msg}")
363 return False
365 # This line should never be reached due to the logic above, but is kept as defensive programming
366 return False # pragma: no cover
369def rename_legacy_file_if_exists(old_path: str, new_path: str) -> None:
370 """Rename legacy file format to new format if it exists.
372 Args:
373 old_path: Path to legacy file format
374 new_path: Path to new file format
375 """
376 if os.path.isfile(old_path):
377 os.rename(old_path, new_path)