mirror of
https://github.com/iio612/DEFENDER.git
synced 2026-02-13 11:14:23 +00:00
New features on branch v6.2.5:
- New capability in base.py to patch the database
- Some minor updates on installation.py.
- Translation feature:
- New library requirement (pyyaml)
- New translation systeme implemented.
- New class translation.py added.
- Module folder updated by adding new folder language.
- Core module updated as well with new language folder.
This commit is contained in:
@@ -211,15 +211,6 @@ class Channel:
|
||||
|
||||
return None
|
||||
|
||||
def get_channel_asdict(self, channel_name: str) -> Optional[dict[str, Any]]:
|
||||
|
||||
channel_obj: Optional['MChannel'] = self.get_channel(channel_name)
|
||||
|
||||
if channel_obj is None:
|
||||
return None
|
||||
|
||||
return channel_obj.to_dict()
|
||||
|
||||
def is_valid_channel(self, channel_to_check: str) -> bool:
|
||||
"""Check if the string has the # caractere and return True if this is a valid channel
|
||||
|
||||
@@ -276,7 +267,7 @@ class Channel:
|
||||
mes_donnees = {'datetime': self.Utils.get_sdatetime(), 'channel_name': channel_name, 'module_name': module_name}
|
||||
insert = self.Base.db_execute_query(f"INSERT INTO {core_table} (datetime, channel_name, module_name) VALUES (:datetime, :channel_name, :module_name)", mes_donnees)
|
||||
if insert.rowcount:
|
||||
self.Logs.debug(f'New channel added: channel={channel_name} / module_name={module_name}')
|
||||
self.Logs.debug(f'Channel added to DB: channel={channel_name} / module_name={module_name}')
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
@@ -286,7 +277,7 @@ class Channel:
|
||||
response = self.Base.db_execute_query(f"DELETE FROM {core_table} WHERE channel_name = :channel_name AND module_name = :module_name", mes_donnes)
|
||||
|
||||
if response.rowcount > 0:
|
||||
self.Logs.debug(f'Channel deleted: channel={channel_name} / module: {module_name}')
|
||||
self.Logs.debug(f'Channel deleted from DB: channel={channel_name} / module: {module_name}')
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
@@ -11,6 +11,7 @@ class Command:
|
||||
def __init__(self, loader: 'Loader'):
|
||||
self.Loader = loader
|
||||
self.Base = loader.Base
|
||||
self.Logs = loader.Logs
|
||||
|
||||
def build(self, new_command_obj: MCommand) -> bool:
|
||||
|
||||
@@ -45,6 +46,27 @@ class Command:
|
||||
|
||||
return False
|
||||
|
||||
def drop_command_by_module(self, module_name: str) -> bool:
|
||||
"""Drop all command by module
|
||||
|
||||
Args:
|
||||
module_name (str): The module name
|
||||
|
||||
Returns:
|
||||
bool: True
|
||||
"""
|
||||
tmp_model: list[MCommand] = []
|
||||
|
||||
for command in self.DB_COMMANDS:
|
||||
if command.module_name.lower() == module_name.lower():
|
||||
tmp_model.append(command)
|
||||
|
||||
for c in tmp_model:
|
||||
self.DB_COMMANDS.remove(c)
|
||||
|
||||
self.Logs.debug(f"[COMMAND] Drop command for module {module_name}")
|
||||
return True
|
||||
|
||||
def get_ordered_commands(self) -> list[MCommand]:
|
||||
return sorted(self.DB_COMMANDS, key=lambda c: (c.command_level, c.module_name))
|
||||
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
'''This class should never be reloaded.
|
||||
'''
|
||||
from logging import Logger
|
||||
from threading import Timer, Thread, RLock
|
||||
from socket import socket
|
||||
from typing import Any, Optional
|
||||
from typing import Any, Optional, TYPE_CHECKING
|
||||
from core.definition import MSModule
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from core.classes.user import User
|
||||
|
||||
class Settings:
|
||||
"""This Class will never be reloaded.
|
||||
Means that the variables are available during
|
||||
@@ -29,6 +33,17 @@ class Settings:
|
||||
__CACHE: dict[str, Any] = {}
|
||||
"""Use set_cache or get_cache instead"""
|
||||
|
||||
__TRANSLATION: dict[str, list[list[str]]] = dict()
|
||||
"""Translation Varibale"""
|
||||
|
||||
__LANG: str = "EN"
|
||||
|
||||
__INSTANCE_OF_USER_UTILS: Optional['User'] = None
|
||||
"""Instance of the User Utils class"""
|
||||
|
||||
__LOGGER: Optional[Logger] = None
|
||||
"""Instance of the logger"""
|
||||
|
||||
def set_cache(self, key: str, value_to_cache: Any):
|
||||
"""When you want to store a variable
|
||||
|
||||
@@ -56,4 +71,38 @@ class Settings:
|
||||
self.__CACHE.clear()
|
||||
|
||||
def show_cache(self) -> dict[str, Any]:
|
||||
return self.__CACHE.copy()
|
||||
return self.__CACHE.copy()
|
||||
|
||||
@property
|
||||
def global_translation(self) -> dict[str, list[list[str]]]:
|
||||
return self.__TRANSLATION
|
||||
|
||||
@global_translation.setter
|
||||
def global_translation(self, translation_var: dict) -> None:
|
||||
self.__TRANSLATION = translation_var
|
||||
|
||||
@property
|
||||
def global_lang(self) -> str:
|
||||
return self.__LANG
|
||||
|
||||
@global_lang.setter
|
||||
def global_lang(self, lang: str) -> None:
|
||||
self.__LANG = lang
|
||||
|
||||
@property
|
||||
def global_user(self) -> 'User':
|
||||
return self.__INSTANCE_OF_USER_UTILS
|
||||
|
||||
@global_user.setter
|
||||
def global_user(self, user_utils_instance: 'User') -> None:
|
||||
self.__INSTANCE_OF_USER_UTILS = user_utils_instance
|
||||
|
||||
@property
|
||||
def global_logger(self) -> Logger:
|
||||
return self.__LOGGER
|
||||
|
||||
@global_logger.setter
|
||||
def global_logger(self, logger: Logger) -> None:
|
||||
self.__LOGGER = logger
|
||||
|
||||
global_settings = Settings()
|
||||
92
core/classes/translation.py
Normal file
92
core/classes/translation.py
Normal file
@@ -0,0 +1,92 @@
|
||||
import yaml
|
||||
import yaml.scanner
|
||||
from os import sep
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from core.loader import Loader
|
||||
|
||||
|
||||
class Translation:
|
||||
|
||||
def __init__(self, loader: 'Loader') -> None:
|
||||
self.Logs = loader.Logs
|
||||
self.Settings = loader.Settings
|
||||
return None
|
||||
|
||||
def get_translation(self) -> dict[str, list[list[str]]]:
|
||||
try:
|
||||
translation: dict[str, list[list[str]]] = dict()
|
||||
sfs: dict[str, list[list[str]]] = {}
|
||||
|
||||
module_translation_directory = Path("mods")
|
||||
core_translation_directory = Path("core")
|
||||
sfs_core = self.get_subfolders_name(core_translation_directory.__str__())
|
||||
sfs_module = self.get_subfolders_name(module_translation_directory.__str__())
|
||||
|
||||
# Combine the 2 dict
|
||||
for d in (sfs_core, sfs_module):
|
||||
for k, v in d.items():
|
||||
sfs.setdefault(k, []).extend(v)
|
||||
|
||||
loaded_files: list[str] = []
|
||||
|
||||
for module, filenames in sfs.items():
|
||||
translation[module] = []
|
||||
for filename in filenames:
|
||||
with open(f"{filename}", "r", encoding="utf-8") as fyaml:
|
||||
data: dict[str, list[dict[str, str]]] = yaml.safe_load(fyaml)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
continue
|
||||
|
||||
for key, list_trad in data.items():
|
||||
for vlist in list_trad:
|
||||
translation[module].append([vlist["orig"], vlist["trad"]])
|
||||
|
||||
loaded_files.append(f"{filename}")
|
||||
|
||||
return translation
|
||||
|
||||
except yaml.scanner.ScannerError as se:
|
||||
self.Logs.error(f"[!] {se} [!]")
|
||||
return {}
|
||||
except yaml.YAMLError as ye:
|
||||
if hasattr(ye, 'problem_mark'):
|
||||
mark = ye.problem_mark
|
||||
self.Logs.error(f"Error YAML: {ye.with_traceback(None)}")
|
||||
self.Logs.error("Error position: (%s:%s)" % (mark.line+1, mark.column+1))
|
||||
return {}
|
||||
except yaml.error.MarkedYAMLError as me:
|
||||
self.Logs.error(f"[!] {me} [!]")
|
||||
return {}
|
||||
except Exception as err:
|
||||
self.Logs.error(f'General Error: {err}', exc_info=True)
|
||||
return {}
|
||||
|
||||
finally:
|
||||
self.Logs.debug("Translation files loaded")
|
||||
for f in loaded_files:
|
||||
self.Logs.debug(f" - {f}")
|
||||
|
||||
def get_subfolders_name(self, directory: str) -> dict[str, list[str]]:
|
||||
try:
|
||||
translation_information: dict[str, list[str]] = dict()
|
||||
main_directory = Path(directory)
|
||||
|
||||
# Init the dictionnary
|
||||
for subfolder in main_directory.rglob(f'*language{sep}*{sep}*.yaml'):
|
||||
if subfolder.name != '__pycache__':
|
||||
translation_information[subfolder.parent.name.lower()] = []
|
||||
|
||||
|
||||
for subfolder in main_directory.rglob(f'*language{sep}*{sep}*.yaml'):
|
||||
if subfolder.name != '__pycache__':
|
||||
translation_information[subfolder.parent.name.lower()].append(subfolder)
|
||||
|
||||
return translation_information
|
||||
|
||||
except Exception as err:
|
||||
self.Logs.error(f'General Error: {err}')
|
||||
return {}
|
||||
@@ -10,10 +10,15 @@ class User:
|
||||
|
||||
UID_DB: list['MUser'] = []
|
||||
|
||||
@property
|
||||
def get_current_user(self) -> 'MUser':
|
||||
return self.current_user
|
||||
|
||||
def __init__(self, loader: 'Loader'):
|
||||
|
||||
self.Logs = loader.Logs
|
||||
self.Base = loader.Base
|
||||
self.current_user: Optional['MUser'] = None
|
||||
|
||||
def insert(self, new_user: 'MUser') -> bool:
|
||||
"""Insert a new User object
|
||||
@@ -126,8 +131,10 @@ class User:
|
||||
"""
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
self.current_user = record
|
||||
return record
|
||||
elif record.nickname == uidornickname:
|
||||
self.current_user = record
|
||||
return record
|
||||
|
||||
return None
|
||||
@@ -147,6 +154,7 @@ class User:
|
||||
if user_obj is None:
|
||||
return None
|
||||
|
||||
self.current_user = user_obj
|
||||
return user_obj.uid
|
||||
|
||||
def get_nickname(self, uidornickname:str) -> Optional[str]:
|
||||
@@ -163,6 +171,7 @@ class User:
|
||||
if user_obj is None:
|
||||
return None
|
||||
|
||||
self.current_user = user_obj
|
||||
return user_obj.nickname
|
||||
|
||||
def get_user_asdict(self, uidornickname: str) -> Optional[dict[str, Any]]:
|
||||
|
||||
Reference in New Issue
Block a user