import random import requests import json import os from flask import Flask, render_template, abort from utils.proxmox_client import ProxmoxClient from utils.check_mk_client import CheckMKClient from config import settings app = Flask(__name__) def make_client(client_cls, base_url: str): """Generic factory to construct API clients with TLS and auth from settings.""" return client_cls( base_url, user=settings.CHECK_MK_USER if client_cls is CheckMKClient else settings.PROXMOX_USER, password=settings.CHECK_MK_PASSWORD if client_cls is CheckMKClient else settings.PROXMOX_PASSWORD, api_token=( settings.CHECK_MK_API_TOKEN if client_cls is CheckMKClient else settings.PROXMOX_API_TOKEN) or None, verify=settings.VERIFY_TLS, ca_bundle=settings.CA_BUNDLE or None, ) proxmox = make_client(ProxmoxClient, settings.PROXMOX_API_URL) checkmk = make_client(CheckMKClient, settings.CHECK_MK_API_URL) @app.route('/') def index(): # gather cluster hosts and VMs try: cluster = proxmox.get_cluster() except Exception as e: return render_template('error.html', error=str(e)), 500 # enrich hosts with check_mk status hosts = [] for node in cluster.get('nodes', []): host = { 'name': node.get('name'), 'status': node.get('status'), # convert to GB 'memory': round((node.get('memory') or node.get('maxmem') or 0)/1024/1024/1024, 2), 'cpu': node.get('cpu') or node.get('maxcpu') or 0, 'vm_count': len(node.get('qemu', [])) if node.get('qemu') else 0, 'lxc_count': len(node.get('lxc', [])) if node.get('lxc') else 0, } try: host['check_mk'] = checkmk.get_host_status(host['name']) except Exception: host['check_mk'] = None hosts.append(host) return render_template('index.html', hosts=hosts) @app.route('/host/') def host_detail(hostname): # get services for host from check_mk try: services = checkmk.get_host_services(hostname) except Exception as e: return render_template('error.html', error=str(e)), 500 return render_template('host_detail.html', hostname=hostname, services=services) @app.route('/service/') def service(url: str): print(f"Received URL parameter: {url}") # urldecode service_url service_url = url.encode('utf-8').decode('unicode_escape') print(f"Decoded service URL: {service_url}") client = make_client(CheckMKClient, settings.CHECK_MK_API_URL) # fetch service detail from check_mk try: request = client.get_service_detail(service_url) ret = request except Exception as e: return render_template('error.html', error=str(e)), 500 return render_template('service_detail.html', service=ret) @app.route('/numbers') def numbers(): # generates numbers for the data cascade design n = 216 # total numbers to generate # n = 24 * 12 # n = 9*7 min_len = 1 max_len = 7 letter_list = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'K', 'L', 'S', 'T', 'U', 'X', 'Y', 'Z'] number_list = [] for i in range(1, n + 1): prefix_length = random.randint(0, 2) prefix = '' for k in range(0, prefix_length): prefix += random.choice(letter_list) if prefix_length > 0 and random.choice([True, False]): prefix += '-' else: prefix = '' max_num = 10**(max_len - len(prefix)) - 1 if max_num < min_len: max_num = min_len number = random.randint(1, max_num) number_list.append(f"{prefix}{number}") return json.dumps(number_list) @app.route('/favicon.ico') def favicon(): svg = """ """ return svg, 200, {'Content-Type': 'image/svg+xml'} if __name__ == '__main__': # configurable port via APP_PORT env var; default to 8081 try: APP_PORT = int(os.getenv('APP_PORT', '8081')) except Exception: APP_PORT = 8081 app.run(host='0.0.0.0', port=APP_PORT, debug=True)