Coverage for src/photo_file_utils.py: 100%
98 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +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# CloudKit fields to request when re-fetching a photo record for fresh download URLs.
24# Mirrors the desiredKeys list used by icloudpy's PhotoAlbum._list_query_gen().
25_DESIRED_KEYS = [
26 "resJPEGFullWidth",
27 "resJPEGFullHeight",
28 "resJPEGFullFileType",
29 "resJPEGFullFingerprint",
30 "resJPEGFullRes",
31 "resJPEGLargeWidth",
32 "resJPEGLargeHeight",
33 "resJPEGLargeFileType",
34 "resJPEGLargeFingerprint",
35 "resJPEGLargeRes",
36 "resJPEGMedWidth",
37 "resJPEGMedHeight",
38 "resJPEGMedFileType",
39 "resJPEGMedFingerprint",
40 "resJPEGMedRes",
41 "resJPEGThumbWidth",
42 "resJPEGThumbHeight",
43 "resJPEGThumbFileType",
44 "resJPEGThumbFingerprint",
45 "resJPEGThumbRes",
46 "resVidFullWidth",
47 "resVidFullHeight",
48 "resVidFullFileType",
49 "resVidFullFingerprint",
50 "resVidFullRes",
51 "resVidMedWidth",
52 "resVidMedHeight",
53 "resVidMedFileType",
54 "resVidMedFingerprint",
55 "resVidMedRes",
56 "resVidSmallWidth",
57 "resVidSmallHeight",
58 "resVidSmallFileType",
59 "resVidSmallFingerprint",
60 "resVidSmallRes",
61 "resSidecarWidth",
62 "resSidecarHeight",
63 "resSidecarFileType",
64 "resSidecarFingerprint",
65 "resSidecarRes",
66 "itemType",
67 "dataClassType",
68 "filenameEnc",
69 "originalOrientation",
70 "resOriginalWidth",
71 "resOriginalHeight",
72 "resOriginalFileType",
73 "resOriginalFingerprint",
74 "resOriginalRes",
75 "resOriginalAltWidth",
76 "resOriginalAltHeight",
77 "resOriginalAltFileType",
78 "resOriginalAltFingerprint",
79 "resOriginalAltRes",
80 "resOriginalVidComplWidth",
81 "resOriginalVidComplHeight",
82 "resOriginalVidComplFileType",
83 "resOriginalVidComplFingerprint",
84 "resOriginalVidComplRes",
85 "isDeleted",
86 "isExpunged",
87 "dateExpunged",
88 "remappedRef",
89 "recordName",
90 "recordType",
91 "recordChangeTag",
92 "masterRef",
93 "adjustmentRenderType",
94 "assetDate",
95 "addedDate",
96 "isFavorite",
97 "isHidden",
98 "orientation",
99 "duration",
100 "assetSubtype",
101 "assetSubtypeV2",
102 "assetHDRType",
103 "burstFlags",
104 "burstFlagsExt",
105 "burstId",
106 "captionEnc",
107 "extendedDescEnc",
108 "locationEnc",
109 "locationV2Enc",
110 "locationLatitude",
111 "locationLongitude",
112 "adjustmentType",
113 "timeZoneOffset",
114 "vidComplDurValue",
115 "vidComplDurScale",
116 "vidComplDispValue",
117 "vidComplDispScale",
118 "vidComplVisibilityState",
119 "customRenderedValue",
120 "containerId",
121 "itemId",
122 "position",
123 "isKeyAsset",
124 "importedByBundleIdentifierEnc",
125 "importedByDisplayNameEnc",
126 "importedBy",
127]
130def _refresh_photo_download_url(photo) -> bool:
131 """Re-fetch the photo's master record from iCloud to obtain fresh download URLs.
133 iCloud download URLs are signed CDN tokens that expire after ~30–40 minutes.
134 When a URL expires (HTTP 410 Gone), clearing ``photo._versions`` alone is
135 insufficient because icloudpy re-parses the same stale ``_master_record``
136 which still contains the expired URL. This function makes a new
137 ``records/query`` API call to get an updated master record with fresh URLs,
138 then updates ``photo._master_record`` in place and clears ``_versions`` so
139 the next ``download()`` call uses the fresh URL.
141 Args:
142 photo: PhotoAsset object from icloudpy
144 Returns:
145 True if the master record was successfully refreshed, False otherwise.
146 """
147 try:
148 record_name = photo._master_record["recordName"] # noqa: SLF001
149 record_type = photo._master_record.get("recordType", "CPLMaster") # noqa: SLF001
150 except (AttributeError, KeyError, TypeError):
151 LOGGER.debug("Cannot refresh download URL: photo missing _master_record or recordName")
152 return False
154 service = getattr(photo, "_service", None)
155 if service is None:
156 LOGGER.debug("Cannot refresh download URL: photo missing _service")
157 return False
159 endpoint = getattr(service, "_service_endpoint", None)
160 session = getattr(service, "session", None)
161 params = getattr(service, "params", None)
162 zone_id = getattr(service, "zone_id", None)
164 if not all([endpoint, session, params, zone_id]):
165 LOGGER.debug("Cannot refresh download URL: photo._service missing required attributes")
166 return False
168 try:
169 url = f"{endpoint}/records/query?{urlencode(params)}"
170 query = {
171 "query": {
172 "recordType": record_type,
173 "filterBy": [
174 {
175 "fieldName": "recordName",
176 "comparator": "IN",
177 "fieldValue": {
178 "type": "STRING_LIST",
179 "value": [record_name],
180 },
181 },
182 ],
183 },
184 "resultsLimit": 1,
185 "desiredKeys": _DESIRED_KEYS,
186 "zoneID": zone_id,
187 }
188 request = session.post(
189 url,
190 data=json.dumps(query),
191 headers={"Content-type": "text/plain"},
192 )
193 response = request.json()
194 records = response.get("records", [])
196 for rec in records:
197 if rec.get("recordName") == record_name:
198 photo._master_record = rec # noqa: SLF001
199 with _versions_refresh_lock:
200 photo._versions = None # noqa: SLF001
201 LOGGER.debug(f"Refreshed download URL for {record_name}")
202 return True
204 LOGGER.debug(f"Record {record_name} not found in iCloud response during URL refresh")
205 return False
207 except Exception as e: # noqa: BLE001
208 LOGGER.debug(f"Failed to refresh download URL for {record_name}: {e!s}")
209 return False
212def check_photo_exists(photo, file_size: str, local_path: str) -> bool:
213 """Check if photo exists locally with correct size.
215 Args:
216 photo: Photo object from iCloudPy
217 file_size: File size variant (original, medium, thumb, etc.)
218 local_path: Local file path to check
220 Returns:
221 True if photo exists locally with correct size, False otherwise
222 """
223 if not (photo and local_path and os.path.isfile(local_path)):
224 return False
226 local_size = os.path.getsize(local_path)
227 remote_size = int(photo.versions[file_size]["size"])
229 if local_size == remote_size:
230 LOGGER.debug(f"No changes detected. Skipping the file {local_path} ...")
231 return True
232 else:
233 LOGGER.debug(f"Change detected: local_file_size is {local_size} and remote_file_size is {remote_size}.")
234 return False
237def create_hardlink(source_path: str, destination_path: str) -> bool:
238 """Create a hard link from source to destination.
240 Args:
241 source_path: Path to existing file to link from
242 destination_path: Path where hardlink should be created
244 Returns:
245 True if hardlink was created successfully, False otherwise
246 """
247 try:
248 # Ensure destination directory exists
249 os.makedirs(os.path.dirname(destination_path), exist_ok=True)
250 # Create hard link
251 os.link(source_path, destination_path)
252 LOGGER.info(f"Created hard link: {destination_path} (linked to existing file: {source_path})")
253 return True
254 except (OSError, FileNotFoundError) as e:
255 LOGGER.warning(f"Failed to create hard link {destination_path}: {e!s}")
256 return False
259def download_photo_from_server(photo, file_size: str, destination_path: str, max_retries: int = 1) -> bool:
260 """Download photo from iCloud server to local path.
262 This function implements automatic retry logic for HTTP 410 (Gone) errors,
263 which occur when iCloud download URLs expire. When a 410 error is detected,
264 the function re-fetches the photo's master record from iCloud to obtain
265 fresh download URLs, then retries the download.
267 Args:
268 photo: Photo object from iCloudPy
269 file_size: File size variant (original, medium, thumb, etc.)
270 destination_path: Local path where photo should be saved
271 max_retries: Maximum number of retries on 410 errors (default: 1)
273 Returns:
274 True if download was successful, False otherwise
275 """
276 if not (photo and file_size and destination_path):
277 return False
279 LOGGER.info(f"Downloading {destination_path} ...")
281 max_retries = max(0, max_retries) # Clamp to minimum 0 for predictable behavior
282 attempt = 0
283 max_attempts = max_retries + 1 # Initial attempt + retries
285 while attempt < max_attempts: # noqa: PERF203
286 try:
287 download = photo.download(file_size)
288 with open(destination_path, "wb") as file_out:
289 shutil.copyfileobj(download.raw, file_out)
291 # Set file modification time to photo's added date.
292 # iCloudPy returns added_date as an aware UTC datetime; replace() is a
293 # safe no-op here because tzinfo is already UTC. If it ever returns a
294 # naive datetime, replace(tzinfo=utc) correctly treats it as UTC.
295 local_modified_time = photo.added_date.replace(tzinfo=timezone.utc).timestamp()
296 os.utime(destination_path, (local_modified_time, local_modified_time))
298 return True
300 except Exception as e: # noqa: PERF203
301 # Enhanced error logging with file path context
302 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException
303 error_msg = str(e)
305 # Check for HTTP 410 Gone error - download URL has expired
306 # The iCloudPy library raises exceptions with "Gone (410)" in the message
307 # when the download URL has expired (typically after 30-40 minutes)
308 if "Gone (410)" in error_msg:
309 attempt += 1
310 if attempt < max_attempts:
311 LOGGER.warning(
312 f"Download URL expired (410) for {destination_path}. "
313 f"Refreshing URL and retrying (attempt {attempt}/{max_attempts})...",
314 )
315 # Re-fetch the master record from iCloud to obtain fresh download URLs.
316 # Simply clearing _versions is insufficient because icloudpy re-parses
317 # the same stale _master_record which still contains the expired URL.
318 _refresh_photo_download_url(photo)
319 continue
320 else:
321 LOGGER.error(
322 f"Failed to download {destination_path} after {max_retries} retries: {error_msg}",
323 )
324 return False
326 # Handle other errors
327 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg:
328 LOGGER.error(f"Photo not found in iCloud Photos - {destination_path}: {error_msg}")
329 else:
330 LOGGER.error(f"Failed to download {destination_path}: {error_msg}")
331 return False
333 # This line should never be reached due to the logic above, but is kept as defensive programming
334 return False # pragma: no cover
337def rename_legacy_file_if_exists(old_path: str, new_path: str) -> None:
338 """Rename legacy file format to new format if it exists.
340 Args:
341 old_path: Path to legacy file format
342 new_path: Path to new file format
343 """
344 if os.path.isfile(old_path):
345 os.rename(old_path, new_path)