Skip to content

Commit

Permalink
Add support for shared library
Browse files Browse the repository at this point in the history 
Based on icloud-photos-downloader/icloud_photos_downloader#678
  • Loading branch information
@tymmej
tymmej committed Oct 5, 2023
1 parent c1bf5fe commit 2379ff2
Showing 1 changed file with 133 additions and 82 deletions.
215 changes: 133 additions & 82 deletions icloudpy/services/photos.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@
# fmt: on


class PhotosService:
"""The 'Photos' iCloud service."""
class PhotoLibrary(object):
"""Represents a library in the user's photos.
This provides access to all the albums as well as the photos.
"""
SMART_FOLDERS = {
"All Photos": {
"obj_type": "CPLAssetByAddedDate",
Expand Down Expand Up @@ -128,110 +130,152 @@ class PhotosService:
},
}

def __init__(self, service_root, session, params):
self.session = session
self.params = dict(params)
self._service_root = service_root
self.service_endpoint = (
f"{self._service_root}/database/1/com.apple.photos.cloud/production/private"
)
def __init__(self, service, zone_id):
self.service = service
self.zone_id = zone_id

self._albums = None

self.params.update({"remapEnums": True, "getCurrentSyncToken": True})
url = ('%s/records/query?%s' %
(self.service._service_endpoint, urlencode(self.service.params)))
json_data = json.dumps({
"query": {"recordType":"CheckIndexingState"},
"zoneID": self.zone_id,
})

