LotusGuild/gandalf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

legacy endpoints

Browse Source
This commit is contained in:
Jared Vititoe
2025-02-07 22:44:45 -05:00
parent 2dea9ddc8d
commit 5298349ac7
1 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
app.py
View File

@ -47,7 +47,7 @@ def send_webhook(device, status, diagnostics):
class UnifiAPI: class UnifiAPI:
def __init__(self, config): def __init__(self, config):
self.base_url = config['unifi']['controller'] self.base_url = config['unifi']['controller']
self.site_id = "1647955405" # Fixed site ID self.site_id = "1647955405"
self.session = requests.Session() self.session = requests.Session()
self.session.verify = False self.session.verify = False
self.headers = { self.headers = {
@ -56,24 +56,30 @@ class UnifiAPI:
} }
def get_all_devices(self): def get_all_devices(self):
url = f"{self.base_url}/integration/v1/sites/{self.site_id}/devices" url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device"
response = self.session.get(url, headers=self.headers) response = self.session.get(url, headers=self.headers)
response.raise_for_status() response.raise_for_status()
devices = response.json()['data'] logger.debug(f"Raw device list response: {response.text}")
# Map IP addresses to device IDs return response.json()['data']
self.device_map = {device['ipAddress']: device['id'] for device in devices}
return devices
def get_device_details(self, ip_address): def get_device_details(self, device_id):
device_id = self.device_map.get(ip_address) url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device/{device_id}"
if not device_id:
self.get_all_devices() # Refresh device map
device_id = self.device_map.get(ip_address)
url = f"{self.base_url}/integration/v1/sites/{self.site_id}/devices/{device_id}"
response = self.session.get(url, headers=self.headers) response = self.session.get(url, headers=self.headers)
response.raise_for_status() response.raise_for_status()
return response.json() logger.debug(f"Raw device details response: {response.text}")
data = response.json()['data'][0]
return {
'state': data['state'],
'firmware': {
'version': data['version'],
'updatable': data.get('upgradable', False)
},
'network': {
'ip': data['ip'],
'mac': data['mac']
},
'interfaces': self._parse_interfaces(data.get('port_table', {}))
}
def get_device_diagnostics(self, device): def get_device_diagnostics(self, device):
details = self.get_device_details(device['ip']) details = self.get_device_details(device['ip'])
@ -116,8 +122,7 @@ class UnifiAPI:
'width': f"{radio['channelWidthMHz']}MHz" 'width': f"{radio['channelWidthMHz']}MHz"
} }
return result return result# Monitoring functions
# Monitoring functions
def run_diagnostics(device): def run_diagnostics(device):
try: try:
config = load_config() config = load_config()