Coverage for icloudpy / services / photos.py: 92%
259 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-06-18 19:11 +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 ICloudPyServiceNotActivatedException
16LOGGER = logging.getLogger(__name__)
19class PhotoLibrary:
20 """Represents a library in the user's photos.
22 This provides access to all the albums as well as the photos.
23 """
25 SMART_FOLDERS = {
26 "All Photos": {
27 "obj_type": "CPLAssetByAddedDate",
28 "list_type": "CPLAssetAndMasterByAddedDate",
29 "direction": "ASCENDING",
30 "query_filter": None,
31 },
32 "Time-lapse": {
33 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Timelapse",
34 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
35 "direction": "ASCENDING",
36 "query_filter": [
37 {
38 "fieldName": "smartAlbum",
39 "comparator": "EQUALS",
40 "fieldValue": {"type": "STRING", "value": "TIMELAPSE"},
41 },
42 ],
43 },
44 "Videos": {
45 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Video",
46 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
47 "direction": "ASCENDING",
48 "query_filter": [
49 {
50 "fieldName": "smartAlbum",
51 "comparator": "EQUALS",
52 "fieldValue": {"type": "STRING", "value": "VIDEO"},
53 },
54 ],
55 },
56 "Slo-mo": {
57 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Slomo",
58 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
59 "direction": "ASCENDING",
60 "query_filter": [
61 {
62 "fieldName": "smartAlbum",
63 "comparator": "EQUALS",
64 "fieldValue": {"type": "STRING", "value": "SLOMO"},
65 },
66 ],
67 },
68 "Bursts": {
69 "obj_type": "CPLAssetBurstStackAssetByAssetDate",
70 "list_type": "CPLBurstStackAssetAndMasterByAssetDate",
71 "direction": "ASCENDING",
72 "query_filter": None,
73 },
74 "Favorites": {
75 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Favorite",
76 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
77 "direction": "ASCENDING",
78 "query_filter": [
79 {
80 "fieldName": "smartAlbum",
81 "comparator": "EQUALS",
82 "fieldValue": {"type": "STRING", "value": "FAVORITE"},
83 },
84 ],
85 },
86 "Panoramas": {
87 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Panorama",
88 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
89 "direction": "ASCENDING",
90 "query_filter": [
91 {
92 "fieldName": "smartAlbum",
93 "comparator": "EQUALS",
94 "fieldValue": {"type": "STRING", "value": "PANORAMA"},
95 },
96 ],
97 },
98 "Screenshots": {
99 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Screenshot",
100 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
101 "direction": "ASCENDING",
102 "query_filter": [
103 {
104 "fieldName": "smartAlbum",
105 "comparator": "EQUALS",
106 "fieldValue": {"type": "STRING", "value": "SCREENSHOT"},
107 },
108 ],
109 },
110 "Live": {
111 "obj_type": "CPLAssetInSmartAlbumByAssetDate:Live",
112 "list_type": "CPLAssetAndMasterInSmartAlbumByAssetDate",
113 "direction": "ASCENDING",
114 "query_filter": [
115 {
116 "fieldName": "smartAlbum",
117 "comparator": "EQUALS",
118 "fieldValue": {"type": "STRING", "value": "LIVE"},
119 },
120 ],
121 },
122 "Recently Deleted": {
123 "obj_type": "CPLAssetDeletedByExpungedDate",
124 "list_type": "CPLAssetAndMasterDeletedByExpungedDate",
125 "direction": "ASCENDING",
126 "query_filter": None,
127 },
128 "Hidden": {
129 "obj_type": "CPLAssetHiddenByAssetDate",
130 "list_type": "CPLAssetAndMasterHiddenByAssetDate",
131 "direction": "ASCENDING",
132 "query_filter": None,
133 },
134 }
136 def __init__(self, service, zone_id):
137 self.service = service
138 self.zone_id = zone_id
140 self._albums = None
142 url = f"{self.service._service_endpoint}/records/query?{urlencode(self.service.params)}"
143 json_data = json.dumps(
144 {
145 "query": {"recordType": "CheckIndexingState"},
146 "zoneID": self.zone_id,
147 },
148 )
150 request = self.service.session.post(
151 url,
152 data=json_data,
153 headers={"Content-type": "text/plain"},
154 )
155 response = request.json()
156 indexing_state = response["records"][0]["fields"]["state"]["value"]
157 if indexing_state != "FINISHED":
158 raise ICloudPyServiceNotActivatedException(
159 ("iCloud Photo Library not finished indexing. Please try " "again in a few minutes"),
160 None,
161 )
163 @property
164 def albums(self):
165 if not self._albums:
166 self._albums = {
167 name: PhotoAlbum(self.service, name, zone_id=self.zone_id, **props)
168 for (name, props) in self.SMART_FOLDERS.items()
169 }
171 for folder in self._fetch_folders():
172 if folder["recordName"] in (
173 "----Root-Folder----",
174 "----Project-Root-Folder----",
175 ) or (folder["fields"].get("isDeleted") and folder["fields"]["isDeleted"]["value"]):
176 continue
178 folder_id = folder["recordName"]
179 folder_obj_type = f"CPLContainerRelationNotDeletedByAssetDate:{folder_id}"
180 folder_name = base64.b64decode(
181 folder["fields"]["albumNameEnc"]["value"],
182 ).decode("utf-8")
183 query_filter = [
184 {
185 "fieldName": "parentId",
186 "comparator": "EQUALS",
187 "fieldValue": {"type": "STRING", "value": folder_id},
188 },
189 ]
191 album = PhotoAlbum(
192 self.service,
193 folder_name,
194 "CPLContainerRelationLiveByAssetDate",
195 folder_obj_type,
196 "ASCENDING",
197 query_filter,
198 folder_id=folder_id,
199 zone_id=self.zone_id,
200 )
201 self._albums[folder_name] = album
203 return self._albums
205 def _fetch_folders(self):
206 url = f"{self.service._service_endpoint}/records/query?{urlencode(self.service.params)}"
207 json_data = json.dumps(
208 {
209 "query": {"recordType": "CPLAlbumByPositionLive"},
210 "zoneID": self.zone_id,
211 },
212 )
214 request = self.service.session.post(
215 url,
216 data=json_data,
217 headers={"Content-type": "text/plain"},
218 )
219 response = request.json()
221 return response["records"]
223 @property
224 def all(self):
225 return self.albums["All Photos"]
228class PhotosService(PhotoLibrary):
229 """The 'Photos' iCloud service.
231 This also acts as a way to access the user's primary library.
232 """
234 def __init__(self, service_root, session, params):
235 self.session = session
236 self.params = dict(params)
237 self._service_root = service_root
238 self._service_endpoint = f"{self._service_root}/database/1/com.apple.photos.cloud/production/private"
240 self._libraries = None
242 self.params.update({"remapEnums": True, "getCurrentSyncToken": True})
244 self._photo_assets = {}
246 super().__init__(service=self, zone_id={"zoneName": "PrimarySync"})
248 @property
249 def libraries(self):
250 if not self._libraries:
251 try:
252 url = f"{self._service_endpoint}/zones/list"
253 request = self.session.post(
254 url,
255 data="{}",
256 headers={"Content-type": "text/plain"},
257 )
258 response = request.json()
259 zones = response["zones"]
260 except Exception as e:
261 LOGGER.error(f"library exception: {str(e)}")
263 libraries = {}
264 for zone in zones:
265 if not zone.get("deleted"):
266 zone_name = zone["zoneID"]["zoneName"]
267 libraries[zone_name] = PhotoLibrary(self, zone_id=zone["zoneID"])
268 # obj_type='CPLAssetByAssetDateWithoutHiddenOrDeleted',
269 # list_type="CPLAssetAndMasterByAssetDateWithoutHiddenOrDeleted",
270 # direction="ASCENDING", query_filter=None,
271 # zone_id=zone['zoneID'])
273 self._libraries = libraries
275 return self._libraries
278class PhotoAlbum:
279 """A photo album."""
281 def __init__(
282 self,
283 service,
284 name,
285 list_type,
286 obj_type,
287 direction,
288 query_filter=None,
289 page_size=100,
290 folder_id=None,
291 zone_id=None,
292 ):
293 self.name = name
294 self.service = service
295 self.list_type = list_type
296 self.obj_type = obj_type
297 self.direction = direction
298 self.query_filter = query_filter
299 self.page_size = page_size
300 self.folder_id = folder_id
302 if zone_id:
303 self._zone_id = zone_id
304 else:
305 self._zone_id = "PrimarySync"
307 self._len = None
309 self._subalbums = {}
311 @property
312 def title(self):
313 """Gets the album name."""
314 return self.name
316 def __iter__(self):
317 return self.photos
319 def iter_chunks(self, chunk_size=1000):
320 """Yield lists of PhotoAsset objects in fixed-size batches.
322 Equivalent to building a list from ``__iter__`` and slicing,
323 but the per-chunk list is yielded eagerly so callers can
324 process + release each chunk before the next is materialised.
325 Lets bulk-download callers bound peak memory by the chunk size
326 rather than ``len(self)`` -- relevant for 100K+ libraries where
327 ``list(album)`` would otherwise hold every PhotoAsset (and any
328 per-asset state callers attach) in memory at once.
330 The underlying ``photos`` property already paginates HTTP
331 responses (``page_size * 2`` records per request); this method
332 is a pure-Python wrapper that batches the yielded
333 ``PhotoAsset`` instances without changing the HTTP fetch
334 pattern.
336 Args:
337 chunk_size: Max number of PhotoAsset objects per yielded
338 list. Values <= 0 are coerced to the default (1000).
340 Yields:
341 List of up to ``chunk_size`` ``PhotoAsset`` instances. The
342 final chunk may be smaller. No empty list is yielded for
343 an empty album.
344 """
345 if chunk_size <= 0:
346 chunk_size = 1000
347 buffer = []
348 for photo in self:
349 buffer.append(photo)
350 if len(buffer) >= chunk_size:
351 yield buffer
352 buffer = []
353 if buffer:
354 yield buffer
356 def __len__(self):
357 if self._len is None:
358 url = f"{self.service._service_endpoint}/internal/records/query/batch?{urlencode(self.service.params)}"
359 request = self.service.session.post(
360 url,
361 data=json.dumps(
362 {
363 "batch": [
364 {
365 "resultsLimit": 1,
366 "query": {
367 "filterBy": {
368 "fieldName": "indexCountID",
369 "fieldValue": {
370 "type": "STRING_LIST",
371 "value": [self.obj_type],
372 },
373 "comparator": "IN",
374 },
375 "recordType": "HyperionIndexCountLookup",
376 },
377 "zoneWide": True,
378 "zoneID": {"zoneName": self._zone_id["zoneName"]},
379 },
380 ],
381 },
382 ),
383 headers={"Content-type": "text/plain"},
384 )
385 response = request.json()
387 self._len = response["batch"][0]["records"][0]["fields"]["itemCount"]["value"]
389 return self._len
391 def _fetch_subalbums(self):
392 url = (f"{self.service._service_endpoint}/records/query?") + urlencode(
393 self.service.params,
394 )
395 # pylint: disable=consider-using-f-string
396 query = """{{
397 "query": {{
398 "recordType":"CPLAlbumByPositionLive",
399 "filterBy": [
400 {{
401 "fieldName": "parentId",
402 "comparator": "EQUALS",
403 "fieldValue": {{
404 "value": "{}",
405 "type": "STRING"
406 }}
407 }}
408 ]
409 }},
410 "zoneID": {{
411 "zoneName":"{}"
412 }}
413 }}""".format(
414 self.folder_id,
415 self._zone_id["zoneName"],
416 )
417 json_data = query
418 request = self.service.session.post(
419 url,
420 data=json_data,
421 headers={"Content-type": "text/plain"},
422 )
423 response = request.json()
425 return response["records"]
427 @property
428 def subalbums(self):
429 """Returns the subalbums"""
430 if not self._subalbums and self.folder_id:
431 for folder in self._fetch_subalbums():
432 if folder["fields"].get("isDeleted") and folder["fields"]["isDeleted"]["value"]:
433 continue
435 folder_id = folder["recordName"]
436 folder_obj_type = f"CPLContainerRelationNotDeletedByAssetDate:{folder_id}"
437 folder_name = base64.b64decode(
438 folder["fields"]["albumNameEnc"]["value"],
439 ).decode("utf-8")
440 query_filter = [
441 {
442 "fieldName": "parentId",
443 "comparator": "EQUALS",
444 "fieldValue": {"type": "STRING", "value": folder_id},
445 },
446 ]
448 album = PhotoAlbum(
449 self.service,
450 name=folder_name,
451 list_type="CPLContainerRelationLiveByAssetDate",
452 obj_type=folder_obj_type,
453 direction="ASCENDING",
454 query_filter=query_filter,
455 folder_id=folder_id,
456 zone_id=self._zone_id,
457 )
458 self._subalbums[folder_name] = album
459 return self._subalbums
461 @property
462 def photos(self):
463 """Returns the album photos."""
464 if self.direction == "DESCENDING":
465 offset = len(self) - 1
466 else:
467 offset = 0
469 while True:
470 url = (f"{self.service._service_endpoint}/records/query?") + urlencode(
471 self.service.params,
472 )
473 request = self.service.session.post(
474 url,
475 data=json.dumps(
476 self._list_query_gen(
477 offset,
478 self.list_type,
479 self.direction,
480 self.query_filter,
481 ),
482 ),
483 headers={"Content-type": "text/plain"},
484 )
485 response = request.json()
487 asset_records = {}
488 master_records = []
489 for rec in response["records"]:
490 if rec["recordType"] == "CPLAsset":
491 master_id = rec["fields"]["masterRef"]["value"]["recordName"]
492 asset_records[master_id] = rec
493 elif rec["recordType"] == "CPLMaster":
494 master_records.append(rec)
496 master_records_len = len(master_records)
497 if master_records_len:
498 if self.direction == "DESCENDING":
499 offset = offset - master_records_len
500 else:
501 offset = offset + master_records_len
503 for master_record in master_records:
504 record_name = master_record["recordName"]
505 yield PhotoAsset(
506 self.service,
507 master_record,
508 asset_records[record_name],
509 )
510 else:
511 break
513 def _list_query_gen(self, offset, list_type, direction, query_filter=None):
514 query = {
515 "query": {
516 "filterBy": [
517 {
518 "fieldName": "startRank",
519 "fieldValue": {"type": "INT64", "value": offset},
520 "comparator": "EQUALS",
521 },
522 {
523 "fieldName": "direction",
524 "fieldValue": {"type": "STRING", "value": direction},
525 "comparator": "EQUALS",
526 },
527 ],
528 "recordType": list_type,
529 },
530 "resultsLimit": self.page_size * 2,
531 "desiredKeys": [
532 "resJPEGFullWidth",
533 "resJPEGFullHeight",
534 "resJPEGFullFileType",
535 "resJPEGFullFingerprint",
536 "resJPEGFullRes",
537 "resJPEGLargeWidth",
538 "resJPEGLargeHeight",
539 "resJPEGLargeFileType",
540 "resJPEGLargeFingerprint",
541 "resJPEGLargeRes",
542 "resJPEGMedWidth",
543 "resJPEGMedHeight",
544 "resJPEGMedFileType",
545 "resJPEGMedFingerprint",
546 "resJPEGMedRes",
547 "resJPEGThumbWidth",
548 "resJPEGThumbHeight",
549 "resJPEGThumbFileType",
550 "resJPEGThumbFingerprint",
551 "resJPEGThumbRes",
552 "resVidFullWidth",
553 "resVidFullHeight",
554 "resVidFullFileType",
555 "resVidFullFingerprint",
556 "resVidFullRes",
557 "resVidMedWidth",
558 "resVidMedHeight",
559 "resVidMedFileType",
560 "resVidMedFingerprint",
561 "resVidMedRes",
562 "resVidSmallWidth",
563 "resVidSmallHeight",
564 "resVidSmallFileType",
565 "resVidSmallFingerprint",
566 "resVidSmallRes",
567 "resSidecarWidth",
568 "resSidecarHeight",
569 "resSidecarFileType",
570 "resSidecarFingerprint",
571 "resSidecarRes",
572 "itemType",
573 "dataClassType",
574 "filenameEnc",
575 "originalOrientation",
576 "resOriginalWidth",
577 "resOriginalHeight",
578 "resOriginalFileType",
579 "resOriginalFingerprint",
580 "resOriginalRes",
581 "resOriginalAltWidth",
582 "resOriginalAltHeight",
583 "resOriginalAltFileType",
584 "resOriginalAltFingerprint",
585 "resOriginalAltRes",
586 "resOriginalVidComplWidth",
587 "resOriginalVidComplHeight",
588 "resOriginalVidComplFileType",
589 "resOriginalVidComplFingerprint",
590 "resOriginalVidComplRes",
591 "isDeleted",
592 "isExpunged",
593 "dateExpunged",
594 "remappedRef",
595 "recordName",
596 "recordType",
597 "recordChangeTag",
598 "masterRef",
599 "adjustmentRenderType",
600 "assetDate",
601 "addedDate",
602 "isFavorite",
603 "isHidden",
604 "orientation",
605 "duration",
606 "assetSubtype",
607 "assetSubtypeV2",
608 "assetHDRType",
609 "burstFlags",
610 "burstFlagsExt",
611 "burstId",
612 "captionEnc",
613 "extendedDescEnc",
614 "locationEnc",
615 "locationV2Enc",
616 "locationLatitude",
617 "locationLongitude",
618 "adjustmentType",
619 "timeZoneOffset",
620 "vidComplDurValue",
621 "vidComplDurScale",
622 "vidComplDispValue",
623 "vidComplDispScale",
624 "vidComplVisibilityState",
625 "customRenderedValue",
626 "containerId",
627 "itemId",
628 "position",
629 "isKeyAsset",
630 "importedByBundleIdentifierEnc",
631 "importedByDisplayNameEnc",
632 "importedBy",
633 ],
634 "zoneID": self._zone_id,
635 }
637 if query_filter:
638 query["query"]["filterBy"].extend(query_filter)
640 return query
642 def __unicode__(self):
643 return self.title
645 def __str__(self):
646 return self.__unicode__()
648 def __repr__(self):
649 return f"<{type(self).__name__}: '{self}'>"
652class PhotoAsset:
653 """A photo."""
655 def __init__(self, service, master_record, asset_record):
656 self._service = service
657 self._master_record = master_record
658 self._asset_record = asset_record
660 self._versions = None
662 # Apple Uniform Type Identifiers (UTIs) for known iCloud photo/video assets.
663 # Mirrors the table in icloud_photos_downloader's pyicloud_ipd.services.photos.
664 # Anything not listed here falls back to the filename-extension heuristic in
665 # `item_type` below.
666 ITEM_TYPES = {
667 "public.heic": "image",
668 "public.heif": "image",
669 "public.jpeg": "image",
670 "public.png": "image",
671 "public.tiff": "image",
672 "com.adobe.raw-image": "image",
673 "com.canon.cr2-raw-image": "image",
674 "com.canon.cr3-raw-image": "image",
675 "com.canon.crw-raw-image": "image",
676 "com.fuji.raw-image": "image",
677 "com.nikon.nrw-raw-image": "image",
678 "com.nikon.raw-image": "image",
679 "com.olympus.or-raw-image": "image",
680 "com.olympus.raw-image": "image",
681 "com.panasonic.rw2-raw-image": "image",
682 "com.pentax.raw-image": "image",
683 "com.sony.arw-raw-image": "image",
684 "com.apple.quicktime-movie": "movie",
685 "public.mpeg-4": "movie",
686 }
688 PHOTO_VERSION_LOOKUP = {
689 "full": "resJPEGFull",
690 "large": "resJPEGLarge",
691 "medium": "resJPEGMed",
692 "thumb": "resJPEGThumb",
693 "sidecar": "resSidecar",
694 "original": "resOriginal",
695 "original_alt": "resOriginalAlt",
696 # Live Photo video components — present alongside the still image for
697 # Live Photos. The CloudKit master_record carries `resOriginalVidCompl*`
698 # (the original .mov), and may also expose `resVidMed*` / `resVidSmall*`
699 # for smaller variants. If the asset is a regular still photo, these
700 # keys are silently skipped by `versions` since the underlying fields
701 # are absent.
702 "live_video_original": "resOriginalVidCompl",
703 "live_video_medium": "resVidMed",
704 "live_video_thumb": "resVidSmall",
705 }
707 VIDEO_VERSION_LOOKUP = {
708 "full": "resVidFull",
709 "medium": "resVidMed",
710 "thumb": "resVidSmall",
711 "original": "resOriginal",
712 "original_compl": "resOriginalVidCompl",
713 }
715 @property
716 def id(self):
717 """Gets the photo id."""
718 return self._master_record["recordName"]
720 @property
721 def filename(self):
722 """Gets the photo file name."""
723 try:
724 return base64.b64decode(
725 self._master_record["fields"]["filenameEnc"]["value"],
726 ).decode("utf-8")
727 except KeyError:
728 # Some photos/videos (e.g., GoPro videos) may not have a filename
729 return None
731 @property
732 def size(self):
733 """Gets the photo size."""
734 return self._master_record["fields"]["resOriginalRes"]["value"]["size"]
736 @property
737 def created(self):
738 """Gets the photo created date."""
739 return self.asset_date
741 @property
742 def asset_date(self):
743 """Gets the photo asset date."""
744 try:
745 return datetime.fromtimestamp(
746 self._asset_record["fields"]["assetDate"]["value"] / 1000.0,
747 tz=UTC,
748 )
749 except KeyError:
750 return datetime.fromtimestamp(0)
752 @property
753 def added_date(self):
754 """Gets the photo added date."""
755 return datetime.fromtimestamp(
756 self._asset_record["fields"]["addedDate"]["value"] / 1000.0,
757 tz=UTC,
758 )
760 @property
761 def dimensions(self):
762 """Gets the photo dimensions."""
763 return (
764 self._master_record["fields"]["resOriginalWidth"]["value"],
765 self._master_record["fields"]["resOriginalHeight"]["value"],
766 )
768 @property
769 def item_type(self):
770 """Returns 'image' or 'movie' for this asset.
772 Reads the CloudKit ``itemType`` field (Apple UTI string) and maps it
773 via ``ITEM_TYPES``. Falls back to a filename-extension heuristic when
774 the UTI is missing or unrecognised — that path catches GoPro footage
775 and a handful of camera-RAW formats that Apple has not assigned a
776 canonical UTI to.
778 Returns ``None`` only when both the UTI is missing AND no filename is
779 available (rare — usually a CloudKit record with no master payload).
780 """
781 fields = self._master_record["fields"]
782 uti = fields.get("itemType", {}).get("value")
783 if uti in self.ITEM_TYPES:
784 return self.ITEM_TYPES[uti]
786 # Extension-based fallback. Mirrors pyicloud_ipd's heuristic.
787 filename = self.filename
788 if filename:
789 lower = filename.lower()
790 if lower.endswith((".heic", ".heif", ".png", ".jpg", ".jpeg", ".tif", ".tiff")):
791 return "image"
792 if lower.endswith((".mov", ".mp4", ".m4v", ".avi", ".3gp")):
793 return "movie"
794 return None
796 @property
797 def versions(self):
798 """Gets the photo versions.
800 For ``item_type == "movie"`` returns the video versions
801 (``VIDEO_VERSION_LOOKUP``). For everything else returns the image
802 versions (``PHOTO_VERSION_LOOKUP``) — which includes Live Photo video
803 keys (``live_video_original`` / ``live_video_medium`` /
804 ``live_video_thumb``) when the underlying CloudKit fields are present.
806 Backward compatibility: callers that previously received only photo
807 versions for still images and only video versions for movies still see
808 those keys. Live Photo callers gain new ``live_video_*`` keys that
809 were previously inaccessible — earlier icloudpy versions misclassified
810 Live Photos as movies (via the ``resVidSmallRes`` presence heuristic)
811 and dropped the still image entirely.
812 """
813 if not self._versions:
814 self._versions = {}
815 if self.item_type == "movie":
816 typed_version_lookup = self.VIDEO_VERSION_LOOKUP
817 else:
818 typed_version_lookup = self.PHOTO_VERSION_LOOKUP
820 for key, prefix in typed_version_lookup.items():
821 if f"{prefix}Res" in self._master_record["fields"]:
822 fields = self._master_record["fields"]
823 version = {"filename": self.filename}
825 width_entry = fields.get(f"{prefix}Width")
826 if width_entry:
827 version["width"] = width_entry["value"]
828 else:
829 version["width"] = None
831 height_entry = fields.get(f"{prefix}Height")
832 if height_entry:
833 version["height"] = height_entry["value"]
834 else:
835 version["height"] = None
837 size_entry = fields.get(f"{prefix}Res")
838 if size_entry:
839 version["size"] = size_entry["value"]["size"]
840 version["url"] = size_entry["value"]["downloadURL"]
841 else:
842 version["size"] = None
843 version["url"] = None
845 type_entry = fields.get(f"{prefix}FileType")
846 if type_entry:
847 version["type"] = type_entry["value"]
848 else:
849 version["type"] = None
851 self._versions[key] = version
853 return self._versions
855 def download(self, version="original", **kwargs):
856 """Returns the photo file."""
857 if version not in self.versions:
858 return None
860 return self._service.session.get(
861 self.versions[version]["url"],
862 stream=True,
863 **kwargs,
864 )
866 def delete(self):
867 """Deletes the photo."""
868 json_data = json.dumps(
869 {
870 "atomic": True,
871 "desiredKeys": ["isDeleted"],
872 "operations": [
873 {
874 "operationType": "update",
875 "record": {
876 "fields": {"isDeleted": {"value": 1}},
877 "recordChangeTag": self._asset_record["recordChangeTag"],
878 "recordName": self._asset_record["recordName"],
879 "recordType": self._asset_record["recordType"],
880 },
881 },
882 ],
883 "zoneID": self._service.zone_id,
884 },
885 )
887 endpoint = self._service._service_endpoint
888 params = urlencode(self._service.params)
889 url = f"{endpoint}/records/modify?{params}"
891 return self._service.session.post(
892 url,
893 data=json_data,
894 headers={"Content-type": "text/plain"},
895 )
897 def __repr__(self):
898 return f"<{type(self).__name__}: id={self.id}>"