147 lines
5.3 KiB
Python
147 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import datetime
|
|
from typing import Any, Optional, Sequence, Self
|
|
from traceback import extract_tb as extract_traceback, format_stack as trace_format_stack
|
|
from re import Match as REMatch
|
|
from Utils.Patterns import RE
|
|
from Utils.Utils import Utils
|
|
from Drivers.FilesDrivers import FilesDriver
|
|
from Managers.I18NManager import I18NManager
|
|
from Managers.SettingsManager import SettingsManager
|
|
from Managers.ModelsManager import ModelsManager
|
|
from Managers.TerminalManager import TerminalManager
|
|
from Managers.RoutesManager import RoutesManager
|
|
from Managers.ControllersManager import ControllersManager
|
|
from Managers.DispatchersManager import DispatchersManager
|
|
from Managers.DatabasesManager import DatabasesManager
|
|
from Managers.WebServersManager import WebServersManager
|
|
|
|
class NucelarMonitor:
|
|
|
|
def __init__(self:Self, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
|
|
|
|
self.__print_format:str = "[{type}] {yyyy}-{mm}-{dd} {hh}:{ii}:{ss} - {message}"
|
|
self.__print_types:list[list[str]] = [
|
|
["unkn", "unknown"],
|
|
["info", "information"],
|
|
["warn", "warning"],
|
|
["erro", "error", "wrong", "failure", "fail", "no"],
|
|
["exce", "exception", "except"],
|
|
[" ok ", "ok", "success", "succeed", "yes"],
|
|
["test", "debug"]
|
|
]
|
|
self.__exception_format:str = " '[{line}]{file}({method})'{lines}\n\n{exception_message}"
|
|
self.files:FilesDriver = FilesDriver(self)
|
|
self.i18n:I18NManager = I18NManager(self)
|
|
self.settings:SettingsManager = SettingsManager(self, inputs)
|
|
|
|
self.i18n.set_defaults()
|
|
self.__print_format:str = self.settings.get("print_format", None, self.__print_format)
|
|
self.__print_types:list[list[str]] = self.settings.get("print_types", None, self.__print_types)
|
|
self.__exception_format:str = self.settings.get("exception_format", None, self.__exception_format)
|
|
self.__started:bool = False
|
|
self.__working:bool = True
|
|
|
|
self.models:ModelsManager = ModelsManager(self)
|
|
self.terminal:TerminalManager = TerminalManager(self)
|
|
self.controllers:ControllersManager = ControllersManager(self)
|
|
self.dispatchers:DispatchersManager = DispatchersManager(self)
|
|
self.routes:RoutesManager = RoutesManager(self)
|
|
self.databases:DatabasesManager = DatabasesManager(self)
|
|
self.web_servers:WebServersManager = WebServersManager(self)
|
|
|
|
if self.settings.get("autostart"):
|
|
self.start()
|
|
else:
|
|
self.terminal.start()
|
|
|
|
def start(self:Self) -> None:
|
|
if self.__started:
|
|
return
|
|
self.__started = True
|
|
self.__working = True
|
|
|
|
self.terminal.start()
|
|
self.web_servers.start()
|
|
|
|
def close(self:Self) -> None:
|
|
if not self.__started:
|
|
return
|
|
self.__started = False
|
|
self.__working = False
|
|
|
|
self.terminal.close()
|
|
self.web_servers.close()
|
|
self.databases.close()
|
|
|
|
def is_working(self:Self) -> bool:
|
|
return self.__working
|
|
|
|
def get_print_type(self:Self, _type:str) -> str:
|
|
|
|
group:list[str]
|
|
|
|
for group in self.__print_types:
|
|
if _type in group:
|
|
return group[0].upper()
|
|
return self.__print_types[0][0].upper()
|
|
|
|
def print(self:Self,
|
|
_type:str,
|
|
message:str|Sequence[str],
|
|
inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None,
|
|
i:int = 0
|
|
) -> None:
|
|
|
|
date:datetime = datetime.datetime.now()
|
|
own:dict[str, Any|None] = {
|
|
"raw_type" : _type,
|
|
"type" : self.get_print_type(_type),
|
|
"i18n" : Utils.get_texts(message),
|
|
"message" : (
|
|
self.i18n.get(message, inputs) if hasattr(self, "i18n") else
|
|
Utils.string_variables(Utils.get_texts(message)[0], inputs)),
|
|
**Utils.get_dictionary(inputs),
|
|
**Utils.get_action_data(i + 1)
|
|
}
|
|
|
|
for key in ("year", "month", "day", "hour", "minute", "second"):
|
|
|
|
k:str = "i" if key == "minute" else key[0]
|
|
|
|
own[k] = own[key] = getattr(date, key)
|
|
own[k + k] = ("00" + str(own[key]))[-2:]
|
|
|
|
own["yyyy"] = own["year"]
|
|
|
|
print(Utils.string_variables(self.__print_format, own) + (own["end"] if "end" in own else ""))
|
|
|
|
def exception(self:Self,
|
|
exception:Exception,
|
|
message:str|Sequence[str],
|
|
inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None,
|
|
i:int = 0
|
|
) -> None:
|
|
|
|
lines:list[str] = extract_traceback(exception.__traceback__).format()
|
|
matches:REMatch = RE.EXCEPTION.match(lines[-1])
|
|
data:dict[str, Any|None] = {
|
|
**Utils.get_dictionary(inputs),
|
|
"lines" : "",
|
|
"exception_message" : str(exception),
|
|
"method" : matches.group(3),
|
|
"line" : matches.group(2),
|
|
"file" : matches.group(1)
|
|
}
|
|
block:str
|
|
j:int
|
|
|
|
for j, block in enumerate(trace_format_stack()[:-2] + lines):
|
|
if block:
|
|
data["lines"] += "\n " + str(j) + " - " + RE.NEW_LINE.split(block.strip())[0]
|
|
|
|
data["end"] = Utils.string_variables(self.__exception_format, data)
|
|
|
|
message and self.print("exception", message, data, i + 2) |