Code refactoring

This commit is contained in:
adator
2025-11-23 18:20:56 +01:00
parent 4c93f85008
commit eb7c6ef8d0
3 changed files with 21 additions and 40 deletions

View File

@@ -93,4 +93,4 @@ class Translation:
except Exception as err: except Exception as err:
self.Logs.error(f'General Error: {err}') self.Logs.error(f'General Error: {err}')
return {} return dict()

View File

@@ -3,20 +3,20 @@ Main utils library.
""" """
import gc import gc
import ssl import ssl
import socket
import sys
from pathlib import Path from pathlib import Path
from re import match, sub from re import match, sub
import threading
from typing import Literal, Optional, Any, TYPE_CHECKING from typing import Literal, Optional, Any, TYPE_CHECKING
from datetime import datetime from datetime import datetime
from time import time from time import time, sleep
from random import choice from random import choice
from hashlib import md5, sha3_512 from hashlib import md5, sha3_512
from core.classes.modules.settings import global_settings from core.classes.modules.settings import global_settings
from asyncio import iscoroutinefunction from asyncio import iscoroutinefunction
if TYPE_CHECKING: if TYPE_CHECKING:
from core.irc import Irc from threading import Event
from core.loader import Loader
def tr(message: str, *args) -> str: def tr(message: str, *args) -> str:
"""Translation Engine system """Translation Engine system
@@ -115,39 +115,6 @@ def get_ssl_context() -> ssl.SSLContext:
ctx.verify_mode = ssl.CERT_NONE ctx.verify_mode = ssl.CERT_NONE
return ctx return ctx
def create_socket(uplink: 'Irc') -> None:
"""Create a socket to connect SSL or Normal connection
"""
try:
soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM or socket.SOCK_NONBLOCK)
connexion_information = (uplink.Config.SERVEUR_IP, uplink.Config.SERVEUR_PORT)
if uplink.Config.SERVEUR_SSL:
# Create SSL Context object
ssl_context = get_ssl_context()
ssl_connexion = ssl_context.wrap_socket(soc, server_hostname=uplink.Config.SERVEUR_HOSTNAME)
ssl_connexion.connect(connexion_information)
uplink.IrcSocket = ssl_connexion
uplink.Config.SSL_VERSION = uplink.IrcSocket.version()
uplink.Logs.info(f"-- Connected using SSL : Version = {uplink.Config.SSL_VERSION}")
else:
soc.connect(connexion_information)
uplink.IrcSocket = soc
uplink.Logs.info("-- Connected in a normal mode!")
return None
except (ssl.SSLEOFError, ssl.SSLError) as soe:
uplink.Logs.critical(f"[SSL ERROR]: {soe}")
except OSError as oe:
uplink.Logs.critical(f"[OS Error]: {oe}")
if 'connection refused' in str(oe).lower():
sys.exit(oe.__str__())
if oe.errno == 10053:
sys.exit(oe.__str__())
except AttributeError as ae:
uplink.Logs.critical(f"AttributeError: {ae}")
def run_python_garbage_collector() -> int: def run_python_garbage_collector() -> int:
"""Run Python garbage collector """Run Python garbage collector
@@ -167,6 +134,21 @@ def get_number_gc_objects(your_object_to_count: Optional[Any] = None) -> int:
return sum(1 for obj in gc.get_objects() if isinstance(obj, your_object_to_count)) return sum(1 for obj in gc.get_objects() if isinstance(obj, your_object_to_count))
def heartbeat(event: 'Event', loader: 'Loader', beat: float) -> None:
"""Execute certaines commandes de nettoyage toutes les x secondes
x étant définit a l'initialisation de cette class (self.beat)
Args:
beat (float): Nombre de secondes entre chaque exécution
"""
while event.is_set():
loader.Base.execute_periodic_action()
sleep(beat)
loader.Logs.debug("Heartbeat is off!")
return None
def generate_random_string(lenght: int) -> str: def generate_random_string(lenght: int) -> str:
"""Retourn une chaîne aléatoire en fonction de la longueur spécifiée. """Retourn une chaîne aléatoire en fonction de la longueur spécifiée.

View File

@@ -1,6 +1,5 @@
import asyncio import asyncio
from core import install from core import install
############################################# #############################################
# @Version : 6.4 # # @Version : 6.4 #
# Requierements : # # Requierements : #
@@ -19,4 +18,4 @@ async def main():
await loader.Irc.run() await loader.Irc.run()
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(main()) asyncio.run(main(), debug=False)