#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Self from Abstracts.ControllerAbstract import ControllerAbstract from Models.ResponseModel import ResponseModel from Models.RequestModel import RequestModel from Models.IterationModel import ( CPUModel, MemoryModel, NetUseModel, InterfaceModel, DiskModel, CandleTimeModel, IterationModel ) from Dispatchers.AgentsDispatcher import AgentsDispatcher from Utils.Utils import Utils class AgentsController(ControllerAbstract): def debian(self:Self, request:RequestModel, response:ResponseModel) -> None: key:str = request.get("key") hostnames:list[str] domain:str|None interfaces_data:list[list[int, str, bool, str, int]] disks:list[list[str, int, int, str|None]] iterations:int candle_times:list[int, int] cpu:list[float, float, float, float, float] memory:list[int, int, int, int, int, float] net_use:list[list[list[str, int, int, int, int, int, int]]] received:list[NetUseModel] = [] transmitted:list[NetUseModel] = [] ( hostnames, domain, interfaces_data, disks, iterations, candle_times, cpu, memory, net_use ) = Utils.json_decode(request.body) print(Utils.json_decode(request.body)) received = {net[0] : NetUseModel(*net[1:]) for i, net in enumerate(net_use[0])} transmitted = {net[0] : NetUseModel(*net[1:]) for i, net in enumerate(net_use[1])} self.nucelar_monitor.dispatchers.get(AgentsDispatcher, "sql_server").save(IterationModel( key, hostnames, domain, [InterfaceModel( i, name, is_ipv6, ip, mask, received[name], transmitted[name] ) for i, name, is_ipv6, ip, mask in interfaces_data], [DiskModel(*disk) for disk in disks], iterations, CandleTimeModel(*candle_times), CPUModel(*cpu), MemoryModel(*memory) )) def windows(self:Self, request:RequestModel, response:ResponseModel) -> None: key:str = request.get("key") hostnames:list[str] domain:str|None interfaces:list[str] disks:list[list[str, int, int]] iterations:int candle_times:list[int, int] cpu:list[float, float, float, float, float] memory:list[int, int, int, int, int, float] net_use:list[list[list[str, int, int, int, int, int, int]]] = [] print(Utils.json_decode(request.body)) hostnames, domain, interfaces, disks, candle_times, iterations, cpu, memory = Utils.json_decode(request.body) print([hostnames, domain, interfaces, disks, iterations, candle_times, cpu, memory, net_use]) def test(self:Self, request:RequestModel, response:ResponseModel) -> None: response.set({ "message": "Test successful", "key" : request.get("key") }, "application/json")