From 4c93f85008efdec8d11e9b352264de5f4132894f Mon Sep 17 00:00:00 2001 From: adator <85586985+adator85@users.noreply.github.com> Date: Sun, 23 Nov 2025 18:19:29 +0100 Subject: [PATCH] Code refactoring on system modules. --- core/classes/modules/admin.py | 39 ++++++++++++--------------- core/classes/modules/client.py | 43 +++++++++--------------------- core/classes/modules/commands.py | 22 ++++++++++----- core/classes/modules/reputation.py | 14 +++++----- core/classes/modules/user.py | 18 +++++++------ core/definition.py | 19 +++++++++++++ 6 files changed, 81 insertions(+), 74 deletions(-) diff --git a/core/classes/modules/admin.py b/core/classes/modules/admin.py index a71a87d..c5f4304 100644 --- a/core/classes/modules/admin.py +++ b/core/classes/modules/admin.py @@ -14,12 +14,7 @@ class Admin: Args: loader (Loader): The Loader Instance. """ - self.Logs = loader.Logs - self.Base = loader.Base - self.Setting = loader.Settings - self.Config = loader.Config - self.User = loader.User - self.Definition = loader.Definition + self._ctx = loader def insert(self, new_admin: MAdmin) -> bool: """Insert a new admin object model @@ -33,11 +28,11 @@ class Admin: for record in self.UID_ADMIN_DB: if record.uid == new_admin.uid: - self.Logs.debug(f'{record.uid} already exist') + self._ctx.Logs.debug(f'{record.uid} already exist') return False self.UID_ADMIN_DB.append(new_admin) - self.Logs.debug(f'A new admin ({new_admin.nickname}) has been created') + self._ctx.Logs.debug(f'A new admin ({new_admin.nickname}) has been created') return True def update_nickname(self, uid: str, new_admin_nickname: str) -> bool: @@ -55,11 +50,11 @@ class Admin: if record.uid == uid: # If the admin exist, update and do not go further record.nickname = new_admin_nickname - self.Logs.debug(f'UID ({record.uid}) has been updated with new nickname {new_admin_nickname}') + self._ctx.Logs.debug(f'UID ({record.uid}) has been updated with new nickname {new_admin_nickname}') return True - self.Logs.debug(f'The new nickname {new_admin_nickname} was not updated, uid = {uid} - The Client is not an admin') + self._ctx.Logs.debug(f'The new nickname {new_admin_nickname} was not updated, uid = {uid} - The Client is not an admin') return False def update_level(self, nickname: str, new_admin_level: int) -> bool: @@ -77,10 +72,10 @@ class Admin: if record.nickname == nickname: # If the admin exist, update and do not go further record.level = new_admin_level - self.Logs.debug(f'Admin ({record.nickname}) has been updated with new level {new_admin_level}') + self._ctx.Logs.debug(f'Admin ({record.nickname}) has been updated with new level {new_admin_level}') return True - self.Logs.debug(f'The new level {new_admin_level} was not updated, nickname = {nickname} - The Client is not an admin') + self._ctx.Logs.debug(f'The new level {new_admin_level} was not updated, nickname = {nickname} - The Client is not an admin') return False @@ -96,10 +91,10 @@ class Admin: admin_obj = self.get_admin(uidornickname) if admin_obj: self.UID_ADMIN_DB.remove(admin_obj) - self.Logs.debug(f'UID ({admin_obj.uid}) has been deleted') + self._ctx.Logs.debug(f'UID ({admin_obj.uid}) has been deleted') return True - self.Logs.debug(f'The UID {uidornickname} was not deleted') + self._ctx.Logs.debug(f'The UID {uidornickname} was not deleted') return False @@ -186,20 +181,20 @@ class Admin: if fp is None: return False - query = f"SELECT user, level, language FROM {self.Config.TABLE_ADMIN} WHERE fingerprint = :fp" + query = f"SELECT user, level, language FROM {self._ctx.Config.TABLE_ADMIN} WHERE fingerprint = :fp" data = {'fp': fp} - exe = await self.Base.db_execute_query(query, data) + exe = await self._ctx.Base.db_execute_query(query, data) result = exe.fetchone() if result: account = result[0] level = result[1] language = result[2] - user_obj = self.User.get_user(uidornickname) + user_obj = self._ctx.User.get_user(uidornickname) if user_obj: - admin_obj = self.Definition.MAdmin(**user_obj.to_dict(), account=account, level=level, language=language) + admin_obj = self._ctx.Definition.MAdmin(**user_obj.to_dict(), account=account, level=level, language=language) if self.insert(admin_obj): - self.Setting.current_admin = admin_obj - self.Logs.debug(f"[Fingerprint login] {user_obj.nickname} ({admin_obj.account}) has been logged in successfully!") + self._ctx.Settings.current_admin = admin_obj + self._ctx.Logs.debug(f"[Fingerprint login] {user_obj.nickname} ({admin_obj.account}) has been logged in successfully!") return True return False @@ -215,8 +210,8 @@ class Admin: """ mes_donnees = {'admin': admin_nickname} - query_search_user = f"SELECT id FROM {self.Config.TABLE_ADMIN} WHERE user = :admin" - r = await self.Base.db_execute_query(query_search_user, mes_donnees) + query_search_user = f"SELECT id FROM {self._ctx.Config.TABLE_ADMIN} WHERE user = :admin" + r = await self._ctx.Base.db_execute_query(query_search_user, mes_donnees) exist_user = r.fetchone() if exist_user: return True diff --git a/core/classes/modules/client.py b/core/classes/modules/client.py index ec422eb..75917a8 100644 --- a/core/classes/modules/client.py +++ b/core/classes/modules/client.py @@ -15,8 +15,7 @@ class Client: Args: loader (Loader): The Loader instance. """ - self.Logs = loader.Logs - self.Base = loader.Base + self._ctx = loader def insert(self, new_client: 'MClient') -> bool: """Insert a new User object @@ -28,7 +27,7 @@ class Client: bool: True if inserted """ - client_obj = self.get_Client(new_client.uid) + client_obj = self.get_client(new_client.uid) if not client_obj is None: # User already created return False @@ -48,7 +47,7 @@ class Client: Returns: bool: True if updated """ - user_obj = self.get_Client(uidornickname=uid) + user_obj = self.get_client(uidornickname=uid) if user_obj is None: return False @@ -68,7 +67,7 @@ class Client: bool: True if user mode has been updaed """ response = True - user_obj = self.get_Client(uidornickname=uidornickname) + user_obj = self.get_client(uidornickname=uidornickname) if user_obj is None: return False @@ -93,7 +92,7 @@ class Client: return False liste_umodes = list(umodes) - final_umodes_liste = [x for x in self.Base.Settings.PROTOCTL_USER_MODES if x in liste_umodes] + final_umodes_liste = [x for x in self._ctx.Base.Settings.PROTOCTL_USER_MODES if x in liste_umodes] final_umodes = ''.join(final_umodes_liste) user_obj.umodes = f"+{final_umodes}" @@ -110,7 +109,7 @@ class Client: bool: True if deleted """ - user_obj = self.get_Client(uidornickname=uid) + user_obj = self.get_client(uidornickname=uid) if user_obj is None: return False @@ -119,7 +118,7 @@ class Client: return True - def get_Client(self, uidornickname: str) -> Optional['MClient']: + def get_client(self, uidornickname: str) -> Optional['MClient']: """Get The Client Object model Args: @@ -146,7 +145,7 @@ class Client: str|None: Return the UID """ - client_obj = self.get_Client(uidornickname=uidornickname) + client_obj = self.get_client(uidornickname=uidornickname) if client_obj is None: return None @@ -162,29 +161,13 @@ class Client: Returns: str|None: the nickname """ - client_obj = self.get_Client(uidornickname=uidornickname) + client_obj = self.get_client(uidornickname=uidornickname) if client_obj is None: return None return client_obj.nickname - def get_client_asdict(self, uidornickname: str) -> Optional[dict[str, Any]]: - """Transform User Object to a dictionary - - Args: - uidornickname (str): The UID or The nickname - - Returns: - Union[dict[str, any], None]: User Object as a dictionary or None - """ - client_obj = self.get_Client(uidornickname=uidornickname) - - if client_obj is None: - return None - - return client_obj.to_dict() - def is_exist(self, uidornickname: str) -> bool: """Check if the UID or the nickname exist in the USER DB @@ -194,7 +177,7 @@ class Client: Returns: bool: True if exist """ - user_obj = self.get_Client(uidornickname=uidornickname) + user_obj = self.get_client(uidornickname=uidornickname) if user_obj is None: return False @@ -211,15 +194,15 @@ class Client: bool: True if exist """ - table_client = self.Base.Config.TABLE_CLIENT + table_client = self._ctx.Base.Config.TABLE_CLIENT account_to_check = {'account': account.lower()} - account_to_check_query = await self.Base.db_execute_query(f""" + account_to_check_query = await self._ctx.Base.db_execute_query(f""" SELECT id FROM {table_client} WHERE LOWER(account) = :account """, account_to_check) account_to_check_result = account_to_check_query.fetchone() if account_to_check_result: - self.Logs.error(f"Account ({account}) already exist") + self._ctx.Logs.error(f"Account ({account}) already exist") return True return False diff --git a/core/classes/modules/commands.py b/core/classes/modules/commands.py index 412756a..2129061 100644 --- a/core/classes/modules/commands.py +++ b/core/classes/modules/commands.py @@ -13,11 +13,21 @@ class Command: Args: loader (Loader): The Loader instance. """ - self.Loader = loader - self.Base = loader.Base - self.Logs = loader.Logs + self._ctx = loader - def build(self, new_command_obj: MCommand) -> bool: + def build_command(self, level: int, module_name: str, command_name: str, command_description: str) -> bool: + """This method build the commands variable + + Args: + level (int): The Level of the command + module_name (str): The module name + command_name (str): The command name + command_description (str): The description of the command + """ + # Build Model. + return self._build(self._ctx.Definition.MCommand(module_name, command_name, command_description, level)) + + def _build(self, new_command_obj: MCommand) -> bool: command = self.get_command(new_command_obj.command_name, new_command_obj.module_name) if command is None: @@ -68,7 +78,7 @@ class Command: for c in tmp_model: self.DB_COMMANDS.remove(c) - self.Logs.debug(f"[COMMAND] Drop command for module {module_name}") + self._ctx.Logs.debug(f"[COMMAND] Drop command for module {module_name}") return True def get_ordered_commands(self) -> list[MCommand]: @@ -86,7 +96,7 @@ class Command: return new_list def is_client_allowed_to_run_command(self, nickname: str, command_name: str) -> bool: - admin = self.Loader.Admin.get_admin(nickname) + admin = self._ctx.Admin.get_admin(nickname) admin_level = admin.level if admin else 0 commands = self.get_commands_by_level(admin_level) diff --git a/core/classes/modules/reputation.py b/core/classes/modules/reputation.py index 242290a..84727ee 100644 --- a/core/classes/modules/reputation.py +++ b/core/classes/modules/reputation.py @@ -14,9 +14,7 @@ class Reputation: Args: loader (Loader): The Loader instance. """ - - self.Logs = loader.Logs - self.MReputation: Optional[MReputation] = None + self._ctx = loader def insert(self, new_reputation_user: MReputation) -> bool: """Insert a new Reputation User object @@ -34,16 +32,16 @@ class Reputation: if record.uid == new_reputation_user.uid: # If the user exist then return False and do not go further exist = True - self.Logs.debug(f'{record.uid} already exist') + self._ctx.Logs.debug(f'{record.uid} already exist') return result if not exist: self.UID_REPUTATION_DB.append(new_reputation_user) result = True - self.Logs.debug(f'New Reputation User Captured: ({new_reputation_user})') + self._ctx.Logs.debug(f'New Reputation User Captured: ({new_reputation_user})') if not result: - self.Logs.critical(f'The Reputation User Object was not inserted {new_reputation_user}') + self._ctx.Logs.critical(f'The Reputation User Object was not inserted {new_reputation_user}') return result @@ -86,11 +84,11 @@ class Reputation: # If the user exist then remove and return True and do not go further self.UID_REPUTATION_DB.remove(record) result = True - self.Logs.debug(f'UID ({record.uid}) has been deleted') + self._ctx.Logs.debug(f'UID ({record.uid}) has been deleted') return result if not result: - self.Logs.critical(f'The UID {uid} was not deleted') + self._ctx.Logs.critical(f'The UID {uid} was not deleted') return result diff --git a/core/classes/modules/user.py b/core/classes/modules/user.py index aff0963..154ec75 100644 --- a/core/classes/modules/user.py +++ b/core/classes/modules/user.py @@ -11,14 +11,16 @@ class User: UID_DB: list['MUser'] = [] @property - def get_current_user(self) -> 'MUser': - return self.current_user + def current_user(self) -> 'MUser': + return self._current_user + + @current_user.setter + def current_user(self, muser: 'MUser') -> None: + self._current_user = muser def __init__(self, loader: 'Loader'): - - self.Logs = loader.Logs - self.Base = loader.Base - self.current_user: Optional['MUser'] = None + self._ctx = loader + self._current_user: Optional['MUser'] = None def insert(self, new_user: 'MUser') -> bool: """Insert a new User object @@ -55,7 +57,7 @@ class User: return False user_obj.nickname = new_nickname - self.Logs.debug(f"UID ({uid}) has benn update with new nickname ({new_nickname}).") + self._ctx.Logs.debug(f"UID ({uid}) has benn update with new nickname ({new_nickname}).") return True def update_mode(self, uidornickname: str, modes: str) -> bool: @@ -94,7 +96,7 @@ class User: return False liste_umodes = list(umodes) - final_umodes_liste = [x for x in self.Base.Settings.PROTOCTL_USER_MODES if x in liste_umodes] + final_umodes_liste = [x for x in self._ctx.Base.Settings.PROTOCTL_USER_MODES if x in liste_umodes] final_umodes = ''.join(final_umodes_liste) user_obj.umodes = f"+{final_umodes}" diff --git a/core/definition.py b/core/definition.py index 6d9ef95..3679c8d 100644 --- a/core/definition.py +++ b/core/definition.py @@ -1,3 +1,7 @@ +import asyncio +import concurrent +import concurrent.futures +import threading from datetime import datetime from json import dumps from dataclasses import dataclass, field, asdict, fields, replace @@ -210,6 +214,12 @@ class MConfig(MainModel): PASSWORD: str = "password" """The password of the admin of the service""" + RPC_HOST: str = "127.0.0.1" + """The host to bind. Default: 127.0.0.1""" + + RPC_PORT: int = 5000 + """The port of the defender json rpc. Default: 5000""" + RPC_USERS: list[dict] = field(default_factory=list) """The Defender rpc users""" @@ -344,6 +354,15 @@ class MConfig(MainModel): self.SERVEUR_CHARSET: list = ["utf-8", "iso-8859-1"] """0: utf-8 | 1: iso-8859-1""" +@dataclass +class MThread(MainModel): + name: str + thread_id: Optional[int] + thread_event: Optional[threading.Event] + thread_obj: threading.Thread + executor: concurrent.futures.ThreadPoolExecutor + future: asyncio.Future + @dataclass class MCommand(MainModel): module_name: str = None