robin
/
anker-solix-api-PR
1
0
Fork 0
Code Pull Requests Activity

export_system code cleanup and fixes 

api preparation for globalization
This commit is contained in:
Thomas Luther 2024-02-09 13:36:49 +00:00
parent 96324d3cbb
commit 0a542013f9
6 changed files with 62 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -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)
[!["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)

BIN
api/__pycache__/__init__.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
api/__pycache__/api.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
api/__pycache__/errors.cpython-311.pyc Normal file
View File

Binary file not shown.

51
api/api.py
View File

@ -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(

48
export_system.py
View File

@ -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)