url = f"{self.service_endpoint}/records/query?{urlencode(self.params)}"
json_data = (
'{"query":{"recordType":"CheckIndexingState"},'
'"zoneID":{"zoneName":"PrimarySync"}}'
)
request = self.session.post(
url, data=json_data, headers={"Content-type": "text/plain"}
request = self.service.session.post(
url,
data=json_data,
headers={'Content-type': 'text/plain'}
)
response = request.json()
indexing_state = response["records"][0]["fields"]["state"]["value"]
if indexing_state != "FINISHED":
raise ICloudPyServiceNotActivatedException(
"iCloud Photo Library not finished indexing. "
"Please try again in a few minutes."
)

# TODO: Does syncToken ever change? # pylint: disable=fixme
# self.params.update({
# 'syncToken': response['syncToken'],
# 'clientInstanceId': self.params.pop('clientId')
# })

self._photo_assets = {}
indexing_state = response['records'][0]['fields']['state']['value']
if indexing_state != 'FINISHED':
raise PyiCloudServiceNotActivatedErrror(
('iCloud Photo Library not finished indexing. Please try '
'again in a few minutes'), None)

@property
def albums(self):
"""Returns photo albums."""
if not self._albums:
self._albums = {}
self._albums = {
name: PhotoAlbum(self.service, name, zone_id=self.zone_id, **props)
for (name, props) in self.SMART_FOLDERS.items()
}

for folder in self._fetch_folders():

# Skipping albums having null name, that can happen sometime
if "albumNameEnc" not in folder["fields"]:
# FIXME: Handle subfolders
if folder['recordName'] in ('----Root-Folder----',
'----Project-Root-Folder----') or \
(folder['fields'].get('isDeleted') and
folder['fields']['isDeleted']['value']):
continue

if folder["recordName"] == "----Root-Folder----" or (
folder["fields"].get("isDeleted")
and folder["fields"]["isDeleted"]["value"]
):
continue

folder_id = folder["recordName"]
folder_obj_type = (
f"CPLContainerRelationNotDeletedByAssetDate:{folder_id}"
)
folder_id = folder['recordName']
folder_obj_type = \
"CPLContainerRelationNotDeletedByAssetDate:%s" % folder_id
folder_name = base64.b64decode(
folder["fields"]["albumNameEnc"]["value"]
).decode("utf-8")

query_filter = [
{
"fieldName": "parentId",
"comparator": "EQUALS",
"fieldValue": {"type": "STRING", "value": folder_id},
folder['fields']['albumNameEnc']['value']).decode('utf-8')
query_filter = [{
"fieldName": "parentId",
"comparator": "EQUALS",
"fieldValue": {
"type": "STRING",
"value": folder_id
}
]
}]

album = PhotoAlbum(
self,
name=folder_name,
list_type="CPLContainerRelationLiveByAssetDate",
obj_type=folder_obj_type,
direction="ASCENDING",
query_filter=query_filter,
folder_id=folder_id,
)
album = PhotoAlbum(self.service, folder_name,
'CPLContainerRelationLiveByAssetDate',
folder_obj_type, 'ASCENDING', query_filter,
zone_id=self.zone_id)
self._albums[folder_name] = album

for (name, props) in self.SMART_FOLDERS.items():
self._albums[name] = PhotoAlbum(self, name, **props)

return self._albums

def _fetch_folders(self):
url = f"{self.service_endpoint}/records/query?{urlencode(self.params)}"
json_data = (
'{"query":{"recordType":"CPLAlbumByPositionLive"},'
'"zoneID":{"zoneName":"PrimarySync"}}'
)
url = ('%s/records/query?%s' %
(self.service._service_endpoint, urlencode(self.service.params)))
json_data = json.dumps({
"query": {"recordType":"CPLAlbumByPositionLive"},
"zoneID": self.zone_id,
})

request = self.session.post(
url, data=json_data, headers={"Content-type": "text/plain"}
request = self.service.session.post(
url,
data=json_data,
headers={'Content-type': 'text/plain'}
)
response = request.json()

return response["records"]
return response['records']

@property
def all(self):
"""Returns all photos."""
return self.albums["All Photos"]
return self.albums['All Photos']


class PhotosService(PhotoLibrary):
"""The 'Photos' iCloud service.
This also acts as a way to access the user's primary library.
"""
def __init__(self, service_root, session, params):
self.session = session
self.params = dict(params)
self._service_root = service_root
self._service_endpoint = \
('%s/database/1/com.apple.photos.cloud/production/private'
% self._service_root)

self._libraries = None

self.params.update({
'remapEnums': True,
'getCurrentSyncToken': True
})

# TODO: Does syncToken ever change?
# self.params.update({
# 'syncToken': response['syncToken'],
# 'clientInstanceId': self.params.pop('clientId')
# })

self._photo_assets = {}

super(PhotosService, self).__init__(
service=self, zone_id={u'zoneName': u'PrimarySync'})

@property
def libraries(self):
if not self._libraries:
try:
url = ('%s/changes/database' %
(self._service_endpoint, ))
request = self.session.post(
url,
data='{}',
headers={'Content-type': 'text/plain'}
)
response = request.json()
zones = response['zones']
except Exception as e:
logger.error("library exception: %s" % str(e))

libraries = {}
for zone in zones:
if not zone.get('deleted'):
zone_name = zone['zoneID']['zoneName']
libraries[zone_name] = PhotoLibrary(
self, zone_id=zone['zoneID'])
# obj_type='CPLAssetByAssetDateWithoutHiddenOrDeleted',
# list_type="CPLAssetAndMasterByAssetDateWithoutHiddenOrDeleted",
# direction="ASCENDING", query_filter=None,
# zone_id=zone['zoneID'])

self._libraries = libraries

return self._libraries


class PhotoAlbum:
Expand All @@ -247,6 +291,7 @@ def __init__(
query_filter=None,
page_size=100,
folder_id=None,
zone_id=None,
):
self.name = name
self.service = service
Expand All @@ -257,6 +302,11 @@ def __init__(
self.page_size = page_size
self.folder_id = folder_id

if zone_id:
self._zone_id = zone_id
else:
self._zone_id = 'PrimarySync'

self._len = None

self._subalbums = {}
Expand Down Expand Up @@ -291,7 +341,7 @@ def __len__(self):
"recordType": "HyperionIndexCountLookup",
},
"zoneWide": True,
"zoneID": {"zoneName": "PrimarySync"},
"zoneID": {"zoneName": self._zone_id},
}
]
}
Expand Down Expand Up @@ -326,10 +376,11 @@ def _fetch_subalbums(self):
]
}},
"zoneID": {{
"zoneName":"PrimarySync"
"zoneName":"{}"
}}
}}""".format(
self.folder_id
self.folder_id,
self._zone_id
)
json_data = query
request = self.service.session.post(
Expand Down Expand Up @@ -388,7 +439,7 @@ def photos(self):
offset = 0

while True:
url = (f"{self.service.service_endpoint}/records/query?") + urlencode(
url = (f"{self.service._service_endpoint}/records/query?") + urlencode(
self.service.params
)
request = self.service.session.post(
Expand Down Expand Up @@ -543,7 +594,7 @@ def _list_query_gen(self, offset, list_type, direction, query_filter=None):
"position",
"isKeyAsset",
],
"zoneID": {"zoneName": "PrimarySync"},
"zoneID": self._zone_id,
}

if query_filter:
Expand Down

0 comments on commit 2379ff2

Please sign in to comment.