Coverage for icloudpy/services/photos.py: 94%
293 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-15 15:10 +0000
1"""Photo service."""
3import base64
4import json
5import logging
6from datetime import datetime
8# fmt: off
9from urllib.parse import urlencode # pylint: disable=bad-option-value,relative-import
11from pytz import UTC
13# fmt: on
14from icloudpy.exceptions import (
15 ICloudPyAPIResponseException,
16 ICloudPyLibraryUnavailableException,
17 ICloudPyServiceNotActivatedException,
18)
20LOGGER = logging.getLogger(__name__)
22# Apple rejects a zone it cannot index (or has dropped) with a definitive
23# error -- 400 "Index has invalid data", ZONE_NOT_FOUND. Those mean "this
24# library is unusable" and are safe to skip per library. Auth failures and
25# 5xx are account-wide or transient and must surface exactly as they always have.
26_DEFINITIVE_ZONE_ERRORS = frozenset({"BAD_REQUEST", "ZONE_NOT_FOUND"})
28# BAD_REQUEST alone is broad: every zone-scoped 400 carries it, including one
29# caused by a bug in our own query template rather than by a dead zone. Only
30# the known dead-zone reason text is a verdict on the zone; a 400 with any
31# other reason propagates unconverted (the consumer sees exactly the raise it
32# sees today) instead of silently dropping a possibly-healthy library.
33_DEFINITIVE_ZONE_REASONS = ("Index has invalid data",)
36def _zone_name(zone_id):
37 """Best-effort zone name for messages; zone_id is a dict or a bare name."""
38 return zone_id.get("zoneName") if isinstance(zone_id, dict) else zone_id
41def _zone_post(session, url, data, zone_name):
42 """POST a zone-scoped query, labelling definitive zone rejections.
44 Raises ICloudPyLibraryUnavailableException (carrying ``zone_name``) when
45 Apple definitively rejects the zone; every other error propagates untouched.
46 """
47 try:
48 return session.post(url, data=data, headers={"Content-type": "text/plain"})
49 except ICloudPyAPIResponseException as err:
50 if err.code in _DEFINITIVE_ZONE_ERRORS and (
51 err.code == "ZONE_NOT_FOUND"
52 or any(text in (err.reason or "") for text in _DEFINITIVE_ZONE_REASONS)
53 ):
54 raise ICloudPyLibraryUnavailableException(
55 err.reason,
56 err.code,
57 zone_name=zone_name,
58 ) from err
59 raise
62class PhotoLibrary:
63 """Represents a library in the user's photos.
65 This provides access to all the albums as well as the photos.
66 """
68 SMART_FOLDERS = {
69 "All Photos": {
70 "obj_type": "CPLAssetByAddedDate",
71 "list_type": "CPLAssetAndMasterByAddedDate",
72 "direction": "ASCENDING",
73 "query_filter": None,
74 },
75 "Time-lapse": {
76 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Timelapse",
77 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
78 "direction": "ASCENDING",
79 "query_filter": [
80 {
81 "fieldName": "smartAlbum",
82 "comparator": "EQUALS",
83 "fieldValue": {"type": "STRING", "value": "TIMELAPSE"},
84 },
85 ],
86 },
87 "Videos": {
88 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Video",
89 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
90 "direction": "ASCENDING",
91 "query_filter": [
92 {
93 "fieldName": "smartAlbum",
94 "comparator": "EQUALS",
95 "fieldValue": {"type": "STRING", "value": "VIDEO"},
96 },
97 ],
98 },
99 "Slo-mo": {
100 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Slomo",
101 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
102 "direction": "ASCENDING",
103 "query_filter": [
104 {
105 "fieldName": "smartAlbum",
106 "comparator": "EQUALS",
107 "fieldValue": {"type": "STRING", "value": "SLOMO"},
108 },
109 ],
110 },
111 "Bursts": {
112 "obj_type": "CPLAssetBurstStackAssetByAssetDate",
113 "list_type": "CPLBurstStackAssetAndMasterByAssetDate",
114 "direction": "ASCENDING",
115 "query_filter": None,
116 },
117 "Favorites": {
118 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Favorite",
119 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
120 "direction": "ASCENDING",
121 "query_filter": [
122 {
123 "fieldName": "smartAlbum",
124 "comparator": "EQUALS",
125 "fieldValue": {"type": "STRING", "value": "FAVORITE"},
126 },
127 ],
128 },
129 "Panoramas": {
130 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Panorama",
131 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
132 "direction": "ASCENDING",
133 "query_filter": [
134 {
135 "fieldName": "smartAlbum",
136 "comparator": "EQUALS",
137 "fieldValue": {"type": "STRING", "value": "PANORAMA"},
138 },
139 ],
140 },
141 "Screenshots": {
142 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Screenshot",
143 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
144 "direction": "ASCENDING",
145 "query_filter": [
146 {
147 "fieldName": "smartAlbum",
148 "comparator": "EQUALS",
149 "fieldValue": {"type": "STRING", "value": "SCREENSHOT"},
150 },
151 ],
152 },
153 "Live": {
154 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Live",
155 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
156 "direction": "ASCENDING",
157 "query_filter": [
158 {
159 "fieldName": "smartAlbum",
160 "comparator": "EQUALS",
161 "fieldValue": {"type": "STRING", "value": "LIVE"},
162 },
163 ],
164 },
165 "Recently Deleted": {
166 "obj_type": "CPLAssetDeletedByExpungedDate",
167 "list_type": "CPLAssetAndMasterDeletedByExpungedDate",
168 "direction": "ASCENDING",
169 "query_filter": None,
170 },
171 "Hidden": {
172 "obj_type": "CPLAssetHiddenByAssetDate",
173 "list_type": "CPLAssetAndMasterHiddenByAssetDate",
174 "direction": "ASCENDING",
175 "query_filter": None,
176 },
177 }
179 def __init__(self, service, zone_id):
180 self.service = service
181 self.zone_id = zone_id
183 self._albums = None
185 url = f"{self.service._service_endpoint}/records/query?{urlencode(self.service.params)}"
186 json_data = json.dumps(
187 {
188 "query": {"recordType": "CheckIndexingState"},
189 "zoneID": self.zone_id,
190 },
191 )
193 request = _zone_post(
194 self.service.session,
195 url,
196 json_data,
197 _zone_name(self.zone_id),
198 )
199 response = request.json()
200 indexing_state = response["records"][0]["fields"]["state"]["value"]
201 if indexing_state != "FINISHED":
202 raise ICloudPyServiceNotActivatedException(
203 ("iCloud Photo Library not finished indexing. Please try " "again in a few minutes"),
204 None,
205 )
207 @property
208 def albums(self):
209 if not self._albums:
210 # Fetch before caching anything: a failure must leave no partial
211 # state, so the next access retries instead of silently returning
212 # only the smart folders populated before the error.
213 folders = self._fetch_folders()
214 albums = {
215 name: PhotoAlbum(self.service, name, zone_id=self.zone_id, **props)
216 for (name, props) in self.SMART_FOLDERS.items()
217 }
219 for folder in folders:
220 if folder["recordName"] in (
221 "----Root-Folder----",
222 "----Project-Root-Folder----",
223 ) or (folder["fields"].get("isDeleted") and folder["fields"]["isDeleted"]["value"]):
224 continue
226 folder_id = folder["recordName"]
227 folder_obj_type = f"CPLContainerRelationNotDeletedByAssetDate:{folder_id}"
228 folder_name = base64.b64decode(
229 folder["fields"]["albumNameEnc"]["value"],
230 ).decode("utf-8")
231 query_filter = [
232 {
233 "fieldName": "parentId",
234 "comparator": "EQUALS",
235 "fieldValue": {"type": "STRING", "value": folder_id},
236 },
237 ]
239 album = PhotoAlbum(
240 self.service,
241 folder_name,
242 "CPLContainerRelationLiveByAssetDate",
243 folder_obj_type,
244 "ASCENDING",
245 query_filter,
246 folder_id=folder_id,
247 zone_id=self.zone_id,
248 )
249 albums[folder_name] = album
251 self._albums = albums
253 return self._albums
255 def _fetch_folders(self):
256 url = f"{self.service._service_endpoint}/records/query?{urlencode(self.service.params)}"
257 json_data = json.dumps(
258 {
259 "query": {"recordType": "CPLAlbumByPositionLive"},
260 "zoneID": self.zone_id,
261 },
262 )
264 request = _zone_post(
265 self.service.session,
266 url,
267 json_data,
268 _zone_name(self.zone_id),
269 )
270 response = request.json()
272 return response["records"]
274 def zone_readable(self):
275 """Probe whether the zone answers album queries.
277 Returns False only for a definitive rejection (the zone is unusable)
278 and True otherwise; transient errors propagate to the caller.
279 """
280 try:
281 self._fetch_folders()
282 except ICloudPyLibraryUnavailableException as err:
283 LOGGER.warning("photo library is unreadable and will be skipped: %s", err)
284 return False
285 return True
287 @property
288 def all(self):
289 return self.albums["All Photos"]
292class PhotosService(PhotoLibrary):
293 """The 'Photos' iCloud service.
295 This also acts as a way to access the user's primary library.
296 """
298 def __init__(self, service_root, session, params):
299 self.session = session
300 self.params = dict(params)
301 self._service_root = service_root
302 self._service_endpoint = f"{self._service_root}/database/1/com.apple.photos.cloud/production/private"
304 self._libraries = None
306 self.params.update({"remapEnums": True, "getCurrentSyncToken": True})
308 self._photo_assets = {}
310 super().__init__(service=self, zone_id={"zoneName": "PrimarySync"})
312 @property
313 def libraries(self):
314 if not self._libraries:
315 # A failed zones/list must raise the real error, not be logged and
316 # then crash iterating a variable that was never bound: no
317 # libraries and "some libraries" are different answers, and only
318 # the caller can decide what to do about the first.
319 url = f"{self._service_endpoint}/zones/list"
320 request = self.session.post(
321 url,
322 data="{}",
323 headers={"Content-type": "text/plain"},
324 )
325 zones = request.json()["zones"]
327 libraries = {}
328 for zone in zones:
329 if zone.get("deleted"):
330 continue
331 zone_name = zone["zoneID"]["zoneName"]
332 if zone_name.startswith("CMM-"):
333 # Cloud Moments containers, not photo libraries: they answer
334 # album queries with 400 "Index has invalid data" (the same
335 # failure class rclone filters as fix-icloud-photos-cmm-zones).
336 LOGGER.debug("ignoring non-photo zone %s", zone_name)
337 continue
338 try:
339 library = PhotoLibrary(self, zone_id=zone["zoneID"])
340 except ICloudPyLibraryUnavailableException as err:
341 LOGGER.warning("skipping unreadable photo library: %s", err)
342 continue
343 try:
344 usable = library.zone_readable()
345 except Exception as err: # transient probe failure: keep the zone
346 LOGGER.debug(
347 "probe of %s failed transiently (%s); keeping it",
348 zone_name,
349 err,
350 )
351 usable = True
352 if not usable:
353 continue
354 libraries[zone_name] = library
356 self._libraries = libraries
358 return self._libraries
361class PhotoAlbum:
362 """A photo album."""
364 def __init__(
365 self,
366 service,
367 name,
368 list_type,
369 obj_type,
370 direction,
371 query_filter=None,
372 page_size=100,
373 folder_id=None,
374 zone_id=None,
375 ):
376 self.name = name
377 self.service = service
378 self.list_type = list_type
379 self.obj_type = obj_type
380 self.direction = direction
381 self.query_filter = query_filter
382 self.page_size = page_size
383 self.folder_id = folder_id
385 if zone_id:
386 self._zone_id = zone_id
387 else:
388 self._zone_id = "PrimarySync"
390 self._len = None
392 self._subalbums = {}
394 @property
395 def title(self):
396 """Gets the album name."""
397 return self.name
399 def __iter__(self):
400 return self.photos
402 def iter_chunks(self, chunk_size=1000):
403 """Yield lists of PhotoAsset objects in fixed-size batches.
405 Equivalent to building a list from ``__iter__`` and slicing,
406 but the per-chunk list is yielded eagerly so callers can
407 process + release each chunk before the next is materialised.
408 Lets bulk-download callers bound peak memory by the chunk size
409 rather than ``len(self)`` -- relevant for 100K+ libraries where
410 ``list(album)`` would otherwise hold every PhotoAsset (and any
411 per-asset state callers attach) in memory at once.
413 The underlying ``photos`` property already paginates HTTP
414 responses (``page_size * 2`` records per request); this method
415 is a pure-Python wrapper that batches the yielded
416 ``PhotoAsset`` instances without changing the HTTP fetch
417 pattern.
419 Args:
420 chunk_size: Max number of PhotoAsset objects per yielded
421 list. Values <= 0 are coerced to the default (1000).
423 Yields:
424 List of up to ``chunk_size`` ``PhotoAsset`` instances. The
425 final chunk may be smaller. No empty list is yielded for
426 an empty album.
427 """
428 if chunk_size <= 0:
429 chunk_size = 1000
430 buffer = []
431 for photo in self:
432 buffer.append(photo)
433 if len(buffer) >= chunk_size:
434 yield buffer
435 buffer = []
436 if buffer:
437 yield buffer
439 def __len__(self):
440 if self._len is None:
441 url = f"{self.service._service_endpoint}/internal/records/query/batch?{urlencode(self.service.params)}"
442 data = json.dumps(
443 {
444 "batch": [
445 {
446 "resultsLimit": 1,
447 "query": {
448 "filterBy": {
449 "fieldName": "indexCountID",
450 "fieldValue": {
451 "type": "STRING_LIST",
452 "value": [self.obj_type],
453 },
454 "comparator": "IN",
455 },
456 "recordType": "HyperionIndexCountLookup",
457 },
458 "zoneWide": True,
459 "zoneID": {"zoneName": self._zone_id["zoneName"]},
460 },
461 ],
462 },
463 )
464 request = _zone_post(
465 self.service.session,
466 url,
467 data,
468 _zone_name(self._zone_id),
469 )
470 response = request.json()
472 self._len = response["batch"][0]["records"][0]["fields"]["itemCount"]["value"]
474 return self._len
476 def _fetch_subalbums(self):
477 url = (f"{self.service._service_endpoint}/records/query?") + urlencode(
478 self.service.params,
479 )
480 # pylint: disable=consider-using-f-string
481 query = """{{
482 "query": {{
483 "recordType":"CPLAlbumByPositionLive",
484 "filterBy": [
485 {{
486 "fieldName": "parentId",
487 "comparator": "EQUALS",
488 "fieldValue": {{
489 "value": "{}",
490 "type": "STRING"
491 }}
492 }}
493 ]
494 }},
495 "zoneID": {{
496 "zoneName":"{}"
497 }}
498 }}""".format(
499 self.folder_id,
500 self._zone_id["zoneName"],
501 )
502 json_data = query
503 request = _zone_post(
504 self.service.session,
505 url,
506 json_data,
507 _zone_name(self._zone_id),
508 )
509 response = request.json()
511 return response["records"]
513 @property
514 def subalbums(self):
515 """Returns the subalbums"""
516 if not self._subalbums and self.folder_id:
517 for folder in self._fetch_subalbums():
518 if folder["fields"].get("isDeleted") and folder["fields"]["isDeleted"]["value"]:
519 continue
521 folder_id = folder["recordName"]
522 folder_obj_type = f"CPLContainerRelationNotDeletedByAssetDate:{folder_id}"
523 folder_name = base64.b64decode(
524 folder["fields"]["albumNameEnc"]["value"],
525 ).decode("utf-8")
526 query_filter = [
527 {
528 "fieldName": "parentId",
529 "comparator": "EQUALS",
530 "fieldValue": {"type": "STRING", "value": folder_id},
531 },
532 ]
534 album = PhotoAlbum(
535 self.service,
536 name=folder_name,
537 list_type="CPLContainerRelationLiveByAssetDate",
538 obj_type=folder_obj_type,
539 direction="ASCENDING",
540 query_filter=query_filter,
541 folder_id=folder_id,
542 zone_id=self._zone_id,
543 )
544 self._subalbums[folder_name] = album
545 return self._subalbums
547 @property
548 def photos(self):
549 """Returns the album photos."""
550 if self.direction == "DESCENDING":
551 offset = len(self) - 1
552 else:
553 offset = 0
555 while True:
556 url = (f"{self.service._service_endpoint}/records/query?") + urlencode(
557 self.service.params,
558 )
559 data = json.dumps(
560 self._list_query_gen(
561 offset,
562 self.list_type,
563 self.direction,
564 self.query_filter,
565 ),
566 )
567 request = _zone_post(
568 self.service.session,
569 url,
570 data,
571 _zone_name(self._zone_id),
572 )
573 response = request.json()
575 asset_records = {}
576 master_records = []
577 for rec in response["records"]:
578 if rec["recordType"] == "CPLAsset":
579 master_id = rec["fields"]["masterRef"]["value"]["recordName"]
580 asset_records[master_id] = rec
581 elif rec["recordType"] == "CPLMaster":
582 master_records.append(rec)
584 master_records_len = len(master_records)
585 if master_records_len:
586 if self.direction == "DESCENDING":
587 offset = offset - master_records_len
588 else:
589 offset = offset + master_records_len
591 for master_record in master_records:
592 record_name = master_record["recordName"]
593 yield PhotoAsset(
594 self.service,
595 master_record,
596 asset_records[record_name],
597 )
598 else:
599 break
601 def _list_query_gen(self, offset, list_type, direction, query_filter=None):
602 query = {
603 "query": {
604 "filterBy": [
605 {
606 "fieldName": "startRank",
607 "fieldValue": {"type": "INT64", "value": offset},
608 "comparator": "EQUALS",
609 },
610 {
611 "fieldName": "direction",
612 "fieldValue": {"type": "STRING", "value": direction},
613 "comparator": "EQUALS",
614 },
615 ],
616 "recordType": list_type,
617 },
618 "resultsLimit": self.page_size * 2,
619 "desiredKeys": [
620 "resJPEGFullWidth",
621 "resJPEGFullHeight",
622 "resJPEGFullFileType",
623 "resJPEGFullFingerprint",
624 "resJPEGFullRes",
625 "resJPEGLargeWidth",
626 "resJPEGLargeHeight",
627 "resJPEGLargeFileType",
628 "resJPEGLargeFingerprint",
629 "resJPEGLargeRes",
630 "resJPEGMedWidth",
631 "resJPEGMedHeight",
632 "resJPEGMedFileType",
633 "resJPEGMedFingerprint",
634 "resJPEGMedRes",
635 "resJPEGThumbWidth",
636 "resJPEGThumbHeight",
637 "resJPEGThumbFileType",
638 "resJPEGThumbFingerprint",
639 "resJPEGThumbRes",
640 "resVidFullWidth",
641 "resVidFullHeight",
642 "resVidFullFileType",
643 "resVidFullFingerprint",
644 "resVidFullRes",
645 "resVidMedWidth",
646 "resVidMedHeight",
647 "resVidMedFileType",
648 "resVidMedFingerprint",
649 "resVidMedRes",
650 "resVidSmallWidth",
651 "resVidSmallHeight",
652 "resVidSmallFileType",
653 "resVidSmallFingerprint",
654 "resVidSmallRes",
655 "resSidecarWidth",
656 "resSidecarHeight",
657 "resSidecarFileType",
658 "resSidecarFingerprint",
659 "resSidecarRes",
660 "itemType",
661 "dataClassType",
662 "filenameEnc",
663 "originalOrientation",
664 "resOriginalWidth",
665 "resOriginalHeight",
666 "resOriginalFileType",
667 "resOriginalFingerprint",
668 "resOriginalRes",
669 "resOriginalAltWidth",
670 "resOriginalAltHeight",
671 "resOriginalAltFileType",
672 "resOriginalAltFingerprint",
673 "resOriginalAltRes",
674 "resOriginalVidComplWidth",
675 "resOriginalVidComplHeight",
676 "resOriginalVidComplFileType",
677 "resOriginalVidComplFingerprint",
678 "resOriginalVidComplRes",
679 "isDeleted",
680 "isExpunged",
681 "dateExpunged",
682 "remappedRef",
683 "recordName",
684 "recordType",
685 "recordChangeTag",
686 "masterRef",
687 "adjustmentRenderType",
688 "assetDate",
689 "addedDate",
690 "isFavorite",
691 "isHidden",
692 "orientation",
693 "duration",
694 "assetSubtype",
695 "assetSubtypeV2",
696 "assetHDRType",
697 "burstFlags",
698 "burstFlagsExt",
699 "burstId",
700 "captionEnc",
701 "extendedDescEnc",
702 "locationEnc",
703 "locationV2Enc",
704 "locationLatitude",
705 "locationLongitude",
706 "adjustmentType",
707 "timeZoneOffset",
708 "vidComplDurValue",
709 "vidComplDurScale",
710 "vidComplDispValue",
711 "vidComplDispScale",
712 "vidComplVisibilityState",
713 "customRenderedValue",
714 "containerId",
715 "itemId",
716 "position",
717 "isKeyAsset",
718 "importedByBundleIdentifierEnc",
719 "importedByDisplayNameEnc",
720 "importedBy",
721 ],
722 "zoneID": self._zone_id,
723 }
725 if query_filter:
726 query["query"]["filterBy"].extend(query_filter)
728 return query
730 def __unicode__(self):
731 return self.title
733 def __str__(self):
734 return self.__unicode__()
736 def __repr__(self):
737 return f"<{type(self).__name__}: '{self}'>"
740class PhotoAsset:
741 """A photo."""
743 def __init__(self, service, master_record, asset_record):
744 self._service = service
745 self._master_record = master_record
746 self._asset_record = asset_record
748 self._versions = None
750 # Apple Uniform Type Identifiers (UTIs) for known iCloud photo/video assets.
751 # Mirrors the table in icloud_photos_downloader's pyicloud_ipd.services.photos.
752 # Anything not listed here falls back to the filename-extension heuristic in
753 # `item_type` below.
754 ITEM_TYPES = {
755 "public.heic": "image",
756 "public.heif": "image",
757 "public.jpeg": "image",
758 "public.png": "image",
759 "public.tiff": "image",
760 "com.adobe.raw-image": "image",
761 "com.canon.cr2-raw-image": "image",
762 "com.canon.cr3-raw-image": "image",
763 "com.canon.crw-raw-image": "image",
764 "com.fuji.raw-image": "image",
765 "com.nikon.nrw-raw-image": "image",
766 "com.nikon.raw-image": "image",
767 "com.olympus.or-raw-image": "image",
768 "com.olympus.raw-image": "image",
769 "com.panasonic.rw2-raw-image": "image",
770 "com.pentax.raw-image": "image",
771 "com.sony.arw-raw-image": "image",
772 "com.apple.quicktime-movie": "movie",
773 "public.mpeg-4": "movie",
774 }
776 PHOTO_VERSION_LOOKUP = {
777 "full": "resJPEGFull",
778 "large": "resJPEGLarge",
779 "medium": "resJPEGMed",
780 "thumb": "resJPEGThumb",
781 "sidecar": "resSidecar",
782 "original": "resOriginal",
783 "original_alt": "resOriginalAlt",
784 # Live Photo video components — present alongside the still image for
785 # Live Photos. The CloudKit master_record carries `resOriginalVidCompl*`
786 # (the original .mov), and may also expose `resVidMed*` / `resVidSmall*`
787 # for smaller variants. If the asset is a regular still photo, these
788 # keys are silently skipped by `versions` since the underlying fields
789 # are absent.
790 "live_video_original": "resOriginalVidCompl",
791 "live_video_medium": "resVidMed",
792 "live_video_thumb": "resVidSmall",
793 }
795 VIDEO_VERSION_LOOKUP = {
796 "full": "resVidFull",
797 "medium": "resVidMed",
798 "thumb": "resVidSmall",
799 "original": "resOriginal",
800 "original_compl": "resOriginalVidCompl",
801 }
803 @property
804 def id(self):
805 """Gets the photo id."""
806 return self._master_record["recordName"]
808 @property
809 def filename(self):
810 """Gets the photo file name."""
811 try:
812 return base64.b64decode(
813 self._master_record["fields"]["filenameEnc"]["value"],
814 ).decode("utf-8")
815 except KeyError:
816 # Some photos/videos (e.g., GoPro videos) may not have a filename
817 return None
819 @property
820 def size(self):
821 """Gets the photo size."""
822 return self._master_record["fields"]["resOriginalRes"]["value"]["size"]
824 @property
825 def created(self):
826 """Gets the photo created date."""
827 return self.asset_date
829 @property
830 def asset_date(self):
831 """Gets the photo asset date."""
832 try:
833 return datetime.fromtimestamp(
834 self._asset_record["fields"]["assetDate"]["value"] / 1000.0,
835 tz=UTC,
836 )
837 except KeyError:
838 return datetime.fromtimestamp(0)
840 @property
841 def added_date(self):
842 """Gets the photo added date."""
843 return datetime.fromtimestamp(
844 self._asset_record["fields"]["addedDate"]["value"] / 1000.0,
845 tz=UTC,
846 )
848 @property
849 def dimensions(self):
850 """Gets the photo dimensions."""
851 return (
852 self._master_record["fields"]["resOriginalWidth"]["value"],
853 self._master_record["fields"]["resOriginalHeight"]["value"],
854 )
856 @property
857 def item_type(self):
858 """Returns 'image' or 'movie' for this asset.
860 Reads the CloudKit ``itemType`` field (Apple UTI string) and maps it
861 via ``ITEM_TYPES``. Falls back to a filename-extension heuristic when
862 the UTI is missing or unrecognised — that path catches GoPro footage
863 and a handful of camera-RAW formats that Apple has not assigned a
864 canonical UTI to.
866 Returns ``None`` only when both the UTI is missing AND no filename is
867 available (rare — usually a CloudKit record with no master payload).
868 """
869 fields = self._master_record["fields"]
870 uti = fields.get("itemType", {}).get("value")
871 if uti in self.ITEM_TYPES:
872 return self.ITEM_TYPES[uti]
874 # Extension-based fallback. Mirrors pyicloud_ipd's heuristic.
875 filename = self.filename
876 if filename:
877 lower = filename.lower()
878 if lower.endswith((".heic", ".heif", ".png", ".jpg", ".jpeg", ".tif", ".tiff")):
879 return "image"
880 if lower.endswith((".mov", ".mp4", ".m4v", ".avi", ".3gp")):
881 return "movie"
882 return None
884 @property
885 def versions(self):
886 """Gets the photo versions.
888 For ``item_type == "movie"`` returns the video versions
889 (``VIDEO_VERSION_LOOKUP``). For everything else returns the image
890 versions (``PHOTO_VERSION_LOOKUP``) — which includes Live Photo video
891 keys (``live_video_original`` / ``live_video_medium`` /
892 ``live_video_thumb``) when the underlying CloudKit fields are present.
894 Backward compatibility: callers that previously received only photo
895 versions for still images and only video versions for movies still see
896 those keys. Live Photo callers gain new ``live_video_*`` keys that
897 were previously inaccessible — earlier icloudpy versions misclassified
898 Live Photos as movies (via the ``resVidSmallRes`` presence heuristic)
899 and dropped the still image entirely.
900 """
901 if not self._versions:
902 self._versions = {}
903 if self.item_type == "movie":
904 typed_version_lookup = self.VIDEO_VERSION_LOOKUP
905 else:
906 typed_version_lookup = self.PHOTO_VERSION_LOOKUP
908 for key, prefix in typed_version_lookup.items():
909 if f"{prefix}Res" in self._master_record["fields"]:
910 fields = self._master_record["fields"]
911 version = {"filename": self.filename}
913 width_entry = fields.get(f"{prefix}Width")
914 if width_entry:
915 version["width"] = width_entry["value"]
916 else:
917 version["width"] = None
919 height_entry = fields.get(f"{prefix}Height")
920 if height_entry:
921 version["height"] = height_entry["value"]
922 else:
923 version["height"] = None
925 size_entry = fields.get(f"{prefix}Res")
926 if size_entry:
927 version["size"] = size_entry["value"]["size"]
928 version["url"] = size_entry["value"]["downloadURL"]
929 else:
930 version["size"] = None
931 version["url"] = None
933 type_entry = fields.get(f"{prefix}FileType")
934 if type_entry:
935 version["type"] = type_entry["value"]
936 else:
937 version["type"] = None
939 self._versions[key] = version
941 return self._versions
943 def download(self, version="original", **kwargs):
944 """Returns the photo file."""
945 if version not in self.versions:
946 return None
948 return self._service.session.get(
949 self.versions[version]["url"],
950 stream=True,
951 **kwargs,
952 )
954 def delete(self):
955 """Deletes the photo."""
956 json_data = json.dumps(
957 {
958 "atomic": True,
959 "desiredKeys": ["isDeleted"],
960 "operations": [
961 {
962 "operationType": "update",
963 "record": {
964 "fields": {"isDeleted": {"value": 1}},
965 "recordChangeTag": self._asset_record["recordChangeTag"],
966 "recordName": self._asset_record["recordName"],
967 "recordType": self._asset_record["recordType"],
968 },
969 },
970 ],
971 "zoneID": self._service.zone_id,
972 },
973 )
975 endpoint = self._service._service_endpoint
976 params = urlencode(self._service.params)
977 url = f"{endpoint}/records/modify?{params}"
979 return self._service.session.post(
980 url,
981 data=json_data,
982 headers={"Content-type": "text/plain"},
983 )
985 def __repr__(self):
986 return f"<{type(self).__name__}: id={self.id}>"