From 98ab94b371fea73c42da8291e542b571fb84db8e Mon Sep 17 00:00:00 2001 From: jneilliii Date: Thu, 14 Nov 2024 23:35:03 -0500 Subject: [PATCH] 0.1.8rc11 update pybambu/bambu_cloud.py from upstream HA project --- .../printer/pybambu/bambu_cloud.py | 130 ++++++++++++++---- setup.py | 2 +- 2 files changed, 102 insertions(+), 30 deletions(-) diff --git a/octoprint_bambu_printer/printer/pybambu/bambu_cloud.py b/octoprint_bambu_printer/printer/pybambu/bambu_cloud.py index 4d8b4c3..797d44b 100644 --- a/octoprint_bambu_printer/printer/pybambu/bambu_cloud.py +++ b/octoprint_bambu_printer/printer/pybambu/bambu_cloud.py @@ -3,7 +3,11 @@ import base64 import json -from curl_cffi import requests +curl_available = True +try: + from curl_cffi import requests as curl_requests +except ImportError: + curl_available = False from dataclasses import dataclass @@ -33,6 +37,9 @@ def _get_headers_with_auth_token(self) -> dict: def _get_authentication_token(self) -> dict: LOGGER.debug("Getting accessToken from Bambu Cloud") + if not curl_available: + LOGGER.debug(f"Curl library is unavailable.") + return 'curlUnavailable' # First we need to find out how Bambu wants us to login. data = { @@ -41,7 +48,14 @@ def _get_authentication_token(self) -> dict: "apiError": "" } - response = requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) + response = curl_requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) + + # Check specifically for cloudflare block + if response.status_code == 403: + if 'cloudflare' in response.text: + LOGGER.error('CloudFlare blocked connection attempt') + return 'cloudFlare' + if response.status_code >= 400: LOGGER.error(f"Login attempt failed with error code: {response.status_code}") LOGGER.debug(f"Response: '{response.text}'") @@ -80,7 +94,7 @@ def _get_email_verification_code(self): } LOGGER.debug("Requesting verification code") - response = requests.post(get_Url(BambuUrl.EMAIL_CODE, self._region), json=data, impersonate=IMPERSONATE_BROWSER) + response = curl_requests.post(get_Url(BambuUrl.EMAIL_CODE, self._region), json=data, impersonate=IMPERSONATE_BROWSER) if response.status_code == 200: LOGGER.debug("Verification code requested successfully.") @@ -96,7 +110,7 @@ def _get_authentication_token_with_verification_code(self, code) -> dict: "code": code } - response = requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) + response = curl_requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) LOGGER.debug(f"Response: {response.status_code}") if response.status_code == 200: @@ -128,7 +142,7 @@ def _get_authentication_token_with_2fa_code(self, code: str) -> dict: "tfaCode": code } - response = requests.post(get_Url(BambuUrl.TFA_LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) + response = curl_requests.post(get_Url(BambuUrl.TFA_LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) LOGGER.debug(f"Response: {response.status_code}") if response.status_code == 200: @@ -209,13 +223,11 @@ def login(self, region: str, email: str, password: str) -> str: self._password = password result = self._get_authentication_token() - if result == 'verifyCode': - return result - elif result == 'tfa': - return result - elif result is None: + if result is None: LOGGER.error("Unable to authenticate.") return None + elif len(result) < 20: + return result else: self._auth_token = result self._username = self._get_username_from_authentication_token() @@ -223,7 +235,7 @@ def login(self, region: str, email: str, password: str) -> str: def login_with_verification_code(self, code: str): result = self._get_authentication_token_with_verification_code(code) - if result == 'codeExpired' or result == 'codeIncorrect': + if len(result) < 20: return result self._auth_token = result self._username = self._get_username_from_authentication_token() @@ -231,16 +243,29 @@ def login_with_verification_code(self, code: str): def login_with_2fa_code(self, code: str): result = self._get_authentication_token_with_2fa_code(code) + if len(result) < 20: + return result self._auth_token = result self._username = self._get_username_from_authentication_token() return 'success' def get_device_list(self) -> dict: LOGGER.debug("Getting device list from Bambu Cloud") - response = requests.get(get_Url(BambuUrl.BIND, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) + if not curl_available: + LOGGER.debug(f"Curl library is unavailable.") + raise None + + response = curl_requests.get(get_Url(BambuUrl.BIND, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) + if response.status_code == 403: + if 'cloudflare' in response.text: + LOGGER.error('CloudFlare blocked connection attempt') + raise ValueError(response.status_code) + if response.status_code >= 400: LOGGER.debug(f"Received error: {response.status_code}") + LOGGER.error(f"Received error: '{response.text}'") raise ValueError(response.status_code) + return response.json()['devices'] # The slicer settings are of the following form: @@ -311,13 +336,23 @@ def get_device_list(self) -> dict: # } def get_slicer_settings(self) -> dict: - LOGGER.debug("Getting slicer settings from Bambu Cloud") - response = requests.get(get_Url(BambuUrl.SLICER_SETTINGS, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) - if response.status_code >= 400: - LOGGER.error(f"Slicer settings load failed: {response.status_code}") - LOGGER.error(f"Slicer settings load failed: '{response.text}'") - return None - return response.json() + LOGGER.debug("DISABLED: Getting slicer settings from Bambu Cloud") + # Disabled for now since it may be contributing to cloudflare detection speed. + # + # if curl_available: + # response = curl_requests.get(get_Url(BambuUrl.SLICER_SETTINGS, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) + # if response.status_code == 403: + # if 'cloudflare' in response.text: + # LOGGER.error(f"Cloudflare blocked slicer settings lookup.") + # return None + + # if response.status_code >= 400: + # LOGGER.error(f"Slicer settings load failed: {response.status_code}") + # LOGGER.error(f"Slicer settings load failed: '{response.text}'") + # return None + + # return response.json() + return None # The task list is of the following form with a 'hits' array with typical 20 entries. # @@ -362,24 +397,46 @@ def get_slicer_settings(self) -> dict: # }, def get_tasklist(self) -> dict: + LOGGER.debug("Getting full task list from Bambu Cloud") + if not curl_available: + LOGGER.debug(f"Curl library is unavailable.") + raise None + url = get_Url(BambuUrl.TASKS, self._region) - response = requests.get(url, headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) + response = curl_requests.get(url, headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) + if response.status_code == 403: + if 'cloudflare' in response.text: + LOGGER.error('CloudFlare blocked connection attempt') + return None + + # Check specifically for cloudflare block + if response.status_code == 403: + if 'cloudflare' in response.text: + LOGGER.error('CloudFlare blocked connection attempt') + return None + if response.status_code >= 400: LOGGER.debug(f"Received error: {response.status_code}") LOGGER.debug(f"Received error: '{response.text}'") - raise ValueError(response.status_code) + raise None + return response.json() - + def get_latest_task_for_printer(self, deviceId: str) -> dict: - LOGGER.debug(f"Getting latest task from Bambu Cloud") - data = self.get_tasklist_for_printer(deviceId) - if len(data) != 0: - return data[0] - LOGGER.debug("No tasks found for printer") + LOGGER.debug(f"Getting latest task for printer from Bambu Cloud") + try: + data = self.get_tasklist_for_printer(deviceId) + if len(data) != 0: + return data[0] + LOGGER.debug("No tasks found for printer") + except: + LOGGER.debug("Unable to make call") + return None + return None def get_tasklist_for_printer(self, deviceId: str) -> dict: - LOGGER.debug(f"Getting task list from Bambu Cloud") + LOGGER.debug(f"Getting full task list for printer from Bambu Cloud") tasks = [] data = self.get_tasklist() for task in data['hits']: @@ -394,10 +451,21 @@ def get_device_type_from_device_product_name(self, device_product_name: str): def download(self, url: str) -> bytearray: LOGGER.debug(f"Downloading cover image: {url}") - response = requests.get(url, timeout=10, impersonate=IMPERSONATE_BROWSER) + if not curl_available: + LOGGER.debug(f"Curl library is unavailable.") + return None + + response = curl_requests.get(url, timeout=10, impersonate=IMPERSONATE_BROWSER) + if response.status_code == 403: + if 'cloudflare' in response.text: + LOGGER.error('CloudFlare blocked connection attempt') + raise ValueError(response.status_code) + if response.status_code >= 400: LOGGER.debug(f"Received error: {response.status_code}") + LOGGER.debug(f"Received error: {response.text}") raise ValueError(response.status_code) + return response.content @property @@ -408,6 +476,10 @@ def username(self): def auth_token(self): return self._auth_token + @property + def bambu_connected(self) -> bool: + return self._auth_token != "" and self._auth_token != None + @property def cloud_mqtt_host(self): return "cn.mqtt.bambulab.com" if self._region == "China" else "us.mqtt.bambulab.com" diff --git a/setup.py b/setup.py index 114492c..c9d085d 100644 --- a/setup.py +++ b/setup.py @@ -14,7 +14,7 @@ plugin_name = "OctoPrint-BambuPrinter" # The plugin's version. Can be overwritten within OctoPrint's internal data via __plugin_version__ in the plugin module -plugin_version = "0.1.8rc10" +plugin_version = "0.1.8rc11" # The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin # module