diff --git a/README.md b/README.md index abf0d54..b513e6e 100644 --- a/README.md +++ b/README.md @@ -143,4 +143,6 @@ Pull requests are the best way to propose changes to the codebase. # Showing Your Appreciation -If you like this project, please give it a star on [GitHub](https://github.com/thomluther/anker-solix-api) \ No newline at end of file +[!["Buy Me A Coffee"](https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png)](https://www.buymeacoffee.com/thomasluthe) + +If you like this project, please give it a star on [GitHub](https://github.com/thomluther/anker-solix-api) diff --git a/api/__pycache__/__init__.cpython-311.pyc b/api/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..66f0303 Binary files /dev/null and b/api/__pycache__/__init__.cpython-311.pyc differ diff --git a/api/__pycache__/api.cpython-311.pyc b/api/__pycache__/api.cpython-311.pyc new file mode 100644 index 0000000..c634e0f Binary files /dev/null and b/api/__pycache__/api.cpython-311.pyc differ diff --git a/api/__pycache__/errors.cpython-311.pyc b/api/__pycache__/errors.cpython-311.pyc new file mode 100644 index 0000000..64e11de Binary files /dev/null and b/api/__pycache__/errors.cpython-311.pyc differ diff --git a/api/api.py b/api/api.py index be9d44f..f8b6218 100644 --- a/api/api.py +++ b/api/api.py @@ -15,7 +15,6 @@ import logging import os import sys import time -from typing import Optional from aiohttp import ClientSession from aiohttp.client_exceptions import ClientError @@ -29,7 +28,11 @@ from . import errors _LOGGER: logging.Logger = logging.getLogger(__name__) """Default definitions required for the Anker Power/Solix Cloud API""" -_API_BASE: str = "https://ankerpower-api-eu.anker.com" +# API servers per region. Country assignment not clear, defaulting to EU server +_API_SERVERS = { + "eu": "https://ankerpower-api-eu.anker.com", + "com": "https://ankerpower-api.anker.com", +} _API_LOGIN = "passport/login" _API_HEADERS = { "Content-Type": "application/json", @@ -37,6 +40,10 @@ _API_HEADERS = { "App-Name": "anker_power", "Os-Type": "android", } +_API_COUNTRIES = { + "com": ["US", "CN"], + "eu": ["DE", "IT", "FR", "ES"], +} # TODO(2): Expand list once more ID assignments are known """Following are the Anker Power/Solix Cloud API endpoints known so far""" _API_ENDPOINTS = { @@ -83,6 +90,8 @@ _API_ENDPOINTS = { 'power_service/v1/app/compatible/save_ota_complete_status', 'power_service/v1/app/compatible/check_third_sn', 'power_service/v1/app/compatible/save_compatible_solar', + 'power_service/v1/app/compatible/get_confirm_permissions', + 'power_service/v1/app/compatible/confirm_permissions_settings', 'power_service/v1/app/after_sale/check_popup', 'power_service/v1/app/after_sale/check_sn', 'power_service/v1/app/after_sale/mark_sn', @@ -95,8 +104,6 @@ _API_ENDPOINTS = { 'power_service/v1/app/check_upgrade_record', 'power_service/v1/app/get_upgrade_record', 'power_service/v1/app/get_phonecode_list', - 'power_service/v1/app/compatible/get_confirm_permissions', - 'power_service/v1/app/compatible/confirm_permissions_settings', 'power_service/v1/message_not_disturb', 'power_service/v1/get_message_not_disturb', 'power_service/v1/read_message', @@ -105,6 +112,7 @@ _API_ENDPOINTS = { 'power_service/v1/product_categories', 'power_service/v1/product_accessories', + Structure of the JSON response for an API Login Request: An unexpired token_id must be used for API request, along with the gtoken which is an MD5 hash of the returned(encrypted) user_id. The combination of the provided token and MD5 hashed user_id authenticate the client to the server. @@ -153,10 +161,16 @@ class AnkerSolixApi: logger=None, ) -> None: """Initialize.""" - self._api_base: str = _API_BASE + self._countryId: str = countryId.upper() + self._api_base: str | None = None + for region, countries in _API_COUNTRIES.items(): + if self._countryId in countries: + self._api_base = _API_SERVERS.get(region) + # default to EU server + if not self._api_base: + self._api_base = _API_SERVERS.get("eu") self._email: str = email self._password: str = password - self._countryId: str = countryId.upper() self._session: ClientSession = websession self._loggedIn: bool = False self._testdir: str = "test" @@ -180,13 +194,14 @@ class AnkerSolixApi: self._timezone: str = ( self._getTimezoneGMTString() ) # Timezone format: 'GMT+01:00' - self._gtoken: Optional[str] = None - self._token: Optional[str] = None - self._token_expiration: Optional[datetime] = None - self._login_response: Optional[LOGIN_RESPONSE] = {} + self._gtoken: str | None = None + self._token: str | None = None + self._token_expiration: datetime | None = None + self._login_response: LOGIN_RESPONSE = {} # Define Encryption for password, using ECDH assymetric key exchange for shared secret calculation, which must be used to encrypt the password using AES-256-CBC with seed of 16 - # uncompressed public key from Anker server in the format 04 [32 byte x vlaue] [32 byte y value] + # uncompressed public key from EU Anker server in the format 04 [32 byte x vlaue] [32 byte y value] + # TODO(2): COM Anker server public key usage to be validated self._api_public_key_hex = "04c5c00c4f8d1197cc7c3167c52bf7acb054d722f0ef08dcd7e0883236e0d72a3868d9750cb47fa4619248f3d83f0f662671dadc6e2d31c2f41db0161651c7c076" self._curve = ( ec.SECP256R1() @@ -430,8 +445,8 @@ class AnkerSolixApi: method: str, endpoint: str, *, - headers: Optional[dict] = None, - json: Optional[dict] = None, # lint W0621 + headers: dict | None = None, + json: dict | None = None, ) -> dict: """Handle all requests to the API. This is also called recursively by login requests if necessary.""" if not headers: @@ -495,7 +510,7 @@ class AnkerSolixApi: # Unauthorized or forbidden request if self._retry_attempt: raise errors.AuthorizationError( - "Login failed for user %s" % self._email + f"Login failed for user {self._email}" ) from err self._logger.warning("Login failed, retrying authentication...") if await self.async_authenticate(restart=True): @@ -504,7 +519,7 @@ class AnkerSolixApi: ) self._logger.error("Login failed for user %s", self._email) raise errors.AuthorizationError( - "Login failed for user %s" % self._email + f"Login failed for user {self._email}" ) from err raise ClientError( f"There was an error while requesting {endpoint}: {err}" @@ -525,7 +540,7 @@ class AnkerSolixApi: ) self._logger.error("Login failed for user %s", self._email) raise errors.AuthorizationError( - "Login failed for user %s" % self._email + f"Login failed for user {self._email}" ) from err except errors.AnkerSolixError as err: # Other Exception from API self._logger.error("ANKER API ERROR: %s", err) @@ -819,7 +834,7 @@ class AnkerSolixApi: if toFile: resp = self._saveToFile( os.path.join(self._testdir, f"set_power_cutoff_{deviceSn}.json"), - json=data, + data=data, ) else: resp = await self.request("post", _API_ENDPOINTS["set_cutoff"], json=data) @@ -904,7 +919,7 @@ class AnkerSolixApi: } if toFile: resp = self._saveToFile( - os.path.join(self._testdir, f"set_device_parm_{siteId}.json"), json=data + os.path.join(self._testdir, f"set_device_parm_{siteId}.json"), data=data ) else: resp = await self.request( diff --git a/export_system.py b/export_system.py index a27a7a5..fa20e84 100644 --- a/export_system.py +++ b/export_system.py @@ -19,7 +19,8 @@ import sys import time from aiohttp import ClientSession -from api import api +from aiohttp.client_exceptions import ClientError +from api import api, errors _LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER.addHandler(logging.StreamHandler(sys.stdout)) @@ -112,16 +113,16 @@ def export(filename: str, d: dict = None) -> None: d = {} time.sleep(1) # central delay between multiple requests if len(d) == 0: - CONSOLE.info(f"WARNING: File {filename} not saved because JSON is empty") + CONSOLE.info("WARNING: File %s not saved because JSON is empty", filename) return elif RANDOMIZE: d = check_keys(d) try: with open(filename, "w", encoding="utf-8") as file: json.dump(d, file, indent=2) - CONSOLE.info(f"Saved JSON to file {filename}") - except Exception as err: - CONSOLE.error(f"ERROR: Failed to save JSON to file {filename}: {err}") + CONSOLE.info("Saved JSON to file %s", filename) + except OSError as err: + CONSOLE.error("ERROR: Failed to save JSON to file %s: %s", filename, err) return @@ -169,7 +170,7 @@ async def main() -> bool: # noqa: C901 # first update sites in API object CONSOLE.info("\nQuerying site information...") await myapi.update_sites() - CONSOLE.info(f"Sites: {len(myapi.sites)}, Devices: {len(myapi.devices)}") + CONSOLE.info("Sites: %s, Devices: %s", len(myapi.sites), len(myapi.devices)) _LOGGER.debug(json.dumps(myapi.devices, indent=2)) # Query API using direct endpoints to save full response of each query in json files @@ -212,7 +213,7 @@ async def main() -> bool: # noqa: C901 ), ) # shows only owner devices for siteId, site in myapi.sites.items(): - CONSOLE.info(f"\nExporting site specific data for site {siteId}...") + CONSOLE.info("\nExporting site specific data for site %s...", siteId) CONSOLE.info("Exporting scene info...") export( os.path.join(folder, f"scene_{randomize(siteId,'site_id')}.json"), @@ -247,7 +248,7 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId}, ), ) - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info("Exporting wifi list...") @@ -262,7 +263,7 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info("Exporting site price...") @@ -277,14 +278,14 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info("Exporting device parameter settings...") try: export( os.path.join( - folder, "device_parm_{randomize(siteId,'site_id')}.json" + folder, f"device_parm_{randomize(siteId,'site_id')}.json" ), await myapi.request( "post", @@ -292,12 +293,14 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId, "param_type": "4"}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") for sn, device in myapi.devices.items(): CONSOLE.info( - f"\nExporting device specific data for device {device.get('name','')} SN {sn}..." + "\nExporting device specific data for device %s SN %s...", + device.get("name", ""), + sn, ) siteId = device.get("site_id", "") admin = site.get("is_admin") @@ -313,7 +316,7 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId, "device_sn": sn}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info("Exporting fittings...") @@ -328,7 +331,7 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId, "device_sn": sn}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info("Exporting load...") @@ -341,16 +344,17 @@ async def main() -> bool: # noqa: C901 json={"site_id": siteId, "device_sn": sn}, ), ) # works only for site owners - except Exception: + except (ClientError,errors.AnkerSolixError): if not admin: CONSOLE.warning("Query requires account of site owner!") CONSOLE.info( - f"\nCompleted export of Anker Solix system data for user {USER}" + "\nCompleted export of Anker Solix system data for user %s", USER ) if RANDOMIZE: CONSOLE.info( - f"Folder {os.path.abspath(folder)} contains the randomized JSON files. Pls check and update fields that may contain unrecognized personalized data." + "Folder %s contains the randomized JSON files. Pls check and update fields that may contain unrecognized personalized data.", + os.path.abspath(folder), ) CONSOLE.info( "Following trace or site IDs, SNs and MAC addresses have been randomized in files (from -> to):" @@ -358,12 +362,12 @@ async def main() -> bool: # noqa: C901 CONSOLE.info(json.dumps(RANDOMDATA, indent=2)) else: CONSOLE.info( - f"Folder {os.path.abspath(folder)} contains the JSON files." + "Folder %s contains the JSON files.", os.path.abspath(folder) ) return True - except Exception as err: - CONSOLE.info(f"{type(err)}: {err}") + except (ClientError,errors.AnkerSolixError) as err: + CONSOLE.info("%s: %s", type(err), err) return False @@ -375,4 +379,4 @@ if __name__ == "__main__": except KeyboardInterrupt: CONSOLE.info("Aborted!") except Exception as exception: - CONSOLE.info(f"{type(exception)}: {exception}") + CONSOLE.info("%s: %s", type(exception), exception)