144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
import logging
|
|
import json
|
|
import platform
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
from flask import Flask, render_template, jsonify
|
|
import requests
|
|
from urllib3.exceptions import InsecureRequestWarning
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger(__name__)
|
|
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
|
|
app = Flask(__name__)
|
|
device_status = {}
|
|
|
|
def load_config():
|
|
with open('config.json') as f:
|
|
return json.load(f)
|
|
|
|
class UnifiAPI:
|
|
def __init__(self, config):
|
|
self.base_url = config['unifi']['controller']
|
|
self.session = requests.Session()
|
|
self.session.verify = False
|
|
self.headers = {
|
|
'X-API-KEY': config['unifi']['api_key'],
|
|
'Accept': 'application/json'
|
|
}
|
|
self.site_id = "default"
|
|
|
|
def get_devices(self):
|
|
try:
|
|
url = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device"
|
|
response = self.session.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
|
|
# Log raw response
|
|
logger.debug(f"Response status: {response.status_code}")
|
|
logger.debug(f"Response headers: {response.headers}")
|
|
logger.debug(f"Raw response text: {response.text}")
|
|
|
|
devices_data = response.json()
|
|
logger.debug(f"Parsed JSON: {devices_data}")
|
|
|
|
# Extract network_devices from the response
|
|
network_devices = devices_data.get('network_devices', [])
|
|
|
|
devices = []
|
|
for device in network_devices:
|
|
devices.append({
|
|
'name': device.get('name', 'Unknown'),
|
|
'ip': device.get('ip', '0.0.0.0'),
|
|
'type': device.get('type', 'unknown'),
|
|
'connection_type': 'fiber' if device.get('uplink', {}).get('media') == 'sfp' else 'copper',
|
|
'critical': True if device.get('type') in ['udm', 'usw'] else False,
|
|
'device_id': device.get('mac')
|
|
})
|
|
|
|
logger.debug(f"Processed devices: {devices}")
|
|
return devices
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching devices: {e}")
|
|
logger.exception("Full traceback:")
|
|
return []
|
|
|
|
def get_device_details(self, device_id):
|
|
try:
|
|
url = f"{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device/{device_id}"
|
|
response = self.session.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
logger.error(f"Failed to get device details: {e}")
|
|
return None
|
|
|
|
def get_device_diagnostics(self, device):
|
|
details = self.get_device_details(device['device_id'])
|
|
if not details:
|
|
return {'state': 'ERROR', 'error': 'Failed to fetch device details'}
|
|
|
|
diagnostics = {
|
|
'state': details.get('state', 'unknown'),
|
|
'interfaces': {
|
|
'ports': {}
|
|
}
|
|
}
|
|
|
|
# Parse port information
|
|
for port in details.get('port_table', []):
|
|
diagnostics['interfaces']['ports'][f"Port {port.get('port_idx')}"] = {
|
|
'state': 'up' if port.get('up') else 'down',
|
|
'speed': {
|
|
'current': port.get('speed', 0),
|
|
'max': port.get('max_speed', 0)
|
|
},
|
|
'poe': port.get('poe_enable', False),
|
|
'media': port.get('media', 'unknown')
|
|
}
|
|
|
|
return diagnostics
|
|
|
|
def _parse_interfaces(self, interfaces):
|
|
result = {
|
|
'ports': {},
|
|
'radios': {}
|
|
}
|
|
for port in interfaces:
|
|
result['ports'][f"port_{port['index']}"] = {
|
|
'state': port['up'] and 'up' or 'down',
|
|
'speed': {
|
|
'current': port.get('speed', 0),
|
|
'max': port.get('max_speed', 0)
|
|
}
|
|
}
|
|
return result
|
|
|
|
@app.route('/')
|
|
def home():
|
|
config = load_config()
|
|
unifi = UnifiAPI(config)
|
|
devices = unifi.get_devices()
|
|
return render_template('index.html', devices=devices)
|
|
|
|
@app.route('/api/status')
|
|
def status():
|
|
return jsonify(device_status)
|
|
|
|
@app.route('/api/diagnostics')
|
|
def get_diagnostics():
|
|
config = load_config()
|
|
unifi = UnifiAPI(config)
|
|
devices = unifi.get_devices()
|
|
diagnostics = {}
|
|
for device in devices:
|
|
diagnostics[device['name']] = unifi.get_device_diagnostics(device)
|
|
return jsonify(diagnostics)
|
|
|
|
if __name__ == '__main__':
|
|
status_thread = threading.Thread(target=update_status, daemon=True)
|
|
status_thread.start()
|
|
app.run(debug=True) |