diff --git a/.gitignore b/.gitignore index 277dfc2..f2715b5 100644 --- a/.gitignore +++ b/.gitignore @@ -23,7 +23,7 @@ ENV/ *.log # Local configuration that might contain sensitive information -network_configuration.json +#network_configuration.json # Backup files *.backup @@ -32,4 +32,4 @@ network_configuration.json current.json deprecated.json TasmotaDevices.json -*.json.backup \ No newline at end of file +*.json.backup diff --git a/TasmotaManager.py b/TasmotaManager.py index 2c7b2e1..f7a0ead 100755 --- a/TasmotaManager.py +++ b/TasmotaManager.py @@ -13,17 +13,78 @@ import argparse # Disable SSL warnings requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) +class AuthenticationError(Exception): + """Raised when UniFi returns an authentication/authorization error (401/403).""" + pass + +class UniFiDataError(Exception): + """Raised when UniFi returns non-JSON or unexpected data across all fallbacks.""" + pass + class UnifiClient: - def __init__(self, base_url, username, password, site_id, verify_ssl=True): + def __init__(self, base_url, username=None, password=None, site_id='default', verify_ssl=True, token=None): self.base_url = base_url.rstrip('/') self.username = username self.password = password self.site_id = site_id + self.token = token self.session = requests.Session() self.session.verify = verify_ssl # Initialize cookie jar self.session.cookies.clear() + + # Set common headers for UniFi requests + self.session.headers.update({ + 'Accept': 'application/json', + 'User-Agent': 'TasmotaManager/1.0', + 'Referer': f'{self.base_url}/', + 'Origin': self.base_url + }) + + # If using a Personal Access Token, set Authorization header + if self.token: + self.session.headers.update({ + 'Authorization': f'Bearer {self.token}' + }) + + def _request_json(self, url: str, method: str = 'GET', timeout: tuple = (5, 10)): + """Perform an HTTP request and return parsed JSON or None. + Raises AuthenticationError on 401/403. Never logs sensitive tokens. + """ + logger = logging.getLogger(__name__) + try: + resp = self.session.request(method, url, timeout=timeout) + except requests.exceptions.RequestException as e: + # Network-level error + return None + status = getattr(resp, 'status_code', None) + # Auth errors + if status in (401, 403): + raise AuthenticationError(f"Auth error {status} for {url}") + # Other HTTP errors + if status and status >= 400: + return None + ctype = (resp.headers.get('Content-Type') or '').lower() + text = None + # If not JSON content-type, try to detect empty body, otherwise return None + if 'json' not in ctype: + try: + text = resp.text + except Exception: + text = '' + if not text: + return None + # Non-JSON body present (likely HTML); return None + logger.debug(f"Non-JSON response from {url} (status {status}, content-type {ctype}).") + return None + # Parse JSON safely + try: + return resp.json() + except ValueError: + # Empty or invalid JSON + logger.debug(f"Failed to parse JSON from {url} (status {status}, content-type {ctype}).") + return None def _login(self) -> requests.Response: # Changed return type annotation """Authenticate with the UniFi Controller.""" @@ -48,49 +109,61 @@ class UnifiClient: self.session.headers['X-CSRF-Token'] = response.headers['X-CSRF-Token'] return response # Return the response object except requests.exceptions.RequestException as e: - if hasattr(e, 'response') and e.response.status_code == 401: + if hasattr(e, 'response') and e.response is not None and e.response.status_code == 401: raise Exception("Authentication failed. Please verify your username and password.") from e raise def get_clients(self) -> list: """Get all clients from the UniFi Controller.""" - # Try the newer API endpoint first - url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/sta" + logger = logging.getLogger(__name__) + # 1) Preferred classic endpoint + url1 = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/sta" + data = self._request_json(url1) + if isinstance(data, dict): + return data.get('data', []) + if isinstance(data, list): + return data + # 2) v2 endpoint (often returns a list directly) + url2 = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/clients" try: - response = self.session.get(url) - response.raise_for_status() - return response.json().get('data', []) - except requests.exceptions.RequestException as e: - # If the newer endpoint fails, try the legacy endpoint - url = f"{self.base_url}/api/s/{self.site_id}/stat/sta" - try: - response = self.session.get(url) - response.raise_for_status() - return response.json().get('data', []) - except requests.exceptions.RequestException as e: - # If both fail, try the v2 API endpoint - url = f"{self.base_url}/v2/api/site/{self.site_id}/clients" - response = self.session.get(url) - response.raise_for_status() - return response.json().get('data', []) + data = self._request_json(url2) + except AuthenticationError: + # Bubble up auth errors + raise + if isinstance(data, list): + return data + if isinstance(data, dict): + return data.get('data', []) + # 3) Legacy classic endpoint (no /proxy) + url3 = f"{self.base_url}/api/s/{self.site_id}/stat/sta" + data = self._request_json(url3) + if isinstance(data, dict): + return data.get('data', []) + if isinstance(data, list): + return data + raise UniFiDataError("UniFi returned no parsable JSON for clients across all endpoints") def get_devices(self) -> list: """Get UniFi network devices (e.g., Access Points) from the Controller.""" - # Try the newer API endpoint first - url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device" + logger = logging.getLogger(__name__) + # 1) Preferred classic endpoint + url1 = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device" try: - response = self.session.get(url) - response.raise_for_status() - return response.json().get('data', []) - except requests.exceptions.RequestException: - # Try legacy endpoint - url = f"{self.base_url}/api/s/{self.site_id}/stat/device" - try: - response = self.session.get(url) - response.raise_for_status() - return response.json().get('data', []) - except requests.exceptions.RequestException: - return [] + data = self._request_json(url1) + except AuthenticationError: + raise + if isinstance(data, dict): + return data.get('data', []) + if isinstance(data, list): + return data + # 2) Legacy classic endpoint + url2 = f"{self.base_url}/api/s/{self.site_id}/stat/device" + data = self._request_json(url2) + if isinstance(data, dict): + return data.get('data', []) + if isinstance(data, list): + return data + raise UniFiDataError("UniFi returned no parsable JSON for devices across all endpoints") class TasmotaDiscovery: def __init__(self, debug: bool = False): @@ -130,33 +203,58 @@ class TasmotaDiscovery: sys.exit(1) def setup_unifi_client(self): - """Set up the UniFi client with better error handling""" + """Set up the UniFi client with better error handling and token support""" self.logger.debug("Setting up UniFi client") if not self.config or 'unifi' not in self.config: raise ValueError("Missing UniFi configuration") unifi_config = self.config['unifi'] - required_fields = ['host', 'username', 'password', 'site'] - missing_fields = [field for field in required_fields if field not in unifi_config] + host = unifi_config.get('host') + site = unifi_config.get('site', 'default') + token = unifi_config.get('token') + username = unifi_config.get('username') + password = unifi_config.get('password') - if missing_fields: - raise ValueError(f"Missing required UniFi configuration fields: {', '.join(missing_fields)}") + missing_core = [] + if not host: + missing_core.append('host') + if not site: + missing_core.append('site') + if missing_core: + raise ValueError(f"Missing required UniFi configuration fields: {', '.join(missing_core)}") + + if not token and (not username or not password): + raise ValueError("Provide either 'unifi.token' OR both 'unifi.username' and 'unifi.password'") try: - self.logger.debug(f"Connecting to UniFi Controller at {unifi_config['host']}") - self.unifi_client = UnifiClient( - base_url=unifi_config['host'], - username=unifi_config['username'], - password=unifi_config['password'], - site_id=unifi_config['site'], - verify_ssl=False # Add this if using self-signed certificates - ) - - # Test the connection by making a simple request - response = self.unifi_client._login() - if not response: - raise ConnectionError(f"Failed to connect to UniFi controller: No response") + self.logger.debug(f"Connecting to UniFi Controller at {host} using {'token' if token else 'username/password'}") + # Instantiate client + if token: + self.unifi_client = UnifiClient( + base_url=host, + site_id=site, + verify_ssl=False, + token=token + ) + # Test by calling a lightweight endpoint + try: + # Attempt to fetch devices (requires auth). Empty list is acceptable. + _ = self.unifi_client.get_devices() + except Exception as test_err: + raise ConnectionError(f"Token authentication failed: {test_err}") + else: + self.unifi_client = UnifiClient( + base_url=host, + username=username, + password=password, + site_id=site, + verify_ssl=False + ) + # Test the connection by logging in + response = self.unifi_client._login() + if not response: + raise ConnectionError("Failed to connect to UniFi controller: No response") self.logger.debug("UniFi client setup successful") @@ -623,7 +721,7 @@ class TasmotaDiscovery: return devices except Exception as e: self.logger.error(f"Error getting devices from UniFi controller: {e}") - return [] + raise def save_tasmota_config(self, devices: list) -> None: """Save Tasmota device information to a JSON file with device tracking.""" @@ -2830,21 +2928,18 @@ def main(): print("- Device list saved to: current.json") print("- Detailed information saved to: TasmotaDevices.json") - except ConnectionError as e: - print(f"Connection Error: {str(e)}") - print("\nTrying to proceed with existing current.json...") - try: - discovery.get_device_details(use_current_json=True) - print("\nSuccessfully retrieved device details from existing current.json") - except Exception as inner_e: - print(f"Error processing existing devices: {str(inner_e)}") - return 1 - except Exception as e: + except (AuthenticationError, UniFiDataError, ConnectionError) as e: print(f"Error: {str(e)}") if args.debug: import traceback traceback.print_exc() return 1 + except Exception as e: + print(f"Unexpected error: {str(e)}") + if args.debug: + import traceback + traceback.print_exc() + return 1 return 0 diff --git a/network_configuration.json b/network_configuration.json index 7bf6c46..1e0809f 100644 --- a/network_configuration.json +++ b/network_configuration.json @@ -3,6 +3,7 @@ "host": "https://192.168.6.1", "username": "Tasmota", "password": "TasmotaManager12!@", + "site": "default", "network_filter": { "NoT_network": { @@ -33,48 +34,48 @@ "device_list": { "TreatLife_SW_SS01S": { "template": "{\"NAME\":\"TL SS01S Swtch\",\"GPIO\":[0,0,0,0,52,158,0,0,21,17,0,0,0],\"FLAG\":0,\"BASE\":18}", - - "console_set": "Default" + "console_set": "Traditional" }, "TreatLife_SW_SS02S": { "template": "{\"NAME\":\"Treatlife SS02\",\"GPIO\":[0,0,0,0,288,576,0,0,224,32,0,0,0,0],\"FLAG\":0,\"BASE\":18}", - "console_set": "Default" + "console_set": "Traditional" }, "TreatLife_SW_SS02S_Orig": { "template": "{\"NAME\":\"Treatlife SS02\",\"GPIO\":[0,0,0,0,289,0,0,0,224,32,0,0,0,0],\"FLAG\":0,\"BASE\":18}", - "console_set": "Default" + "console_set": "Traditional" }, "TreatLife_DIM_DS02S": { "template": "{\"NAME\":\"DS02S Dimmer\",\"GPIO\":[0,107,0,108,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":54}", - "console_set": "Default" + "console_set": "Traditional" }, "CloudFree_SW1": { "template": "{\"NAME\":\"CloudFree SW1\",\"GPIO\":[0,224,0,32,320,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":1}", - "console_set": "Default" + "console_set": "Traditional" }, "Gosund_WP5_Plug": { "template": "{\"NAME\":\"Gosund-WP5\",\"GPIO\":[0,0,0,0,17,0,0,0,56,57,21,0,0],\"FLAG\":0,\"BASE\":18}", - "console_set": "Default" + "console_set": "Traditional" }, "Gosund_Plug": { "template": "{\"NAME\":\"Gosund-WP5\",\"GPIO\":[0,0,0,0,32,0,0,0,320,321,224,0,0,0],\"FLAG\":0,\"BASE\":18}", - "console_set": "Default" + "console_set": "Traditional" }, "CloudFree_X10S_Plug": { "template": "{\"NAME\":\"Aoycocr X10S\",\"GPIO\":[56,0,57,0,21,134,0,0,131,17,132,0,0],\"FLAG\":0,\"BASE\":45}", - "console_set": "alt" + "console_set": "Traditional" }, "Sonoff_S31_PM_Plug": { "template": "{\"NAME\":\"Sonoff S31\",\"GPIO\":[17,145,0,146,0,0,0,0,21,56,0,0,0],\"FLAG\":0,\"BASE\":41}", - "console_set": "Default" + "console_set": "Traditional" }, - "Sonoff S31": { - "template": "", - "console_set": "Default" + "Sonoff TX Ultimate 1": { + "template": "{\"NAME\":\"Sonoff T5-1C-120\",\"GPIO\":[0,0,7808,0,7840,3872,0,0,0,1376,0,7776,0,0,224,3232,0,480,3200,0,0,0,3840,0,0,0,0,0,0,0,0,0,0,0,0,0],\"FLAG\":0,\"BASE\":1}", + "console_set": "SONOFF_ULTIMATE" } }, "console_set": { - "Default": [ + "Traditional": [ + "", "SwitchRetain Off", "ButtonRetain Off", "PowerRetain On", @@ -85,24 +86,28 @@ "SetOption13 0", "SetOption19 0", "SetOption32 8", + "SetOption40 40", "SetOption53 1", "SetOption73 1", "rule1 on button1#state=10 do power0 toggle endon" ], - "alt": [ + "SONOFF_ULTIMATE": [ "SwitchRetain Off", "ButtonRetain Off", "PowerRetain On", "PowerOnState 3", + "Pixels 32", "SetOption1 0", "SetOption3 1", "SetOption4 1", "SetOption13 0", "SetOption19 0", "SetOption32 8", + "SetOption40 40", + "SetOption53 1", - "SetOption73 1", - "rule1 on button1#state=10 do power0 toggle endon" + "SetOption73 1" + ] } } \ No newline at end of file