#!/usr/bin/env python3 # Beszel API Client - Pour qu'Effi interroge les métriques Beszel # Usage: python3 beszel_client.py [--system SYSTEM_ID] [--check CHECK_TYPE] import requests import argparse import sys import os import json from datetime import datetime BESZEL_URL = os.getenv("BESZEL_URL", "http://192.168.1.25:8090") class BeszelClient: def __init__(self, username=None, password=None): self.url = BESZEL_URL self.token = None self.username = username or os.getenv("BESZEL_USER", "admin") self.password = password or os.getenv("BESZEL_PASSWORD", "") def login(self): try: response = requests.post(f"{self.url}/api/auth/login", json={ "identity": self.username, "password": self.password }, timeout=10) if response.status_code == 200: self.token = response.json().get("token") return True return False except Exception as e: print(f"Erreur login: {e}") return False def get_systems(self): if not self.token: return None try: headers = {"Authorization": f"Bearer {self.token}"} response = requests.get(f"{self.url}/api/systems", headers=headers, timeout=10) if response.status_code == 200: return response.json() return None except Exception as e: print(f"Erreur get_systems: {e}") return None def get_system_stats(self, system_id): if not self.token: return None try: headers = {"Authorization": f"Bearer {self.token}"} response = requests.get(f"{self.url}/api/systems/{system_id}/stats", headers=headers, timeout=10) if response.status_code == 200: return response.json() return None except Exception as e: print(f"Erreur get_stats: {e}") return None def get_system_info(self, system_id): if not self.token: return None try: headers = {"Authorization": f"Bearer {self.token}"} response = requests.get(f"{self.url}/api/systems/{system_id}", headers=headers, timeout=10) if response.status_code == 200: return response.json() return None except Exception as e: print(f"Erreur get_info: {e}") return None def format_stats_text(self, stats, system_name="System"): lines = [] lines.append(f"=== {system_name} ===") lines.append(f"CPU: {stats.get('cpu', 0):.1f}%") lines.append(f"Mémoire: {stats.get('memory', 0):.1f}%") if 'swap' in stats: lines.append(f"Swap: {stats.get('swap', 0):.1f}%") if 'disk' in stats: for disk, info in stats['disk'].items(): lines.append(f"Disque {disk}: {info.get('used_percent', 0):.1f}%") if 'load' in stats: load = stats['load'] lines.append(f"Load: {load[0]:.2f} {load[1]:.2f} {load[2]:.2f}") if 'gpu' in stats and stats['gpu']: lines.append("") lines.append("GPU:") for gpu_name, gpu_data in stats['gpu'].items(): vram_used = gpu_data.get('memory_used', 0) / (1024**3) vram_total = gpu_data.get('memory_total', 0) / (1024**3) vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100 lines.append(f" {gpu_name}:") lines.append(f" VRAM: {vram_used:.2f}GB / {vram_total:.2f}GB ({vram_pct:.1f}%)") lines.append(f" Utilisation GPU: {gpu_data.get('utilization_gpu', 0)}%") if 'temperature' in gpu_data: lines.append(f" Température: {gpu_data.get('temperature', 0)}°C") if 'power_draw' in gpu_data: lines.append(f" Consommation: {gpu_data.get('power_draw', 0)}W") if 'containers' in stats and stats['containers']: lines.append("") lines.append("Containers Docker:") for container in stats['containers']: name = container.get('name', 'unknown') cpu = container.get('cpu_percent', 0) mem = container.get('memory_percent', 0) lines.append(f" {name}: CPU {cpu:.1f}% MEM {mem:.1f}%") return "\n".join(lines) def check_thresholds(self, stats, thresholds=None): if thresholds is None: thresholds = { 'cpu': 80, 'memory': 90, 'vram': 90 } alerts = [] cpu = stats.get('cpu', 0) if cpu > thresholds['cpu']: alerts.append(f"⚠️ CPU elevated: {cpu:.1f}%") memory = stats.get('memory', 0) if memory > thresholds['memory']: alerts.append(f"⚠️ Memory elevated: {memory:.1f}%") if 'gpu' in stats and stats['gpu']: for gpu_name, gpu_data in stats['gpu'].items(): vram_pct = (gpu_data.get('memory_used', 0) / gpu_data.get('memory_total', 1)) * 100 if vram_pct > thresholds['vram']: alerts.append(f"⚠️ VRAM {gpu_name} elevated: {vram_pct:.1f}%") util = gpu_data.get('utilization_gpu', 0) if util > thresholds['vram']: alerts.append(f"⚠️ GPU {gpu_name} utilization elevated: {util}%") return alerts def main(): parser = argparse.ArgumentParser(description="Beszel API Client") parser.add_argument("--system", "-s", help="System ID to query") parser.add_argument("--list-systems", "-l", action="store_true", help="List all systems") parser.add_argument("--check", "-c", choices=["stats", "alerts", "full"], default="stats", help="Type of check") parser.add_argument("--json", "-j", action="store_true", help="Output JSON") parser.add_argument("--user", "-u", help="Beszel username") parser.add_argument("--password", "-p", help="Beszel password") args = parser.parse_args() client = BeszelClient(username=args.user, password=args.password) if not client.login(): print("❌ Échec de connexion à Beszel") sys.exit(1) if args.list_systems: systems = client.get_systems() if systems: print("=== Systèmes surveillés ===") for sys in systems.get('items', []): print(f" - {sys.get('name')} (ID: {sys.get('id')})") return if args.system: if args.check == "stats": stats = client.get_system_stats(args.system) if stats: if args.json: print(json.dumps(stats, indent=2)) else: info = client.get_system_info(args.system) name = info.get('name', args.system) if info else args.system print(client.format_stats_text(stats, name)) elif args.check == "alerts": stats = client.get_system_stats(args.system) if stats: alerts = client.check_thresholds(stats) if alerts: print("\n".join(alerts)) else: print("✅ Tout est normal") elif args.check == "full": stats = client.get_system_stats(args.system) info = client.get_system_info(args.system) if stats and info: if args.json: print(json.dumps({'info': info, 'stats': stats}, indent=2, default=str)) else: name = info.get('name', args.system) print(f"=== {name} ===") print(client.format_stats_text(stats, name)) alerts = client.check_thresholds(stats) if alerts: print("\n" + "\n".join(alerts)) else: systems = client.get_systems() if systems: print("Systèmes disponibles (utiliser --system ID):") for sys in systems.get('items', []): print(f" - {sys.get('name')} (ID: {sys.get('id')})") if __name__ == "__main__": main()