delete icloud photo in respective shared lib #802 (#842)

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- fix: delete iCloud asset in respective shared library [#802](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/802)
+
 ## 1.17.6 (2024-05-23)
 
 - fix: missing exception [#836](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/836)

scripts/type_check

@@ -1,5 +1,5 @@
 #!/bin/bash
 set -euo pipefail
 echo "Running mypy..."
-python3 -m mypy src tests --strict-equality --warn-return-any --disallow-any-generics 
-# too strict now: --disallow-untyped-defs --disallow-untyped-calls --check-untyped-defs
+python3 -m mypy src tests --strict --check-untyped-defs
+# --strict-equality --warn-return-any --disallow-any-generics --disallow-untyped-defs --disallow-untyped-calls --check-untyped-defs

src/icloudpd/authentication.py

@@ -5,6 +5,7 @@
 from typing import Callable, Optional
 import click
 import pyicloud_ipd
+from pyicloud_ipd.base import PyiCloudService
 
 
 class TwoStepAuthRequiredError(Exception):
@@ -14,22 +15,22 @@ class TwoStepAuthRequiredError(Exception):
  """
 
 
-def authenticator(logger: logging.Logger, domain: str) -> Callable[[str, Optional[str], Optional[str], bool, Optional[str]], pyicloud_ipd.PyiCloudService]:
+def authenticator(logger: logging.Logger, domain: str) -> Callable[[str, Optional[str], Optional[str], bool, Optional[str]], PyiCloudService]:
  """Wraping authentication with domain context"""
  def authenticate_(
  username:str,
  password:Optional[str],
  cookie_directory:Optional[str]=None,
  raise_error_on_2sa:bool=False,
  client_id:Optional[str]=None,
- ) -> pyicloud_ipd.PyiCloudService:
+ ) -> PyiCloudService:
  """Authenticate with iCloud username and password"""
  logger.debug("Authenticating...")
  while True:
  try:
  # If password not provided on command line variable will be set to None
  # and PyiCloud will attempt to retrieve from its keyring
- icloud = pyicloud_ipd.PyiCloudService(
+ icloud = PyiCloudService(
  domain,
  username, password,
  cookie_directory=cookie_directory,
@@ -60,7 +61,7 @@ def authenticate_(
  return authenticate_
 
 
-def request_2sa(icloud: pyicloud_ipd.PyiCloudService, logger: logging.Logger) -> None:
+def request_2sa(icloud: PyiCloudService, logger: logging.Logger) -> None:
  """Request two-step authentication. Prompts for SMS or device"""
  devices = icloud.trusted_devices
  devices_count = len(devices)
@@ -100,7 +101,7 @@ def request_2sa(icloud: pyicloud_ipd.PyiCloudService, logger: logging.Logger) ->
  )
 
 
-def request_2fa(icloud: pyicloud_ipd.PyiCloudService, logger: logging.Logger) -> None:
+def request_2fa(icloud: PyiCloudService, logger: logging.Logger) -> None:
  """Request two-factor authentication."""
  try:
  devices = icloud.trusted_devices

src/icloudpd/base.py

@@ -10,9 +10,9 @@
 from icloudpd.email_notifications import send_2sa_notification
 from icloudpd import download
 from icloudpd.authentication import authenticator, TwoStepAuthRequiredError
-from pyicloud_ipd.services.photos import PhotoAsset, PhotoLibrary
+from pyicloud_ipd.services.photos import PhotoAsset, PhotoLibrary, PhotosService
 from pyicloud_ipd.exceptions import PyiCloudAPIResponseException
-from pyicloud_ipd import PyiCloudService
+from pyicloud_ipd.base import PyiCloudService
 from tzlocal import get_localzone
 from tqdm.contrib.logging import logging_redirect_tqdm
 from tqdm import tqdm
@@ -603,15 +603,16 @@ def download_photo_(counter: Counter, photo: PhotoAsset) -> bool:
 
 def delete_photo(
  logger: logging.Logger,
- icloud: PyiCloudService,
+ photo_service: PhotosService,
+ library_object: PhotoLibrary,
  photo: PhotoAsset) -> None:
  """Delete a photo from the iCloud account."""
  clean_filename_local = clean_filename(photo.filename)
  logger.debug(
  "Deleting %s in iCloud...", clean_filename_local)
  # pylint: disable=W0212
- url = f"{icloud.photos._service_endpoint}/records/modify?"\
- f"{urllib.parse.urlencode(icloud.photos.params)}"
+ url = f"{photo_service._service_endpoint}/records/modify?"\
+ f"{urllib.parse.urlencode(photo_service.params)}"
  post_data = json.dumps(
  {
  "atomic": True,
@@ -625,10 +626,10 @@ def delete_photo(
  "recordType": "CPLAsset",
  }
  }],
- "zoneID": {"zoneName": "PrimarySync"}
+ "zoneID": library_object.zone_id
  }
  )
- icloud.photos.session.post(
+ photo_service.session.post(
  url, data=post_data, headers={
  "Content-type": "application/json"})
  logger.info(
@@ -637,12 +638,14 @@ def delete_photo(
 
 def delete_photo_dry_run(
  logger: logging.Logger,
- _icloud: PyiCloudService,
+ _photo_service: PhotosService,
+ library_object: PhotoLibrary,
  photo: PhotoAsset) -> None:
  """Dry run for deleting a photo from the iCloud"""
  logger.info(
- "[DRY RUN] Would delete %s in iCloud",
- clean_filename(photo.filename)
+ "[DRY RUN] Would delete %s in iCloud library %s",
+ clean_filename(photo.filename),
+ library_object.zone_id['zoneName']
  )
 
 
@@ -911,7 +914,7 @@ def should_break(counter: Counter) -> bool:
 
  def delete_cmd() -> None:
  delete_local = delete_photo_dry_run if dry_run else delete_photo
- delete_local(logger, icloud, item)
+ delete_local(logger, icloud.photos, library_object, item)
 
  retrier(delete_cmd, error_handler)
 

src/icloudpd/download.py

@@ -11,6 +11,7 @@
 import pyicloud_ipd # pylint: disable=redefined-builtin
 from pyicloud_ipd.exceptions import PyiCloudAPIResponseException
 from pyicloud_ipd.services.photos import PhotoAsset
+from pyicloud_ipd.base import PyiCloudService
 
 # Import the constants object so that we can mock WAIT_SECONDS in tests
 from icloudpd import constants
@@ -100,7 +101,7 @@ def download_response_to_path_dry_run(
 def download_media(
  logger: logging.Logger,
  dry_run: bool,
- icloud: pyicloud_ipd.PyiCloudService,
+ icloud: PyiCloudService,
  photo: PhotoAsset,
  download_path: str,
  size: str) -> bool:

src/icloudpd/exif_datetime.py

@@ -7,17 +7,17 @@
 from piexif._exceptions import InvalidImageDataError
 
 
-def get_photo_exif(logger: logging.Logger, path: str) -> (typing.Optional[datetime.datetime]):
+def get_photo_exif(logger: logging.Logger, path: str) -> (typing.Optional[str]):
  """Get EXIF date for a photo, return nothing if there is an error"""
  try:
  exif_dict: piexif.ExifIFD = piexif.load(path)
- return typing.cast(typing.Optional[datetime.datetime], exif_dict.get("Exif").get(36867))
+ return typing.cast(typing.Optional[str], exif_dict.get("Exif").get(36867))
  except (ValueError, InvalidImageDataError):
  logger.debug("Error fetching EXIF data for %s", path)
  return None
 
 
-def set_photo_exif(logger: logging.Logger, path: str, date: datetime.datetime) -> None:
+def set_photo_exif(logger: logging.Logger, path: str, date: str) -> None:
  """Set EXIF date on a photo, do nothing if there is an error"""
  try:
  exif_dict = piexif.load(path)

src/pyicloud_ipd/base.py

@@ -1,5 +1,5 @@
 import sys
-from typing import Any, Dict, NoReturn, Optional, Sequence
+from typing import Any, Dict, Iterable, NoReturn, Optional, Sequence
 from typing_extensions import override
 import typing
 from uuid import uuid1
@@ -20,15 +20,13 @@
  PyiCloud2SARequiredException,
  PyiCloudServiceNotActivatedException,
 )
-from pyicloud_ipd.services import (
- FindMyiPhoneServiceManager,
- CalendarService,
- UbiquityService,
- ContactsService,
- RemindersService,
- PhotosService,
- AccountService,
-)
+from pyicloud_ipd.services.findmyiphone import AppleDevice, FindMyiPhoneServiceManager
+from pyicloud_ipd.services.calendar import CalendarService
+from pyicloud_ipd.services.ubiquity import UbiquityService
+from pyicloud_ipd.services.contacts import ContactsService
+from pyicloud_ipd.services.reminders import RemindersService
+from pyicloud_ipd.services.photos import PhotosService
+from pyicloud_ipd.services.account import AccountService
 from pyicloud_ipd.utils import get_password_from_keyring
 
 
@@ -68,12 +66,14 @@ def __init__(self, service: Any):
  super().__init__()
 
  @override
- def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
+ # type: ignore 
+ # pylint: disable=arguments-differ 
+ def request(self, method: str, url, **kwargs): 
 
  # Charge logging to the right service endpoint
  callee = inspect.stack()[2]
  module = inspect.getmodule(callee[0])
- request_logger = logging.getLogger(module.__name__).getChild("http")
+ request_logger = logging.getLogger(module.__name__).getChild("http") #type: ignore[union-attr]
  if self.service.password_filter not in request_logger.filters:
  request_logger.addFilter(self.service.password_filter)
 
@@ -101,7 +101,7 @@ def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
  LOGGER.debug("Saved session data to file")
 
  # Save cookies to file
- self.cookies.save(ignore_discard=True, ignore_expires=True)
+ self.cookies.save(ignore_discard=True, ignore_expires=True) # type: ignore[attr-defined]
  LOGGER.debug("Cookies saved to %s", self.service.cookiejar_path)
 
  if not response.ok and (
@@ -144,7 +144,7 @@ def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
  if self.service.session_data.get("apple_rscd") == "401":
  code: Optional[str] = "401"
  reason: Optional[str] = "Invalid username/password combination."
- self._raise_error(code, reason)
+ self._raise_error(code or "Unknown", reason or "Unknown")
 
  return response
 
@@ -158,14 +158,15 @@ def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
 
  if isinstance(data, dict):
  if data.get("hasError"):
- errors = data.get("service_errors")
+ errors: Optional[Sequence[Dict[str, Any]]] = typing.cast(Optional[Sequence[Dict[str, Any]]], data.get("service_errors"))
  # service_errors returns a list of dict
  # dict includes the keys: code, title, message, supressDismissal
  # Assuming a single error for now
  # May need to revisit to capture and handle multiple errors
- code = errors[0].get("code")
- reason = errors[0].get("message")
- self._raise_error(code, reason)
+ if errors:
+ code = errors[0].get("code")
+ reason = errors[0].get("message")
+ self._raise_error(code or "Unknown", reason or "Unknown")
  elif not data.get("success"):
  reason = data.get("errorMessage")
  reason = reason or data.get("reason")
@@ -182,7 +183,7 @@ def request(self, method, url, **kwargs): # pylint: disable=arguments-differ
  code = data.get("error")
 
  if reason:
- self._raise_error(code, reason)
+ self._raise_error(code or "Unknown", reason)
 
  return response
 
@@ -601,14 +602,14 @@ def _get_webservice_url(self, ws_key: str) -> str:
  return typing.cast(str, self._webservices[ws_key]["url"])
 
  @property
- def devices(self): # type: ignore
+ def devices(self) -> Sequence[AppleDevice]: 
  """ Return all devices."""
  service_root = self._get_webservice_url("findme")
- return FindMyiPhoneServiceManager( # type: ignore
+ return typing.cast(Sequence[AppleDevice], FindMyiPhoneServiceManager(
  service_root,
  self.session,
  self.params
- )
+ ))
 
  @property
  def account(self): # type: ignore

src/pyicloud_ipd/cmdline.py

@@ -1,24 +1,24 @@
 #! /usr/bin/env python
-# mypy: ignore-errors
 """
 A Command Line Wrapper to allow easy use of pyicloud for
 command line scripts, and related.
 """
 import argparse
 import pickle
 import sys
-from typing import NoReturn
+from typing import NoReturn, Optional, Sequence
 
 from click import confirm
 
-from pyicloud_ipd import PyiCloudService
+from pyicloud_ipd.base import PyiCloudService
 from pyicloud_ipd.exceptions import PyiCloudFailedLoginException
+from pyicloud_ipd.services.findmyiphone import AppleDevice
 from . import utils
 
 DEVICE_ERROR = "Please use the --device switch to indicate which device to use."
 
 
-def create_pickled_data(idevice, filename):
+def create_pickled_data(idevice: AppleDevice, filename: str) -> None:
  """
  This helper will output the idevice to a pickled file named
  after the passed filename.
@@ -30,7 +30,7 @@ def create_pickled_data(idevice, filename):
  pickle.dump(idevice.content, pickle_file, protocol=pickle.HIGHEST_PROTOCOL)
 
 
-def main(args=None) -> NoReturn:
+def main(args:Optional[Sequence[str]]=None) -> NoReturn:
  """Main commandline entrypoint."""
  if args is None:
  args = sys.argv[1:]
@@ -247,8 +247,8 @@ def main(args=None) -> NoReturn:
  )
 
  print("\nWhich device would you like to use?")
- device = int(input("(number) --> "))
- device = devices[device]
+ device_index = int(input("(number) --> "))
+ device = devices[device_index]
  if not api.send_verification_code(device):
  print("Failed to send verification code")
  sys.exit(1)