sh-edraft.de/sh_discord_bot
5
1
Fork 1
Code Issues 26 Pull Requests Actions Packages Projects 3 Releases 27 Wiki Activity

Formatted files with black

Browse Source
This commit is contained in:
Sven Heidemann
2023-01-13 23:35:42 +01:00
parent 37ec0cf0c7
commit e2b2fb5abb
252 changed files with 6085 additions and 3791 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
kdb-bot/src/modules/technician/__init__.py
View File

@@ -11,16 +11,16 @@ Discord bot for the Keksdose discord Server
"""
__title__ = 'modules.technician'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.3.0'
__title__ = "modules.technician"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 sh-edraft.de"
__version__ = "0.3.0"
from collections import namedtuple
# imports:
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='3', micro='0')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="0", minor="3", micro="0")

14
kdb-bot/src/modules/technician/command/__init__.py
View File

@@ -11,16 +11,16 @@ Discord bot for the Keksdose discord Server
"""
__title__ = 'modules.technician.command'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.3.0'
__title__ = "modules.technician.command"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 sh-edraft.de"
__version__ = "0.3.0"
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='3', micro='0')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="0", minor="3", micro="0")

56
kdb-bot/src/modules/technician/command/log_command.py
View File

@@ -22,18 +22,17 @@ from modules.permission.abc.permission_service_abc import PermissionServiceABC
class LogCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
logging_settings: LoggingSettings,
services: ServiceProviderABC,
message_service: MessageServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
time_format: TimeFormatSettings,
env: ApplicationEnvironmentABC
self,
logger: CommandLogger,
logging_settings: LoggingSettings,
services: ServiceProviderABC,
message_service: MessageServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
time_format: TimeFormatSettings,
env: ApplicationEnvironmentABC,
):
DiscordCommandABC.__init__(self)
@@ -49,10 +48,10 @@ class LogCommand(DiscordCommandABC):
self._log_settings: LoggingSettings = logging_settings
self._time_format_settings: TimeFormatSettings = time_format
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
def _reduce_path(self, p: str) -> str:
if len(p.split('/')) == 1 or p == '':
if len(p.split("/")) == 1 or p == "":
return p
return self._reduce_path(os.path.dirname(p))
@@ -62,15 +61,15 @@ class LogCommand(DiscordCommandABC):
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def log(self, ctx: Context):
self._logger.debug(__name__, f'Received command log {ctx}')
self._logger.debug(__name__, f"Received command log {ctx}")
possible_log_paths = List(str)
possible_log_paths.append(self._reduce_path(self._logging_settings.path))
file_extensions = List(str)
if '.' in self._logging_settings.filename:
if "." in self._logging_settings.filename:
split_filename = self._logging_settings.filename.split(".")
file_extensions.append(f'.{split_filename[len(split_filename) - 1]}')
file_extensions.append(f".{split_filename[len(split_filename) - 1]}")
for subclass in CustomFileLoggerABC.__subclasses__():
logger: CustomFileLoggerABC = self._services.get_service(subclass)
@@ -78,9 +77,9 @@ class LogCommand(DiscordCommandABC):
continue
path = self._reduce_path(logger.settings.path)
if '.' in logger.settings.filename:
if "." in logger.settings.filename:
split_filename = logger.settings.filename.split(".")
file_extension = f'.{split_filename[len(split_filename) - 1]}'
file_extension = f".{split_filename[len(split_filename) - 1]}"
if file_extension not in file_extensions:
file_extensions.append(file_extension)
@@ -89,28 +88,35 @@ class LogCommand(DiscordCommandABC):
possible_log_paths.append(path)
files_str = "\n\t".join(possible_log_paths.to_list())
self._logger.debug(__name__, f'Possible log files: \n\t{files_str}')
self._logger.debug(__name__, f"Possible log files: \n\t{files_str}")
files = List(str)
for possible_path in possible_log_paths:
for r, d, f in os.walk(possible_path):
for file in f:
if '.' not in file:
if "." not in file:
continue
split_filename = file.split(".")
if f'.{split_filename[len(split_filename) - 1]}' not in file_extensions:
if (
f".{split_filename[len(split_filename) - 1]}"
not in file_extensions
):
continue
files.append(os.path.join(r, file))
files_str = "\n\t".join(files.to_list())
self._logger.debug(__name__, f'Log files: \n\t{files_str}')
self._logger.debug(__name__, f"Log files: \n\t{files_str}")
zip_file = ZipFile('logs.zip', 'w')
zip_file = ZipFile("logs.zip", "w")
files.for_each(lambda x: zip_file.write(x))
zip_file.close()
await self._message_service.send_interaction_msg(ctx.interaction, self._t.transform('modules.technician.log_message'), file=discord.File(zip_file.filename, 'logs.zip'))
await self._message_service.send_interaction_msg(
ctx.interaction,
self._t.transform("modules.technician.log_message"),
file=discord.File(zip_file.filename, "logs.zip"),
)
os.remove(zip_file.filename)
self._logger.trace(__name__, f'Finished log command')
self._logger.trace(__name__, f"Finished log command")

33
kdb-bot/src/modules/technician/command/restart_command.py
View File

@@ -16,17 +16,16 @@ from modules.permission.abc.permission_service_abc import PermissionServiceABC
class RestartCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings,
):
DiscordCommandABC.__init__(self)
@@ -39,19 +38,21 @@ class RestartCommand(DiscordCommandABC):
self._permissions = permissions
self._settings = settings
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
@commands.hybrid_command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def restart(self, ctx: Context):
self._logger.debug(__name__, f'Received command restart {ctx}')
self._logger.debug(__name__, f"Received command restart {ctx}")
self._config.add_configuration('IS_RESTART', 'true')
await self._client_utils.presence_game('common.presence.restart')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.technician.restart_message'))
self._config.add_configuration("IS_RESTART", "true")
await self._client_utils.presence_game("common.presence.restart")
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.technician.restart_message")
)
await asyncio.sleep(self._settings.wait_for_restart)
await self._bot.stop_async()
self._logger.trace(__name__, f'Finished restart command')
self._logger.trace(__name__, f"Finished restart command")

31
kdb-bot/src/modules/technician/command/shutdown_command.py
View File

@@ -17,17 +17,16 @@ from modules.permission.abc.permission_service_abc import PermissionServiceABC
class ShutdownCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings,
):
DiscordCommandABC.__init__(self)
@@ -40,18 +39,20 @@ class ShutdownCommand(DiscordCommandABC):
self._permissions = permissions
self._settings = settings
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
@commands.hybrid_command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def shutdown(self, ctx: Context):
self._logger.debug(__name__, f'Received command shutdown {ctx}')
self._logger.debug(__name__, f"Received command shutdown {ctx}")
await self._client_utils.presence_game('common.presence.shutdown')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.technician.shutdown_message'))
await self._client_utils.presence_game("common.presence.shutdown")
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.technician.shutdown_message")
)
await asyncio.sleep(self._settings.wait_for_shutdown)
await self._bot.stop_async()
self._logger.trace(__name__, f'Finished shutdown command')
self._logger.trace(__name__, f"Finished shutdown command")

9
kdb-bot/src/modules/technician/technician_module.py
View File

@@ -13,14 +13,17 @@ from modules.base.service.base_helper_service import BaseHelperService
class TechnicianModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.base_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
def configure_configuration(
self, config: ConfigurationABC, env: ApplicationEnvironmentABC
):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
def configure_services(
self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC
):
services.add_transient(BaseHelperABC, BaseHelperService)
# commands
self._dc.add_command(RestartCommand)