gandalf/app.py

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

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# Disable InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# Initialize Flask app
app = Flask(__name__)

# Global state
device_status = {}

# Configuration functions
def load_config():
    with open('config.json') as f:
        return json.load(f)

# Network utility functions
def ping(host):
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    command = ['ping', param, '1', host]
    return subprocess.call(command, stdout=subprocess.DEVNULL) == 0

def send_webhook(device, status, diagnostics):
    config = load_config()
    webhook_data = {
        "device": device,
        "status": status,
        "timestamp": datetime.now().isoformat(),
        "diagnostics": diagnostics
    }
    requests.post(config['webhook_url'], json=webhook_data)

# UniFi API integration
class UnifiAPI:
    def __init__(self, config):
        self.base_url = config['unifi']['controller']
        self.session = requests.Session()
        self.session.verify = False
        self.headers = {
            'Authorization': f'Bearer {config["unifi"]["api_key"]}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }
        self.site_id = self.get_site_id()

    def get_site_id(self):
        url = f"{self.base_url}/proxy/network/api/self/sites"
        response = self.session.get(url, headers=self.headers)
        response.raise_for_status()
        sites = response.json()['data']
        return sites[0]['_id']

    def get_all_devices(self):
        url = f"{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device"
        response = self.session.get(url, headers=self.headers)
        response.raise_for_status()
        return response.json()['data']    
    
    def get_device_details(self, device_id):
        url = f"{self.base_url}/proxy/network/integration/v1/sites/{self.site_id}/devices/{device_id}"
        response = self.session.get(url, headers=self.headers)
        response.raise_for_status()
        logger.debug(f"Device details response: {response.text}")
        data = response.json()
        return {
            'state': data['state'],
            'firmware': {
                'version': data['firmwareVersion'],
                'updatable': data.get('firmwareUpdatable', False)
            },
            'network': {
                'ip': data['ipAddress'],
                'mac': data['macAddress']
            }
        }

    def get_device_diagnostics(self, device):
        details = self.get_device_details(device['ip'])
        if not details:
            return {'state': 'ERROR', 'error': 'Failed to fetch device details'}
            
        diagnostics = {
            'state': details['state'],
            'firmware': details['firmware'],
            'network': details['network'],
            'uptime': {
                'adopted_at': details.get('adoptedAt'),
                'provisioned_at': details.get('provisionedAt')
            },
            'interfaces': details['interfaces']
        }
        
        return diagnostics

    def _parse_interfaces(self, interfaces):
        result = {
            'ports': {},
            'radios': {}
        }
        
        for port in interfaces.get('ports', []):
            result['ports'][f"port_{port['idx']}"] = {
                'state': port['state'],
                'type': port['connector'],
                'speed': {
                    'current': port['speedMbps'],
                    'max': port['maxSpeedMbps']
                }
            }
            
        for radio in interfaces.get('radios', []):
            result['radios'][f"{radio['frequencyGHz']}GHz"] = {
                'standard': radio['wlanStandard'],
                'channel': radio['channel'],
                'width': f"{radio['channelWidthMHz']}MHz"
            }
            
        return result
def run_diagnostics(device):
    try:
        config = load_config()
        unifi = UnifiAPI(config)
        diagnostics = unifi.get_device_diagnostics(device)
        logger.debug(f"Got diagnostics for {device['name']}: {diagnostics}")
        return diagnostics
    except Exception as e:
        logger.error(f"Error getting diagnostics for {device['name']}: {str(e)}")
        return {"error": str(e)}

def update_status():
    while True:
        config = load_config()
        for device in config['devices']:
            current_status = ping(device['ip'])
            previous_status = device_status.get(device['name'], True)
            
            if current_status != previous_status:
                diagnostics = run_diagnostics(device)
                send_webhook(device, current_status, diagnostics)
            
            device_status[device['name']] = current_status
        time.sleep(config['check_interval'])

# Flask routes
@app.route('/')
def home():
    config = load_config()
    logger.debug(f"Loaded devices: {config['devices']}")
    devices = config['devices']
    logger.debug(f"Rendering template with devices: {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():
    try:
        config = load_config()
        unifi = UnifiAPI(config)
        logger.debug("Getting all devices...")
        devices = unifi.get_all_devices()
        logger.debug(f"Raw devices response: {devices}")
        
        diagnostics = {}
        for device in config['devices']:
            if device.get('device_id'):
                logger.debug(f"Getting diagnostics for {device['name']}")
                device_details = unifi.get_device_details(device['device_id'])
                logger.debug(f"Raw device details: {device_details}")
                diagnostics[device['name']] = device_details
                
        return jsonify(diagnostics)
    except Exception as e:
        logger.error(f"Error in diagnostics endpoint: {str(e)}")
        logger.exception("Full traceback:")
        return jsonify({"error": str(e)}), 500

# Application entry point
if __name__ == '__main__':
    status_thread = threading.Thread(target=update_status, daemon=True)
    status_thread.start()
    app.run(debug=True)